fix(archives): a print that finishes while Bambuddy restarts is completed (issue #3310)

- After a restart the printer's FINISH is the first status of a new
  MQTT connection, so the completion never fired and the reconnect
  reconcile closed the archive as aborted/noStatusUpdate
- The reconcile now closes it as completed when the printer reports
  FINISH for the archive's own subtask_id, and as failed for FAILED
  (the printer's HMS reason, else noStatusUpdate); IDLE, another job,
  "0" or no id stay aborted
- A reconciled completion ends at its start plus the slicer estimate
  of the plate that printed (from its queue job), capped at the
  reconnect; with no plate only a single-plate file's estimate is used
- New print_log_entries.duration_unmeasured marks runs closed at
  reconnect; time accuracy (archive card, Statistics) skips them,
  totals still count the time
- Progress and layer are passed for completed and failed runs
- HMSError.completion_payload() shared with the MQTT completion
- Docstrings that said a later completion would correct the status

CHANGELOG.md

@@ -176,6 +176,7 @@ 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
+- **A print that finishes while Bambuddy restarts is archived as completed (#3310, reported by @Thomansky)** — When a print ended while Bambuddy was down (a container update, a host reboot), the printer's FINISH arrived as the first status of a new connection, so it never counted as the print ending. The reconnect then closed the archive as aborted with "No status update", logged it with no print time, cancelled its queue job and sent "print stopped". Now, when the printer still reports the archive's own job (the same job ID) as finished, the archive is closed as completed, with the library print count, batch run, finish notification and smart-plug auto-off that go with it. Its end time is the start plus the slicer's estimate for the plate that printed, never later than the reconnect, since the printer doesn't say when it finished. That time counts in the print-time totals but not in time accuracy, on the archive card or in Statistics. A failed job is closed as failed, with the printer's error as the reason when it has one. When the printer is idle, shows another job or reports no job ID, the archive is closed as aborted, as before.
 - **Home Assistant's connection test names a certificate problem (#3300)** — After Home Assistant moved to https, **Test Connection** said "Could not connect to Home Assistant" when Home Assistant was up and answering, because a certificate Bambuddy does not trust fails like an address nothing answers on. It now says the certificate is not trusted, why (self-signed or from your own authority, issued for a hostname while the URL uses the IP address, or expired) and what to do, and says when https was used on a port that serves plain http. The Smart Plugs page of the wiki explains how to add your own certificate authority, also on a native install.
 - **A bug report is no longer sent when debug logging stops (#3363, reported by @choumarin)** — "Stop & Submit" turned debug logging off and filed the report as a public GitHub issue in the same click, and the 5-minute auto-stop did the same with nobody watching, so a report meant only to capture a log was sent before it was ready. The button is now "Stop Logging": it collects the logs and ends on a review that says a public issue will be created and how many log lines were collected, with Submit and Discard. Closing the panel keeps the review, and a failed send goes back to it with the logs kept.
 - **A camera turned sideways fits the camera window (#3362, reported by @BurgerKerman)** — With a camera's rotation set to 90 or 270 degrees, the camera window that floats over the page laid the picture out as if it were still landscape and then turned it, so it was cut off in a wide window and surrounded by black bars in a tall one. It now fits whatever size the window is resized to, as the camera page opened in its own browser window already did. The same happened to Cam Wall tiles, where a portrait camera was cut off at the top and bottom, and to the classic stream overlay without a fixed layout; both are fixed too. The camera page in its own window sized the picture to the screen rather than the space below its header, which cut a little off the top and bottom in a wide window; it fits the space now.

backend/app/api/routes/archives.py

@@ -283,6 +283,11 @@ def compute_time_accuracy(archive: PrintArchive, run_aggregate: dict | None = No
     # Both fields are cleared so the card shows estimate + no badge.
     if run_aggregate and (run_aggregate.get("run_count") or 0) > 1:
         return result
+    # A run whose times were not measured (#3310): its end is the slicer's
+    # estimate or the reconnect. The card shows the estimate and no badge,
+    # as for multi-run archives.
+    if run_aggregate and run_aggregate.get("unmeasured_run_count"):
+        return result
 
     if archive.started_at and archive.completed_at and archive.status == "completed":
         actual_seconds = int((archive.completed_at - archive.started_at).total_seconds())
@@ -304,7 +309,7 @@ async def _load_run_aggregates(db: AsyncSession, archive_ids: list[int]) -> dict
     """Batch-load per-archive run aggregates from PrintLogEntry.
 
     Returns ``{archive_id: {run_count, last_run_at, total_filament_actual_grams,
-    successful_run_count, failed_run_count}}``. Archives with no logged runs are
+    successful_run_count, failed_run_count, unmeasured_run_count}}``. Archives with no logged runs are
     absent from the map; callers should treat that as zero/none.
     """
     from backend.app.models.print_log import PrintLogEntry
@@ -319,18 +324,20 @@ async def _load_run_aggregates(db: AsyncSession, archive_ids: list[int]) -> dict
             func.coalesce(func.sum(PrintLogEntry.filament_used_grams), 0).label("total_filament"),
             func.sum(case((PrintLogEntry.status == "completed", 1), else_=0)).label("successful"),
             func.sum(case((PrintLogEntry.status == "failed", 1), else_=0)).label("failed"),
+            func.sum(case((PrintLogEntry.duration_unmeasured.is_(True), 1), else_=0)).label("estimated"),
         )
         .where(PrintLogEntry.archive_id.in_(archive_ids))
         .group_by(PrintLogEntry.archive_id)
     )
     aggregates: dict[int, dict] = {}
-    for archive_id, run_count, last_run_at, total_filament, successful, failed in rows.all():
+    for archive_id, run_count, last_run_at, total_filament, successful, failed, estimated in rows.all():
         aggregates[archive_id] = {
             "run_count": int(run_count or 0),
             "last_run_at": last_run_at,
             "total_filament_actual_grams": float(total_filament) if total_filament else None,
             "successful_run_count": int(successful or 0),
             "failed_run_count": int(failed or 0),
+            "unmeasured_run_count": int(estimated or 0),
         }
     return aggregates
 
@@ -1423,6 +1430,8 @@ async def get_archive_stats(
         .where(
             PrintLogEntry.status == "completed",
             PrintArchive.print_time_seconds.isnot(None),
+            # A run whose times were not measured is no sample (#3310).
+            PrintLogEntry.duration_unmeasured.is_not(True),
             *base_conditions,
         )
     )

backend/app/core/database.py

@@ -5868,6 +5868,11 @@ async def _run_migrations(conn):
     # default -- identical DDL on SQLite and Postgres.
     await _safe_execute(conn, "ALTER TABLE active_print_sessions ADD COLUMN spool_change_log JSON")
 
+    # Migration: a print log run whose times were not measured, a print that
+    # finished while Bambuddy restarted (#3310). Nullable, no
+    # default -- identical DDL on SQLite and Postgres.
+    await _safe_execute(conn, "ALTER TABLE print_log_entries ADD COLUMN duration_unmeasured BOOLEAN")
+
     await _migrate_auth_datetimes_to_timestamp(conn)
     await _migrate_add_missing_set_null_fks(conn)
     await _migrate_wallet_transactions_keep_deleted_users(conn)

backend/app/main.py

@@ -7154,11 +7154,11 @@ def _is_active_archive_stale(archive, state) -> tuple[bool, str]:
 
     Conservative on purpose: PAUSE / PREPARE / SLICING and any RUNNING state
     with matching subtask_id+subtask_name is left alone. The cost of a false
-    positive is a duplicate archive on the next real PRINT COMPLETE — the
-    reactive handler uses ``_active_prints`` for lookup, which the reconcile
-    clears on synthesis, so the real completion creates a fresh row instead
-    of overwriting the synthesised one (#1679). The cost of a false negative
-    is the ghost-print loop in #1542.
+    positive is a print closed while it still runs: the reconcile takes its
+    ``_active_prints`` entry and the archive is no longer ``printing``, so
+    the real PRINT COMPLETE later finds no archive to close (#1679). The
+    cost of a false negative is the ghost-print loop in #1542. What the
+    archive is closed as is ``_reconciled_outcome``'s call.
 
     Pre-push guard (#1679): when ``state.state`` is empty or ``"unknown"``,
     MQTT has connected but the first ``push_status`` response hasn't been
@@ -7281,6 +7281,71 @@ def _sd_files_in_use(state) -> set[str]:
     return in_use
 
 
+def _reconciled_outcome(archive, state) -> str:
+    """What a stale archive is closed as at reconnect (#3310).
+
+    A print that ends while Bambuddy restarts is never seen to end: the new
+    MQTT client's first status is already FINISH, which is not a transition.
+    The printer keeps reporting the finished job's ``subtask_id`` until the
+    next print, so a matching id proves the FINISH or FAILED is this
+    archive's. Anything else -- IDLE, another job, an id of "0" (a print
+    started outside the cloud) or none -- says only that the print is over,
+    not how, and stays ``aborted``.
+    """
+    current_state = (state.state or "").upper()
+    if current_state not in ("FINISH", "FAILED"):
+        return "aborted"
+    archive_id = (archive.subtask_id or "").strip()
+    current_id = str(state.subtask_id or "").strip()
+    if archive_id in ("", "0") or archive_id != current_id:
+        return "aborted"
+    return "completed" if current_state == "FINISH" else "failed"
+
+
+def _reconciled_end_time(archive, now: datetime, plate_id: int | None = None) -> datetime | None:
+    """When a print reconciled as completed most likely ended, or None (#3310).
+
+    The printer does not say when it finished, and the reconnect can be hours
+    later. Its start plus the slicer's estimate for the plate it printed,
+    never later than now, is the best guess; None when either is missing,
+    and the run then keeps no duration, as an aborted one does.
+
+    ``plate_id`` is the plate this run printed, from its queue job: a reprint
+    of another plate reuses the archive without changing ``archive.plate_id``.
+    """
+    started = archive.started_at
+    if not started:
+        return None
+    if started.tzinfo is None:
+        started = started.replace(tzinfo=timezone.utc)
+    plate = plate_id if plate_id is not None else archive.plate_id
+    estimate = None
+    if archive.file_path:
+        from backend.app.utils.threemf_tools import extract_print_time_from_3mf
+
+        full_path = (
+            app_settings.base_dir / archive.file_path
+        )  # SEC-PATH-OK: archive.file_path is DB-stored, internally generated
+        try:
+            estimate = extract_print_time_from_3mf(full_path, plate)
+        except Exception as exc:  # noqa: BLE001
+            logging.getLogger(__name__).debug(
+                "Plate estimate for reconciled archive %s unreadable: %s", archive.id, exc
+            )
+        if plate is None and estimate != archive.print_time_seconds:
+            # Without a plate the first plate's estimate is read. The archive
+            # carries the sum over all plates (#1593), so a difference means
+            # several plates and no telling which one printed.
+            estimate = None
+    elif plate is None:
+        # Nothing to read the plate from; the archive's own estimate, as for
+        # a single-plate file.
+        estimate = archive.print_time_seconds
+    if not estimate or estimate <= 0:
+        return None
+    return min(started + timedelta(seconds=int(estimate)), now)
+
+
 async def reconcile_stale_active_prints(printer_id: int) -> int:
     """Synthesise ``on_print_complete`` for archives whose print can't be
     running on the printer anymore.
@@ -7297,10 +7362,11 @@ async def reconcile_stale_active_prints(printer_id: int) -> int:
     cleanup chain handles SD-file deletion, status updates, usage tracking,
     and notifications.
 
-    Synthesised ``status="aborted"`` is the conservative label: we have no
-    proof the print finished successfully (and no progress evidence to
-    promote to ``"completed"``). The real PRINT COMPLETE callback, if it
-    fires later, overwrites the status with the correct value.
+    The status is ``_reconciled_outcome``'s: ``completed`` or ``failed``
+    when the printer still reports the archive's own job as finished or
+    failed, ``aborted`` otherwise. Nothing corrects it later: the archive is
+    no longer ``printing``, so a later PRINT COMPLETE finds no archive, and
+    after a restart the printer's FINISH is not one (#3310).
 
     Returns the number of archives reconciled.
     """
@@ -7327,6 +7393,19 @@ async def reconcile_stale_active_prints(printer_id: int) -> int:
     if not active:
         return 0
 
+    # The plate each run printed, from its queue job (#3310).
+    from backend.app.models.print_queue import PrintQueueItem
+
+    async with async_session() as db:
+        plate_rows = await db.execute(
+            select(PrintQueueItem.archive_id, PrintQueueItem.plate_id).where(
+                PrintQueueItem.printer_id == printer_id,
+                PrintQueueItem.status == "printing",
+                PrintQueueItem.archive_id.in_([a.id for a in active]),
+            )
+        )
+        run_plates = {archive_id: plate for archive_id, plate in plate_rows.all() if plate is not None}
+
     logger = logging.getLogger(__name__)
     for archive in active:
         is_stale, reason = _is_active_archive_stale(archive, state)
@@ -7345,18 +7424,30 @@ async def reconcile_stale_active_prints(printer_id: int) -> int:
         # metrics / debug logging). raw_data is the live printer state so
         # the usage tracker can compare end-of-print remain% against the
         # captured start values.
+        outcome = _reconciled_outcome(archive, state)
+        payload = {
+            "status": outcome,
+            "filename": archive.filename,
+            "subtask_name": archive.print_name or "",
+            "subtask_id": archive.subtask_id or "",
+            "raw_data": state.raw_data or {},
+            "_reconciled": True,
+        }
+        if outcome != "aborted":
+            # The printer's progress and layer are this job's, so a partial
+            # run is charged what it printed rather than nothing.
+            if state.progress:
+                payload["last_progress"] = state.progress
+            if state.layer_num:
+                payload["last_layer_num"] = state.layer_num
+        if outcome == "failed":
+            payload["hms_errors"] = [e.completion_payload() for e in state.hms_errors or []]
+        if outcome == "completed":
+            ended = _reconciled_end_time(archive, datetime.now(timezone.utc), run_plates.get(archive.id))
+            if ended:
+                payload["_reconciled_completed_at"] = ended
         try:
-            await on_print_complete(
-                printer_id,
-                {
-                    "status": "aborted",
-                    "filename": archive.filename,
-                    "subtask_name": archive.print_name or "",
-                    "subtask_id": archive.subtask_id or "",
-                    "raw_data": state.raw_data or {},
-                    "_reconciled": True,
-                },
-            )
+            await on_print_complete(printer_id, payload)
             reconciled += 1
         except Exception as e:
             # Catch-all: a reconciliation failure must not block the
@@ -8863,12 +8954,13 @@ async def _close_out_print(printer_id: int, data: dict):
             if hms_errors:
                 logger.info("[ARCHIVE] HMS errors at failure: %s", hms_errors)
             failure_reason = derive_failure_reason(status, hms_errors)
-            if data.get("_reconciled"):
+            if data.get("_reconciled") and (status == "aborted" or (status == "failed" and not failure_reason)):
                 # A reconciled completion closes out a stale archive at
                 # reconnect — it is not a user action, so don't mislabel it
                 # "userCancelled". It shares the stale-cleanup path's key
                 # (issue #2974) and records that the real end time is unknown,
-                # which is also why its logged duration is 0 (#2592).
+                # which is also why its logged duration is 0 (#2592). A failed
+                # one the printer's errors explain keeps their reason (#3310).
                 failure_reason = "noStatusUpdate"
             elif data.get("_kill_switch"):
                 # Not the user's cancel: the kill switch stopped a print that did
@@ -8883,7 +8975,11 @@ async def _close_out_print(printer_id: int, data: dict):
                 archive_id,
                 status=status,
                 completed_at=(
-                    datetime.now(timezone.utc) if status in ("completed", "failed", "aborted", "cancelled") else None
+                    # A reconciled completed print ended at an estimated time,
+                    # not at the reconnect (#3310).
+                    data.get("_reconciled_completed_at") or datetime.now(timezone.utc)
+                    if status in ("completed", "failed", "aborted", "cancelled")
+                    else None
                 ),
                 failure_reason=failure_reason,
             )
@@ -9095,8 +9191,11 @@ async def _close_out_print(printer_id: int, data: dict):
                     created_by_id=archive.created_by_id,
                     created_by_username=_print_user_info.get("username") if _print_user_info else None,
                     # Reconciled completions have an unknown real end time —
-                    # log 0 duration instead of the whole disconnect gap (#2592).
-                    reconciled=bool(data.get("_reconciled")),
+                    # log 0 duration instead of the whole disconnect gap (#2592),
+                    # unless it was estimated (#3310).
+                    reconciled=bool(data.get("_reconciled")) and not data.get("_reconciled_completed_at"),
+                    # Its time is the estimate or nothing, never measured (#3310).
+                    duration_unmeasured=bool(data.get("_reconciled")),
                 )
                 await record_archive_wear(db, archive, run_entry)
                 await db.commit()
@@ -9523,7 +9622,12 @@ async def _close_out_print(printer_id: int, data: dict):
                         # print ended while the printer was offline and
                         # completed_at is the reconnect, so the gap would read
                         # as the duration -- 14 h for a night offline (DB-C9).
-                        if archive.started_at and archive.completed_at and not data.get("_reconciled"):
+                        # An estimated end (#3310) gives the estimate.
+                        if (
+                            archive.started_at
+                            and archive.completed_at
+                            and (not data.get("_reconciled") or data.get("_reconciled_completed_at"))
+                        ):
                             elapsed = (archive.completed_at - archive.started_at).total_seconds()
                             if elapsed > 0:
                                 actual_time_seconds = int(elapsed)

backend/app/models/print_log.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, func
+from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from backend.app.core.database import Base
@@ -46,6 +46,11 @@ class PrintLogEntry(Base):
     energy_cost: Mapped[float | None] = mapped_column(Float)
     # Printer wear for this run: duration at the printer's hourly wear cost (#694).
     wear_cost: Mapped[float | None] = mapped_column(Float)
+    # True when the run's times were not measured: a print closed at reconnect
+    # (#3310), whose end is its start plus the slicer's estimate when it
+    # completed, or the reconnect (duration 0). Counted in
+    # the totals, left out of the time accuracy. NULL for a measured run.
+    duration_unmeasured: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
     failure_reason: Mapped[str | None] = mapped_column(String(100))
     # User quality verdict ('good' / 'reject'), mirrored from the archive on
     # PATCH exactly like status/failure_reason (#1444 mirror) so verdict-aware

backend/app/services/bambu_mqtt.py

@@ -819,6 +819,22 @@ class HMSError:
     # (#1830) and on `hms[]`-sourced faults generally.
     full_code: str = ""
 
+    def completion_payload(self) -> dict:
+        """This error as a print-complete payload carries it, for the failure reason."""
+        return {
+            "code": self.code,
+            "attr": self.attr,
+            "module": self.module,
+            "severity": self.severity,
+            # Lets the queue's failure reason label an `hms[]` fault
+            # with its full code rather than a lossy short one (#2728).
+            "full_code": self.full_code,
+            # Carried so the queue's failure reason quotes the same
+            # sentence the status response and the broadcast do,
+            # rather than resolving the code a fourth time (#2926).
+            "description": self.description,
+        }
+
 
 # HMS short codes the firmware emits during normal user-cancel sequences.
 # These aren't faults — they're status echoes that confirm the cancel happened.
@@ -5966,26 +5982,7 @@ class BambuMQTTClient:
             self._was_running = False
             self._timelapse_during_print = False  # Reset for next print
             # Include HMS errors for failure reason detection
-            hms_errors_data = (
-                [
-                    {
-                        "code": e.code,
-                        "attr": e.attr,
-                        "module": e.module,
-                        "severity": e.severity,
-                        # Lets the queue's failure reason label an `hms[]` fault
-                        # with its full code rather than a lossy short one (#2728).
-                        "full_code": e.full_code,
-                        # Carried so the queue's failure reason quotes the same
-                        # sentence the status response and the broadcast do,
-                        # rather than resolving the code a fourth time (#2926).
-                        "description": e.description,
-                    }
-                    for e in self.state.hms_errors
-                ]
-                if self.state.hms_errors
-                else []
-            )
+            hms_errors_data = [e.completion_payload() for e in self.state.hms_errors or []]
             self.on_print_complete(
                 {
                     "status": status,

backend/app/services/print_log.py

@@ -50,6 +50,7 @@ async def write_log_entry(
     created_by_id: int | None = None,
     created_by_username: str | None = None,
     reconciled: bool = False,
+    duration_unmeasured: bool = False,
 ) -> PrintLogEntry:
     """Write a print log entry.
 
@@ -62,6 +63,10 @@ async def write_log_entry(
     entries we store an explicit ``0`` ("no measured runtime") rather than a
     fabricated duration; the stats total trusts a stored 0 instead of
     recomputing from the stale timestamps.
+
+    ``duration_unmeasured`` marks a run closed at reconnect, whose times were
+    not measured (#3310): ``completed_at`` is its start plus the slicer's
+    estimate, or the reconnect. The time accuracy leaves it out.
     """
     if reconciled:
         duration: int | None = 0
@@ -87,6 +92,7 @@ async def write_log_entry(
         "energy_kwh": energy_kwh,
         "energy_cost": energy_cost,
         "wear_cost": wear_cost_for_run(duration, wear_cost_per_hour),
+        "duration_unmeasured": True if duration_unmeasured else None,
         "failure_reason": failure_reason,
         "thumbnail_path": thumbnail_path,
         "created_by_id": created_by_id,

backend/tests/integration/test_print_complete_end_to_end.py

@@ -9,7 +9,7 @@ printer's SD card) is mocked; the printer manager knows no printer.
 """
 
 import asyncio
-from datetime import datetime
+from datetime import datetime, timezone
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -143,7 +143,7 @@ async def test_a_completed_print(test_engine, db_session, queued_print):
     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 [(log.status, log.duration_unmeasured) for log in logs] == [("completed", None)]
     assert len(charges) == 1
     charge = charges[0]
     assert (charge.user_id, charge.cost_center_id, charge.print_queue_id, charge.print_run_id) == (
@@ -416,3 +416,105 @@ async def test_a_hung_sd_cleanup_does_not_hold_up_the_print_s_end(test_engine, d
     _archive, item, _logs, charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
     assert item.status == "completed"
     assert len(charges) == 1
+
+
+async def test_a_reconciled_abort_says_no_status_update(test_engine, db_session, queued_print):
+    await _complete(test_engine, queued_print["printer"], "aborted", extra={"_reconciled": True})
+
+    archive, item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert (archive.status, archive.failure_reason) == ("aborted", "noStatusUpdate")
+    assert [(log.status, log.duration_seconds, log.duration_unmeasured) for log in logs] == [("aborted", 0, True)]
+
+
+async def test_a_print_that_finished_while_bambuddy_restarted_is_completed(test_engine, db_session, queued_print):
+    """#3310: the reconnect closes it as completed, ended at its estimate."""
+    ended = datetime(2026, 10, 19, 11, 0, tzinfo=timezone.utc)
+    notif = await _complete(
+        test_engine,
+        queued_print["printer"],
+        "completed",
+        extra={"_reconciled": True, "_reconciled_completed_at": ended},
+    )
+
+    archive, item, logs, charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert (archive.status, archive.failure_reason) == ("completed", None)
+    assert archive.completed_at.replace(tzinfo=None) == datetime(2026, 10, 19, 11, 0)
+    assert item.status == "completed"
+    assert [(log.status, log.failure_reason, log.duration_seconds, log.duration_unmeasured) for log in logs] == [
+        ("completed", None, 3 * 3600, True)
+    ]
+    assert len(charges) == 1
+    assert notif.on_print_complete.await_args.kwargs["archive_data"]["actual_time_seconds"] == 3 * 3600
+
+
+async def test_a_reconciled_completion_without_an_estimate_keeps_no_duration(test_engine, db_session, queued_print):
+    await _complete(test_engine, queued_print["printer"], "completed", extra={"_reconciled": True})
+
+    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.duration_seconds, log.duration_unmeasured) for log in logs] == [("completed", 0, True)]
+
+
+@pytest.mark.parametrize(
+    ("hms_errors", "reason"),
+    [
+        ([{"code": "0x4057", "attr": 0x0300_0000, "module": 0x03, "severity": 1}], "layerShift"),
+        ([], "noStatusUpdate"),
+    ],
+)
+async def test_a_reconciled_failure_takes_the_printer_s_reason(
+    test_engine, db_session, queued_print, hms_errors, reason
+):
+    await _complete(
+        test_engine,
+        queued_print["printer"],
+        "failed",
+        extra={"_reconciled": True, "hms_errors": hms_errors},
+    )
+
+    archive, _item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    assert (archive.status, archive.failure_reason) == ("failed", reason)
+    assert [(log.status, log.duration_seconds, log.duration_unmeasured) for log in logs] == [("failed", 0, True)]
+
+
+async def test_the_reconcile_reads_the_plate_the_queue_job_printed(test_engine, db_session, queued_print):
+    """#3310: a reprint of plate 3 reuses an archive that says plate 1."""
+    from types import SimpleNamespace
+
+    archive = await db_session.get(PrintArchive, queued_print["archive"])
+    archive.subtask_id, archive.plate_id = "JOB1", 1
+    item = await db_session.get(PrintQueueItem, queued_print["item"])
+    item.plate_id = 3
+    # An earlier run of the archive, of another plate.
+    db_session.add(
+        PrintQueueItem(
+            printer_id=queued_print["printer"],
+            archive_id=archive.id,
+            status="completed",
+            position=2,
+            plate_id=5,
+        )
+    )
+    await db_session.commit()
+    state = SimpleNamespace(
+        state="FINISH",
+        subtask_id="JOB1",
+        subtask_name="part",
+        connected=True,
+        raw_data={},
+        progress=100,
+        layer_num=0,
+        hms_errors=[],
+    )
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    with (
+        patch("backend.app.main.async_session", maker),
+        patch("backend.app.main.printer_manager") as pm,
+        patch("backend.app.main._reconciled_end_time", return_value=None) as end_time,
+        patch("backend.app.main.on_print_complete", new=AsyncMock()) as complete,
+    ):
+        pm.get_status.return_value = state
+        assert await main_module.reconcile_stale_active_prints(queued_print["printer"]) == 1
+
+    assert end_time.call_args.args[2] == 3
+    assert complete.call_args.args[1]["status"] == "completed"

backend/tests/unit/test_archive_run_aggregation.py

@@ -496,3 +496,107 @@ async def test_archive_list_keeps_time_accuracy_for_single_run_archives(
     assert row["actual_time_seconds"] == 3600
     assert row["time_accuracy"] == 100.0
     assert row["run_count"] == 1
+
+
+# ---------------------------------------------------------------------------
+# #3310: a print that finished while Bambuddy restarted is logged with its
+# slicer estimate as its duration. That is no measurement, so neither the
+# card's badge nor the Statistics average may compare it with the estimate.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_an_estimated_run_shows_no_time_accuracy(
+    async_client: AsyncClient, archive_factory, printer_factory, db_session
+):
+    printer = await printer_factory()
+    archive = await archive_factory(
+        printer.id,
+        status="completed",
+        print_time_seconds=3600,
+        started_at=datetime(2026, 5, 1, 10, 0, tzinfo=timezone.utc),
+        completed_at=datetime(2026, 5, 1, 11, 0, tzinfo=timezone.utc),
+        with_run=False,
+    )
+    db_session.add(
+        PrintLogEntry(
+            archive_id=archive.id,
+            printer_id=printer.id,
+            status="completed",
+            started_at=datetime(2026, 5, 1, 10, 0, tzinfo=timezone.utc),
+            completed_at=datetime(2026, 5, 1, 11, 0, tzinfo=timezone.utc),
+            duration_seconds=3600,
+            duration_unmeasured=True,
+        )
+    )
+    await db_session.commit()
+
+    row = next(r for r in (await async_client.get("/api/v1/archives/")).json() if r["id"] == archive.id)
+    assert (row["actual_time_seconds"], row["time_accuracy"]) == (None, None)
+    assert row["print_time_seconds"] == 3600
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_an_estimated_run_is_no_accuracy_sample_but_counts_its_time(
+    async_client: AsyncClient, archive_factory, printer_factory, db_session
+):
+    printer = await printer_factory()
+    measured = await archive_factory(printer.id, print_time_seconds=3600, with_run=False)
+    estimated = await archive_factory(printer.id, print_time_seconds=7200, with_run=False)
+    db_session.add(
+        PrintLogEntry(archive_id=measured.id, printer_id=printer.id, status="completed", duration_seconds=4000)
+    )
+    db_session.add(
+        PrintLogEntry(
+            archive_id=estimated.id,
+            printer_id=printer.id,
+            status="completed",
+            duration_seconds=7200,
+            duration_unmeasured=True,
+        )
+    )
+    await db_session.commit()
+
+    body = (await async_client.get("/api/v1/archives/stats")).json()
+    assert body["average_time_accuracy"] == pytest.approx(90.0, abs=0.1)
+    assert body["time_accuracy_by_printer"][str(printer.id)] == pytest.approx(90.0, abs=0.1)
+    assert body["total_print_time_hours"] == pytest.approx(11200 / 3600, abs=0.06)
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_a_run_without_an_estimate_shows_no_reconnect_gap(
+    async_client: AsyncClient, archive_factory, printer_factory, db_session
+):
+    """#3310: with no estimate the archive ends at the reconnect, 9 h after a
+    6 h print started; that gap is neither its time nor an accuracy sample."""
+    printer = await printer_factory()
+    started = datetime(2026, 5, 1, 10, 0, tzinfo=timezone.utc)
+    reconnect = datetime(2026, 5, 1, 19, 0, tzinfo=timezone.utc)
+    archive = await archive_factory(
+        printer.id,
+        status="completed",
+        print_time_seconds=6 * 3600,
+        started_at=started,
+        completed_at=reconnect,
+        with_run=False,
+    )
+    db_session.add(
+        PrintLogEntry(
+            archive_id=archive.id,
+            printer_id=printer.id,
+            status="completed",
+            started_at=started,
+            completed_at=reconnect,
+            duration_seconds=0,
+            duration_unmeasured=True,
+        )
+    )
+    await db_session.commit()
+
+    row = next(r for r in (await async_client.get("/api/v1/archives/")).json() if r["id"] == archive.id)
+    assert (row["actual_time_seconds"], row["time_accuracy"]) == (None, None)
+    body = (await async_client.get("/api/v1/archives/stats")).json()
+    assert body["average_time_accuracy"] is None

backend/tests/unit/test_reconcile_stale_active_prints.py

@@ -19,15 +19,26 @@ These tests cover:
     each stale archive.
 """
 
+from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
-from backend.app.main import _is_active_archive_stale
+from backend.app.main import _is_active_archive_stale, _reconciled_end_time, _reconciled_outcome
+from backend.app.services.bambu_mqtt import HMSError
 
 
-def _state(state: str, *, subtask_id: str = "", subtask_name: str = "", connected: bool = True) -> SimpleNamespace:
+def _state(
+    state: str,
+    *,
+    subtask_id: str = "",
+    subtask_name: str = "",
+    connected: bool = True,
+    progress: float = 0.0,
+    layer_num: int = 0,
+    hms_errors: list | None = None,
+) -> SimpleNamespace:
     """Minimal PrinterState stub for the pure decision function."""
     return SimpleNamespace(
         state=state,
@@ -35,11 +46,19 @@ def _state(state: str, *, subtask_id: str = "", subtask_name: str = "", connecte
         subtask_name=subtask_name,
         connected=connected,
         raw_data={},
+        progress=progress,
+        layer_num=layer_num,
+        hms_errors=hms_errors or [],
     )
 
 
 def _archive(
-    subtask_id: str | None = "ABC123", filename: str = "ghost.3mf", print_name: str = "ghost"
+    subtask_id: str | None = "ABC123",
+    filename: str = "ghost.3mf",
+    print_name: str = "ghost",
+    *,
+    started_at: datetime | None = None,
+    print_time_seconds: int | None = None,
 ) -> SimpleNamespace:
     """Minimal PrintArchive stub — only the fields the decision function reads."""
     return SimpleNamespace(
@@ -47,6 +66,10 @@ def _archive(
         subtask_id=subtask_id,
         filename=filename,
         print_name=print_name,
+        started_at=started_at,
+        print_time_seconds=print_time_seconds,
+        plate_id=None,
+        file_path=None,
     )
 
 
@@ -264,3 +287,159 @@ class TestReconcileStaleActivePrints:
         # Only the second archive is recorded as reconciled (first raised).
         assert count == 1
         assert mock_complete.await_count == 2
+
+
+class TestReconciledOutcome:
+    """#3310: only the printer's FINISH or FAILED for the archive's own job
+    says how the print ended; anything else stays aborted."""
+
+    @pytest.mark.parametrize(
+        ("printer_state", "archive_id", "printer_id", "outcome"),
+        [
+            ("FINISH", "ABC123", "ABC123", "completed"),
+            ("finish", "ABC123", "ABC123", "completed"),
+            ("FAILED", "ABC123", "ABC123", "failed"),
+            # Reported as a number by some firmware.
+            ("FINISH", "123", 123, "completed"),
+            ("IDLE", "ABC123", "ABC123", "aborted"),
+            # A print started from the screen while Bambuddy was down.
+            ("FINISH", "ABC123", "XYZ789", "aborted"),
+            ("FAILED", "ABC123", "XYZ789", "aborted"),
+            ("FINISH", "ABC123", "", "aborted"),
+            ("FINISH", None, "", "aborted"),
+            # "0" is every print started outside the cloud, not one job.
+            ("FINISH", "0", "0", "aborted"),
+        ],
+    )
+    def test_outcome(self, printer_state, archive_id, printer_id, outcome):
+        archive = _archive(subtask_id=archive_id)
+        state = _state(printer_state, subtask_id=printer_id, subtask_name="ghost")
+        assert _reconciled_outcome(archive, state) == outcome
+
+
+class TestReconciledEndTime:
+    NOW = datetime(2026, 10, 20, 12, 0, tzinfo=timezone.utc)
+
+    def test_start_plus_estimate(self):
+        archive = _archive(started_at=datetime(2026, 10, 20, 6, 0), print_time_seconds=3 * 3600)
+        assert _reconciled_end_time(archive, self.NOW) == datetime(2026, 10, 20, 9, 0, tzinfo=timezone.utc)
+
+    def test_never_later_than_now(self):
+        archive = _archive(started_at=datetime(2026, 10, 20, 11, 0), print_time_seconds=5 * 3600)
+        assert _reconciled_end_time(archive, self.NOW) == self.NOW
+
+    @pytest.mark.parametrize(
+        ("started_at", "estimate"),
+        [(None, 3600), (datetime(2026, 10, 20, 6, 0), None), (datetime(2026, 10, 20, 6, 0), 0)],
+    )
+    def test_unknown_without_start_or_estimate(self, started_at, estimate):
+        archive = _archive(started_at=started_at, print_time_seconds=estimate)
+        assert _reconciled_end_time(archive, self.NOW) is None
+
+    READ = "backend.app.utils.threemf_tools.extract_print_time_from_3mf"
+
+    def _multi_plate(self, plate_id=None):
+        archive = _archive(started_at=datetime(2026, 10, 20, 6, 0), print_time_seconds=5 * 3600)
+        archive.plate_id, archive.file_path = plate_id, "archives/x/part.3mf"
+        return archive
+
+    def test_the_plate_s_own_estimate(self):
+        """A multi-plate archive carries the whole file's estimate (#1593)."""
+        with patch(self.READ, return_value=1800) as read:
+            ended = _reconciled_end_time(self._multi_plate(plate_id=2), self.NOW)
+        assert ended == datetime(2026, 10, 20, 6, 30, tzinfo=timezone.utc)
+        assert read.call_args.args[1] == 2
+
+    def test_the_run_s_plate_over_the_archive_s(self):
+        """A reprint of plate 3 reuses an archive that still says plate 1."""
+        with patch(self.READ, return_value=1800) as read:
+            _reconciled_end_time(self._multi_plate(plate_id=1), self.NOW, plate_id=3)
+        assert read.call_args.args[1] == 3
+
+    def test_no_plate_on_a_single_plate_file(self):
+        with patch(self.READ, return_value=5 * 3600):
+            ended = _reconciled_end_time(self._multi_plate(), self.NOW)
+        assert ended == datetime(2026, 10, 20, 11, 0, tzinfo=timezone.utc)
+
+    def test_no_plate_on_a_multi_plate_file_is_unknown(self):
+        """The first plate's estimate differs from the sum: which plate printed?"""
+        with patch(self.READ, return_value=1800):
+            assert _reconciled_end_time(self._multi_plate(), self.NOW) is None
+
+    def test_an_unreadable_file_is_unknown(self):
+        with patch(self.READ, side_effect=OSError("gone")):
+            assert _reconciled_end_time(self._multi_plate(plate_id=2), self.NOW) is None
+
+
+class TestReconcilePayload:
+    async def _payload(self, archive, state):
+        from backend.app.main import reconcile_stale_active_prints
+
+        with patch("backend.app.main.printer_manager") as mock_pm:
+            mock_pm.get_status.return_value = state
+            with patch("backend.app.main.async_session") as mock_session:
+                session_ctx = AsyncMock()
+                session_ctx.execute = AsyncMock(
+                    return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [archive]))
+                )
+                mock_session.return_value.__aenter__.return_value = session_ctx
+                with patch("backend.app.main.on_print_complete", new=AsyncMock()) as mock_complete:
+                    assert await reconcile_stale_active_prints(printer_id=1) == 1
+        return mock_complete.call_args.args[1]
+
+    @pytest.mark.asyncio
+    async def test_finished_while_bambuddy_restarted(self):
+        started = datetime.now(timezone.utc) - timedelta(hours=10)
+        archive = _archive(subtask_id="ABC123", started_at=started, print_time_seconds=2 * 3600)
+        payload = await self._payload(
+            archive, _state("FINISH", subtask_id="ABC123", subtask_name="ghost", progress=100, layer_num=250)
+        )
+        assert payload["status"] == "completed"
+        assert payload["_reconciled"] is True
+        assert payload["_reconciled_completed_at"] == started + timedelta(hours=2)
+        assert (payload["last_progress"], payload["last_layer_num"]) == (100, 250)
+        assert "hms_errors" not in payload
+
+    @pytest.mark.asyncio
+    async def test_failed_while_bambuddy_restarted(self):
+        error = HMSError(code="0x4057", attr=0x0300_0000, module=0x03, severity=1, full_code="0300000000004057")
+        payload = await self._payload(
+            _archive(subtask_id="ABC123"),
+            _state("FAILED", subtask_id="ABC123", subtask_name="ghost", progress=40, layer_num=80, hms_errors=[error]),
+        )
+        assert payload["status"] == "failed"
+        assert (payload["last_progress"], payload["last_layer_num"]) == (40, 80)
+        assert payload["hms_errors"] == [error.completion_payload()]
+        assert "_reconciled_completed_at" not in payload
+
+    @pytest.mark.asyncio
+    async def test_another_job_s_progress_is_not_this_archive_s(self):
+        archive = _archive(subtask_id="ABC123", started_at=datetime.now(timezone.utc), print_time_seconds=3600)
+        payload = await self._payload(
+            archive, _state("FINISH", subtask_id="XYZ789", subtask_name="other", progress=100, layer_num=250)
+        )
+        assert payload["status"] == "aborted"
+        for key in ("last_progress", "last_layer_num", "hms_errors", "_reconciled_completed_at"):
+            assert key not in payload
+
+
+@pytest.mark.asyncio
+async def test_the_queue_job_s_plate_reaches_the_estimate():
+    from backend.app.main import reconcile_stale_active_prints
+
+    archive = _archive(subtask_id="ABC123", started_at=datetime.now(timezone.utc), print_time_seconds=3600)
+    archive.plate_id = 1
+    archives = MagicMock(scalars=lambda: MagicMock(all=lambda: [archive]))
+    plates = MagicMock(all=lambda: [(archive.id, 3)])
+    with patch("backend.app.main.printer_manager") as mock_pm:
+        mock_pm.get_status.return_value = _state("FINISH", subtask_id="ABC123", subtask_name="ghost")
+        with patch("backend.app.main.async_session") as mock_session:
+            session_ctx = AsyncMock()
+            session_ctx.execute = AsyncMock(side_effect=[archives, plates])
+            mock_session.return_value.__aenter__.return_value = session_ctx
+            with (
+                patch("backend.app.main._reconciled_end_time", return_value=None) as end_time,
+                patch("backend.app.main.on_print_complete", new=AsyncMock()),
+            ):
+                await reconcile_stale_active_prints(printer_id=1)
+    assert end_time.call_args.args[2] == 3