fix(prints): connected-edge reconciliation closes the missed-PRINT-COMPLETE loop behind smart-plug ghost prints (#1542 follow-up)

  Reporter ran a fresh trace after the doubled-extension fix landed and
  found a distinct second cause behind his ghost prints, hitting 4-of-4
  of his A1s. Timeline:

    22:50 PRINT START
    ...print runs all night...
    23:13 / 00:47 / 09:35 MQTT disconnects (A1 keepalives are unstable)
    print finishes during one of those disconnect windows → PRINT COMPLETE
      is never observed
    smart plug detects idle → cuts power
    power resumes for the next scheduled print → firmware auto-replays the
      leftover .3mf from the SD card
    09:46 Bambuddy reconnects to a fresh PRINT START for the ghost

  The existing IDLE-after-RUNNING completion check at
  backend/app/services/bambu_mqtt.py:3022 was meant to catch the simple
  disconnect-then-finish case via `_previous_gcode_state` preserved across
  reconnects, but with multiple disconnect/reconnect cycles + a smart-plug
  power-off that Bambuddy can't distinguish from any other transient drop,
  the IDLE window that branch needs simply never reaches it. The SD .3mf
  lingers, the firmware ghost-replays every power cycle, and the loop
  repeats.

  Fix: new connected-edge reconciliation pass.

  * `_is_active_archive_stale(archive, state)` — pure decision function
    with three triggers:
      (1) printer state is terminal (IDLE / FINISH / FAILED)
      (2) printer running with a different `subtask_id` than the archive —
          Bambu firmware mints a fresh subtask_id for each print including
          the ghost replay, so a mismatch is unambiguous
      (3) printer running but `subtask_name` is empty — printer doesn't
          know what it's running, archive reference is broken
    Conservative on PAUSE / PREPARE / SLICING and on RUNNING with matching
    subtask. False-positive cost = one misreported "aborted" status that
    the next real PRINT COMPLETE would have overwritten anyway. False-
    negative cost = the ghost-print loop.

  * `reconcile_stale_active_prints(printer_id)` — queries archives in
    `status="printing"` for the printer, runs the decision function, and
    synthesises `on_print_complete(status="aborted", _reconciled=True)`
    for each stale match. Reuses the existing PRINT COMPLETE chain (SD
    cleanup, status update, usage tracker, notifications) — no reimplem-
    entation. Per-archive try/except so one failure doesn't block the
    rest. Returns 0 when status is None / disconnected — the connected
    edge is the only legitimate trigger.

  * `on_printer_status_change` now runs a connected-edge check at the
    start. New `_printer_reconciled_since_connect: dict[int, bool]`
    tracker flips False → True on the first connected status update for
    this connection and back to False on disconnect, so reconciliation
    fires exactly once per (re)connection. The flag is set BEFORE the
    task is spawned so concurrent status updates within the same
    connection don't re-trigger it (no await between check and set,
    asyncio guarantees atomicity). The reconciliation runs as
    `asyncio.create_task` so the hot WebSocket dedup / broadcast path
    isn't blocked.

  * Single handler covers both startup and reconnect — when the first
    MQTT connection completes after startup the printer pushes status,
    the connected edge fires, reconciliation runs. No separate startup
    hook needed.

  Idempotency: when the existing #3022 branch DOES fire on a clean
  disconnect-then-IDLE-on-reconnect, it lands `archive.status` to terminal
  synchronously. The async-scheduled reconciliation then queries
  `WHERE status="printing"` and finds 0 rows → no-op. The narrow race
  where reconcile's query lands between a real on_print_complete's
  archive update and its archive lookup produces at most a duplicate
  notification (no double SD-cleanup since FTP delete on a missing file
  is a no-op 550).

  Ghost-print collateral worth being explicit about: if the ghost is
  already running when reconciliation fires, the synthesised SD-cleanup
  hits 550-file-locked (same root cause as the #1542 first case). The
  cleanup retries 3× then logs "lingering". The ghost runs to completion,
  its own end-of-print cleanup deletes the file, the next power cycle
  has nothing to replay, the loop breaks. A perfect cancel would require
  a `print_stop` MQTT command to the printer mid-ghost — invasive,
  explicitly out of scope.

backend/app/main.py

@@ -330,6 +330,14 @@ logging.info("Bambuddy starting - debug=%s, log_level=%s", app_settings.debug, l
 # Track active prints: {(printer_id, filename): archive_id}
 _active_prints: dict[tuple[int, str], int] = {}
 
+# Per-printer "connected" edge tracker. Used by `on_printer_status_change`
+# to fire `reconcile_stale_active_prints` exactly once per (re)connection
+# (#1542 follow-up — power-cycle ghost prints). The value is True after
+# the first connected status update for that connection; transitions back
+# to False whenever we observe `state.connected = False` so the next
+# reconnect re-arms reconciliation. Keyed by printer_id.
+_printer_reconciled_since_connect: dict[int, bool] = {}
+
 # Track expected prints from reprint/scheduled (skip auto-archiving for these)
 # {(printer_id, filename): archive_id}
 _expected_prints: dict[tuple[int, str], int] = {}
@@ -798,6 +806,26 @@ _nozzle_count_updated: set[int] = set()
 
 async def on_printer_status_change(printer_id: int, state: PrinterState):
     """Handle printer status changes - broadcast via WebSocket."""
+    # Connected-edge reconciliation (#1542 follow-up). When the printer
+    # transitions disconnected → connected — which covers both Bambuddy
+    # startup (no prior connection) and a mid-session MQTT reconnect — fire
+    # `reconcile_stale_active_prints` exactly once for this connection so
+    # any archive still in `status="printing"` that can't actually be
+    # running anymore (printer IDLE / different subtask / empty subtask)
+    # gets a synthesised PRINT COMPLETE. Without this, a print that
+    # finished during a disconnect window + a smart-plug power cycle
+    # leaves the .3mf on the SD card and the firmware ghost-replays it on
+    # next boot. Reconciliation runs concurrently — it must not block the
+    # WebSocket dedup / broadcast logic below, and the connected edge is
+    # marked True BEFORE the await so concurrent status updates inside
+    # the same connection don't re-trigger reconciliation.
+    if state.connected and not _printer_reconciled_since_connect.get(printer_id, False):
+        _printer_reconciled_since_connect[printer_id] = True
+        asyncio.create_task(reconcile_stale_active_prints(printer_id))
+    elif not state.connected and _printer_reconciled_since_connect.get(printer_id, False):
+        # Re-arm so the next reconnect triggers reconciliation again.
+        _printer_reconciled_since_connect[printer_id] = False
+
     # Only broadcast if something meaningful changed (reduce WebSocket spam)
     # Include rounded temperatures to detect meaningful temp changes (within 1 degree)
     temps = state.temperatures or {}
@@ -3163,6 +3191,136 @@ async def on_print_running_observed(printer_id: int, data: dict):
     await _capture_timelapse_baseline_at_start(printer, printer_id, logger)
 
 
+def _is_active_archive_stale(archive, state) -> tuple[bool, str]:
+    """Return ``(is_stale, reason)`` for an archive in ``status="printing"``
+    against the printer's current MQTT state.
+
+    Reconciliation triggers (#1542 follow-up — recovers from missed PRINT
+    COMPLETE events, typically a print finishing during an MQTT disconnect
+    window followed by a smart-plug power cycle):
+
+      1. Printer state is terminal (IDLE / FINISH / FAILED). The print is
+         provably not running anymore — only branch that should fire under
+         normal disconnect-then-reconnect timing.
+      2. Printer has a different ``subtask_id`` than the archive. Bambu
+         firmware mints a fresh ``subtask_id`` for each print, including the
+         ghost replay it runs after a power cycle from a leftover SD file —
+         so a mismatch unambiguously means the in-DB archive is no longer
+         the print on the printer.
+      3. Printer is running but ``subtask_name`` is empty. The printer
+         doesn't know what it's running; the archive's reference to it is
+         already broken.
+
+    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 single misreported "aborted" status that the next real
+    PRINT COMPLETE would have overwritten with the correct status anyway.
+    The cost of a false negative is the ghost-print loop in #1542.
+    """
+    current_state = (state.state or "").upper()
+    if current_state in ("IDLE", "FINISH", "FAILED"):
+        return True, f"printer state {current_state}"
+    # Below here the printer is in a running / pre-running state (RUNNING /
+    # PAUSE / PREPARE / SLICING / etc.) — decide based on subtask identity.
+    current_subtask_id = (state.subtask_id or "").strip()
+    if archive.subtask_id and current_subtask_id and archive.subtask_id != current_subtask_id:
+        return True, f"subtask_id changed ({archive.subtask_id!r} → {current_subtask_id!r})"
+    current_subtask_name = (state.subtask_name or "").strip()
+    if not current_subtask_name:
+        return True, "printer subtask_name empty"
+    return False, ""
+
+
+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.
+
+    Called once per MQTT (re)connection (from on_printer_status_change when
+    the connected edge flips False → True) and at Bambuddy startup (from
+    the FastAPI lifespan). Without this, a print that completes during a
+    disconnect window — followed by a smart-plug-driven power cycle — leaves
+    the ``.3mf`` on the SD card, the firmware auto-replays it on next boot,
+    and Bambuddy fires a fresh PRINT START for the ghost rather than the
+    SD cleanup that PRINT COMPLETE was supposed to run. Repeats every
+    power cycle until the operator notices (#1542 follow-up). Reconciliation
+    closes the loop by faking the missed PRINT COMPLETE — the existing
+    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.
+
+    Returns the number of archives reconciled.
+    """
+    state = printer_manager.get_status(printer_id)
+    if not state:
+        return 0
+    # Don't reconcile while disconnected — we'd be making a decision against
+    # stale cached state. The connected → reconcile edge handles this.
+    if not state.connected:
+        return 0
+
+    from backend.app.models.archive import PrintArchive
+
+    reconciled = 0
+    async with async_session() as db:
+        result = await db.execute(
+            select(PrintArchive).where(
+                PrintArchive.printer_id == printer_id,
+                PrintArchive.status == "printing",
+            )
+        )
+        active = list(result.scalars().all())
+
+    if not active:
+        return 0
+
+    logger = logging.getLogger(__name__)
+    for archive in active:
+        is_stale, reason = _is_active_archive_stale(archive, state)
+        if not is_stale:
+            continue
+        logger.info(
+            "[RECONCILE] Printer %s: synthesising missed PRINT COMPLETE for archive %s (%s) — %s",
+            printer_id,
+            archive.id,
+            archive.filename,
+            reason,
+        )
+        # Synthesised payload: minimal fields the on_print_complete chain
+        # needs. `_reconciled` marker lets downstream code distinguish this
+        # from a real MQTT-driven completion if it ever needs to (e.g. for
+        # 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.
+        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,
+                },
+            )
+            reconciled += 1
+        except Exception as e:
+            # Catch-all: a reconciliation failure must not block the
+            # printer's normal status flow. The archive stays in
+            # ``status="printing"`` and the next reconnect retries.
+            logger.warning(
+                "[RECONCILE] on_print_complete synthesis failed for archive %s: %s",
+                archive.id,
+                e,
+            )
+
+    return reconciled
+
+
 async def on_print_complete(printer_id: int, data: dict):
     """Handle print completion - update the archive status."""
     import time

backend/tests/unit/test_reconcile_stale_active_prints.py

@@ -0,0 +1,248 @@
+"""Tests for the connected-edge reconciliation that recovers from missed
+PRINT COMPLETE events (#1542 follow-up).
+
+Background: the PRINT COMPLETE MQTT callback is purely reactive to a single
+state transition (RUNNING → IDLE / FINISH / FAILED). When the printer
+finishes during an MQTT disconnect window — typical on the A1 line with
+unstable MQTT keepalives — Bambuddy never observes the transition. If a
+smart plug then cuts power between completion and the next reconnect, the
+firmware auto-replays whatever's still on the SD card and produces a ghost
+print on next power-up. Reporter (#1542 second case) saw this hit 4 out of
+4 of his A1s.
+
+These tests cover:
+  * `_is_active_archive_stale` — the pure decision function for whether an
+    archive in `status="printing"` should be reconciled given the printer's
+    current state.
+  * `reconcile_stale_active_prints` — the orchestrator that queries the DB,
+    runs the decision function, and synthesises `on_print_complete` for
+    each stale archive.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.main import _is_active_archive_stale
+
+
+def _state(state: str, *, subtask_id: str = "", subtask_name: str = "", connected: bool = True) -> SimpleNamespace:
+    """Minimal PrinterState stub for the pure decision function."""
+    return SimpleNamespace(
+        state=state,
+        subtask_id=subtask_id,
+        subtask_name=subtask_name,
+        connected=connected,
+        raw_data={},
+    )
+
+
+def _archive(
+    subtask_id: str | None = "ABC123", filename: str = "ghost.3mf", print_name: str = "ghost"
+) -> SimpleNamespace:
+    """Minimal PrintArchive stub — only the fields the decision function reads."""
+    return SimpleNamespace(
+        id=42,
+        subtask_id=subtask_id,
+        filename=filename,
+        print_name=print_name,
+    )
+
+
+class TestIsActiveArchiveStale:
+    """Decision function — covers all three stale triggers + the
+    intentionally-conservative no-op cases."""
+
+    # Trigger 1: printer is in a terminal state.
+    @pytest.mark.parametrize("terminal_state", ["IDLE", "FINISH", "FAILED", "idle", "finish", "failed"])
+    def test_terminal_state_marks_stale(self, terminal_state):
+        archive = _archive(subtask_id="ABC123")
+        state = _state(terminal_state, subtask_id="ABC123", subtask_name="ghost")
+        is_stale, reason = _is_active_archive_stale(archive, state)
+        assert is_stale is True
+        assert terminal_state.upper() in reason
+
+    # Trigger 2: printer is running a different subtask_id.
+    def test_subtask_id_changed_marks_stale(self):
+        archive = _archive(subtask_id="OLD_ID")
+        state = _state("RUNNING", subtask_id="NEW_ID", subtask_name="something")
+        is_stale, reason = _is_active_archive_stale(archive, state)
+        assert is_stale is True
+        assert "subtask_id" in reason
+        assert "OLD_ID" in reason
+        assert "NEW_ID" in reason
+
+    # Trigger 3: printer is running but doesn't know what it's running.
+    def test_empty_subtask_name_marks_stale(self):
+        archive = _archive(subtask_id="ABC123")
+        state = _state("RUNNING", subtask_id="", subtask_name="")
+        is_stale, reason = _is_active_archive_stale(archive, state)
+        assert is_stale is True
+        assert "empty" in reason.lower() or "subtask_name" in reason
+
+    # Healthy case: same subtask_id, running.
+    def test_matching_running_print_not_stale(self):
+        archive = _archive(subtask_id="ABC123")
+        state = _state("RUNNING", subtask_id="ABC123", subtask_name="ghost")
+        is_stale, _ = _is_active_archive_stale(archive, state)
+        assert is_stale is False
+
+    # PAUSE is not a stale signal — the print is paused, not ended.
+    def test_paused_print_with_matching_subtask_not_stale(self):
+        archive = _archive(subtask_id="ABC123")
+        state = _state("PAUSE", subtask_id="ABC123", subtask_name="ghost")
+        is_stale, _ = _is_active_archive_stale(archive, state)
+        assert is_stale is False
+
+    # PREPARE / SLICING are not stale either — pre-print phases.
+    @pytest.mark.parametrize("pre_running_state", ["PREPARE", "SLICING"])
+    def test_pre_running_states_with_matching_subtask_not_stale(self, pre_running_state):
+        archive = _archive(subtask_id="ABC123")
+        state = _state(pre_running_state, subtask_id="ABC123", subtask_name="ghost")
+        is_stale, _ = _is_active_archive_stale(archive, state)
+        assert is_stale is False
+
+    # Missing subtask_id on the archive side: don't have evidence either
+    # way, fall through to the empty-subtask_name check.
+    def test_archive_with_no_subtask_id_falls_to_subtask_name_check(self):
+        archive = _archive(subtask_id=None)
+        state = _state("RUNNING", subtask_id="ANYTHING", subtask_name="something")
+        # Subtask_name is populated → not stale, no false positive.
+        is_stale, _ = _is_active_archive_stale(archive, state)
+        assert is_stale is False
+
+    # Missing subtask_id on both sides: still triggers the empty-subtask_name
+    # branch if the printer doesn't know what it's running.
+    def test_both_subtask_ids_missing_running_with_empty_name_stale(self):
+        archive = _archive(subtask_id=None)
+        state = _state("RUNNING", subtask_id="", subtask_name="")
+        is_stale, _ = _is_active_archive_stale(archive, state)
+        assert is_stale is True
+
+    # IDLE wins over PRINT-STATE checks — the terminal-state branch fires
+    # first regardless of what the subtask fields look like.
+    def test_idle_state_overrides_matching_subtask(self):
+        archive = _archive(subtask_id="ABC123")
+        state = _state("IDLE", subtask_id="ABC123", subtask_name="ghost")
+        is_stale, reason = _is_active_archive_stale(archive, state)
+        assert is_stale is True
+        assert "IDLE" in reason
+
+
+class TestReconcileStaleActivePrints:
+    """Orchestrator-level tests — mock the printer manager + DB session so
+    we can drive the decision flow end-to-end without standing up real
+    fixtures.
+
+    These cover:
+      * No printer status (disconnected) → no-op, no on_print_complete fired.
+      * No active archives → no-op.
+      * Stale archive → synthesised on_print_complete called with status
+        ``"aborted"`` and the `_reconciled: True` marker so downstream code
+        can distinguish synthetic from real completions.
+      * Non-stale archive → on_print_complete NOT called (no false positive
+        on a healthy in-flight print).
+      * Exception inside on_print_complete must NOT block reconciliation
+        for subsequent archives or crash the caller.
+    """
+
+    @pytest.mark.asyncio
+    async def test_no_status_skips_reconciliation(self):
+        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 = None
+            count = await reconcile_stale_active_prints(printer_id=1)
+        assert count == 0
+
+    @pytest.mark.asyncio
+    async def test_disconnected_status_skips_reconciliation(self):
+        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("RUNNING", connected=False)
+            count = await reconcile_stale_active_prints(printer_id=1)
+        # Disconnected state would be making decisions against cached state —
+        # the connected-edge handler in on_printer_status_change is the only
+        # place that should drive reconciliation.
+        assert count == 0
+
+    @pytest.mark.asyncio
+    async def test_no_active_archives_returns_zero(self):
+        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("IDLE")
+            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: [])))
+                mock_session.return_value.__aenter__.return_value = session_ctx
+                count = await reconcile_stale_active_prints(printer_id=1)
+        assert count == 0
+
+    @pytest.mark.asyncio
+    async def test_stale_archive_synthesises_aborted_completion(self):
+        from backend.app.main import reconcile_stale_active_prints
+
+        stale = _archive(subtask_id="OLD_ID", filename="ghost.3mf", print_name="ghost")
+        with patch("backend.app.main.printer_manager") as mock_pm:
+            mock_pm.get_status.return_value = _state("IDLE", subtask_id="", subtask_name="")
+            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: [stale])))
+                mock_session.return_value.__aenter__.return_value = session_ctx
+                with patch("backend.app.main.on_print_complete", new=AsyncMock()) as mock_complete:
+                    count = await reconcile_stale_active_prints(printer_id=1)
+        assert count == 1
+        mock_complete.assert_awaited_once()
+        # Verify the synthesised payload shape.
+        args, kwargs = mock_complete.call_args
+        assert args[0] == 1
+        payload = args[1]
+        assert payload["status"] == "aborted"
+        assert payload["filename"] == "ghost.3mf"
+        assert payload["_reconciled"] is True
+
+    @pytest.mark.asyncio
+    async def test_non_stale_archive_does_not_synthesise(self):
+        from backend.app.main import reconcile_stale_active_prints
+
+        healthy = _archive(subtask_id="ABC123")
+        with patch("backend.app.main.printer_manager") as mock_pm:
+            mock_pm.get_status.return_value = _state("RUNNING", subtask_id="ABC123", subtask_name="ghost")
+            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: [healthy]))
+                )
+                mock_session.return_value.__aenter__.return_value = session_ctx
+                with patch("backend.app.main.on_print_complete", new=AsyncMock()) as mock_complete:
+                    count = await reconcile_stale_active_prints(printer_id=1)
+        assert count == 0
+        mock_complete.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_on_print_complete_failure_does_not_block_rest(self):
+        """An exception during one archive's synthesis must not abort
+        reconciliation for the other archives — and must not propagate to
+        the caller (the connected-edge handler is a hot path)."""
+        from backend.app.main import reconcile_stale_active_prints
+
+        a1 = _archive(subtask_id="A", filename="a.3mf")
+        a1.id = 1
+        a2 = _archive(subtask_id="B", filename="b.3mf")
+        a2.id = 2
+        with patch("backend.app.main.printer_manager") as mock_pm:
+            mock_pm.get_status.return_value = _state("IDLE")
+            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: [a1, a2])))
+                mock_session.return_value.__aenter__.return_value = session_ctx
+                # First call raises, second call must still happen.
+                mock_complete = AsyncMock(side_effect=[RuntimeError("boom"), None])
+                with patch("backend.app.main.on_print_complete", new=mock_complete):
+                    count = await reconcile_stale_active_prints(printer_id=1)
+        # Only the second archive is recorded as reconciled (first raised).
+        assert count == 1
+        assert mock_complete.await_count == 2