Stop auto K-profile calibration leaving an archive behind

    With flow dynamics calibration on, the printer lays down a
    pressure-advance line before the print itself and announces it over MQTT
    through the same print-start event a real print uses. Bambuddy archived
    it: a row named auto_pa_line_calib_mode marked as having no 3MF, sitting
    in among the user's actual prints, with a Print Started and a Print
    Completed notification for each one.

    The printer's other internal jobs were already skipped, but only by the
    /usr/ path they carry -- bed levelling reports as
    /usr/etc/print/auto_cali_for_user.gcode. The pressure-advance line
    carries no path at all. It arrives as a bare subtask name, so a rule
    that only ever looked at the filename could not see it. Falling past the
    guard it reached the no-3MF fallback, where print_name is subtask_name
    or filename, and named the row after the calibration.

    Internal jobs are now recognised by name as well as by path, from either
    field, in one place both callbacks share. The match is exact after
    normalising away the directory, one print-file suffix and case, rather
    than a prefix or substring rule: "auto" and "calib" are ordinary words in
    a user's own filenames, and a rule loose enough to catch some unnamed
    future calibration would quietly swallow somebody's print.

    The completion is suppressed too, and that half matters more than the
    noise. With no archive to close, the completion falls into the
    no-archive notification path -- which attributes an unmatched completion
    to any queue item the printer finished in the last five minutes and
    emails its owner. This calibration runs alongside a real print, so
    silencing only the start would have told that print's owner their job
    was done, early, and again for real later. The guard sits inside the
    no-archive branch, so the plate-clear gate, the queue reconciliation and
    the SD-card cleanup all still run; only the notification is skipped.

    Skipping the run early also drops the FTP sweep that preceded the
    fallback -- six candidate names across five directories with retries,
    around a hundred connections looking for a file that cannot exist, aimed
    at a printer that is in the middle of calibrating.

    auto_cali_for_user no longer sends a Print Started notification either.
    It is the same event about the same kind of job, and the archive was
    never the only thing wrong with treating it as a user's print.

backend/app/main.py

@@ -129,6 +129,7 @@ from backend.app.services.spoolman_tracking import (
 )
 from backend.app.services.tasmota import tasmota_service
 from backend.app.utils.ams_drying import is_drying_active, temperature_alarm_suppressed
+from backend.app.utils.print_jobs import is_internal_printer_job
 
 
 # =============================================================================
@@ -3043,12 +3044,21 @@ async def on_print_start(printer_id: int, data: dict):
 
         logger.info("[CALLBACK] Print start detected - filename: %s, subtask: %s", filename, subtask_name)
 
-        # Skip calibration prints — internal printer files should not be archived
-        # Bambu calibration gcode lives under /usr/ (e.g. /usr/etc/print/auto_cali_for_user.gcode)
-        if filename and filename.startswith("/usr/"):
-            logger.info("[CALLBACK] Skipping archive — internal printer file detected: %s", filename)
-            if not notification_sent:
-                await _send_print_start_notification(printer_id, data, logger=logger)
+        # Skip the printer's own jobs — a calibration run is not a user's print.
+        # See is_internal_printer_job for what counts and why both fields are
+        # tested; the pressure-advance line reports as a subtask name with no
+        # /usr/ path, which the old prefix-only test here missed entirely.
+        #
+        # No notification either. The event describes the printer calibrating
+        # itself, so "Print started" is as wrong as the archive was, and the
+        # matching completion is suppressed in on_print_complete for the same
+        # reason.
+        if is_internal_printer_job(filename, subtask_name):
+            logger.info(
+                "[CALLBACK] Skipping archive — internal printer job detected: filename=%s, subtask=%s",
+                filename,
+                subtask_name,
+            )
             return
 
         if not filename and not subtask_name:
@@ -5875,6 +5885,23 @@ async def on_print_complete(printer_id: int, data: dict):
     log_timing("Filament usage tracking")
 
     if not archive_id:
+        # The printer's own calibration run has no archive by design, so this
+        # arrives here every time one finishes. Returning before the no-archive
+        # notification is not just noise control: that path attributes an
+        # unmatched completion to any queue item this printer finished in the
+        # last five minutes, which for a calibration that runs alongside a real
+        # print means emailing its owner that their print is done, twice and
+        # early. Everything above this point has already run — the plate-clear
+        # gate, the queue reconciliation, the SD-card cleanup — so only the
+        # notification is skipped.
+        if is_internal_printer_job(filename, subtask_name):
+            logger.info(
+                "[CALLBACK] Internal printer job completed, no notification: filename=%s, subtask=%s",
+                filename,
+                subtask_name,
+            )
+            return
+
         logger.warning("Could not find archive for print complete: filename=%s, subtask=%s", filename, subtask_name)
 
         # Still send print-complete/failed/stopped notifications even without an archive.

backend/app/utils/print_jobs.py

@@ -0,0 +1,72 @@
+"""Telling the printer's own internal jobs apart from a user's print.
+
+Bambu firmware runs jobs on its own behalf -- bed levelling, vibration
+compensation, the pressure-advance line it lays down before a print when flow
+dynamics calibration is on -- and reports them over MQTT through exactly the
+same print-start and print-complete events a real print uses. Nothing about the
+event says "this one is mine": Bambuddy has to recognise the job by name.
+
+Getting that wrong is not free. An unrecognised calibration run has no 3MF
+anywhere on the printer, so the archive path sweeps FTP for a file that cannot
+exist -- six candidate names across five directories, with retries -- and then
+writes a no-3MF archive named after the calibration, on a printer that is in
+the middle of calibrating.
+
+Kept as a leaf module with no imports of its own so both the print-start and
+print-complete callbacks can share one answer.
+"""
+
+# Job names the printer runs for itself. Matched exactly (after normalising),
+# not by prefix or substring: "auto" and "calib" are ordinary words in a user's
+# own filenames, and a rule loose enough to catch an unnamed future calibration
+# would silently swallow somebody's print.
+#
+# ``auto_cali_for_user`` is the bed-levelling / vibration run, normally reported
+# with a ``/usr/etc/print/`` path that the rule below catches on its own; it is
+# listed anyway because the path is not guaranteed and a name-only report of it
+# would otherwise slip through.
+#
+# ``auto_pa_line_calib_mode`` is the pressure-advance (K profile) line. This one
+# is reported as a *subtask name* with no ``/usr/`` path at all, which is why
+# the path rule alone was never enough.
+INTERNAL_JOB_NAMES = frozenset(
+    {
+        "auto_cali_for_user",
+        "auto_pa_line_calib_mode",
+    }
+)
+
+# Longest first: ``.gcode.3mf`` has to be stripped whole, or ``.3mf`` would
+# match first and leave a trailing ``.gcode`` behind.
+_PRINT_SUFFIXES = (".gcode.3mf", ".gcode", ".3mf")
+
+
+def _normalize_job_name(value: str) -> str:
+    """Reduce a reported name to something comparable against the set above.
+
+    Drops any directory part, one print-file suffix, and case. The printer is
+    not consistent about which of these it includes -- the same calibration
+    reports as a bare name in ``subtask_name`` and, when it appears at all, as
+    a full path in ``gcode_file``.
+    """
+    name = value.rsplit("/", 1)[-1].rsplit("\\", 1)[-1].strip().casefold()
+    for suffix in _PRINT_SUFFIXES:
+        if name.endswith(suffix):
+            return name[: -len(suffix)]
+    return name
+
+
+def is_internal_printer_job(filename: str | None, subtask_name: str | None = None) -> bool:
+    """True when this print event belongs to the printer, not to a user.
+
+    Both fields are tested because neither is reliably populated: the
+    pressure-advance line arrives as a subtask name with no filename, while the
+    levelling run arrives as a ``/usr/etc/print/...`` path. A job is internal if
+    *either* field says so.
+    """
+    if filename and filename.startswith("/usr/"):
+        # Bambu keeps its own calibration gcode on the read-only system
+        # partition. Nothing a user can print ever lives there, so the whole
+        # prefix is safe to treat as internal without naming each file.
+        return True
+    return any(_normalize_job_name(value) in INTERNAL_JOB_NAMES for value in (filename, subtask_name) if value)

backend/tests/unit/test_archive_filtering.py

@@ -59,13 +59,16 @@ class TestCalibrationPrintFiltering:
                     },
                 )
 
-                # Notification should still be sent
-                mock_notif_send.assert_called_once()
+                # And no notification. The event describes the printer
+                # calibrating itself, so "Print started: auto_cali_for_user" is
+                # as wrong as the archive would have been. This assertion used
+                # to demand the opposite.
+                mock_notif_send.assert_not_called()
 
         # Verify the skip was logged
         info_messages = [r.message for r in capture_logs.records if r.levelno >= 20]
-        skip_msgs = [m for m in info_messages if "internal printer file" in str(m)]
-        assert skip_msgs, "Should log that internal printer file was skipped"
+        skip_msgs = [m for m in info_messages if "internal printer job" in str(m)]
+        assert skip_msgs, "Should log that an internal printer job was skipped"
 
     @pytest.mark.asyncio
     async def test_usr_prefix_various_paths(self, capture_logs):
@@ -108,7 +111,7 @@ class TestCalibrationPrintFiltering:
 
                 await on_print_start(1, {"filename": path, "subtask_name": "test"})
 
-            skip_msgs = [r for r in capture_logs.records if "internal printer file" in str(r.message)]
+            skip_msgs = [r for r in capture_logs.records if "internal printer job" in str(r.message)]
             assert skip_msgs, f"Path {path} should be skipped"
             capture_logs.clear()
 

backend/tests/unit/test_internal_printer_jobs.py

@@ -0,0 +1,212 @@
+"""The printer's own calibration runs leave no archive and send no notification.
+
+Auto pressure-advance calibration -- the K-profile line the printer lays down
+before a print when flow dynamics calibration is on -- reports over MQTT through
+the same print-start event a real print uses, as the subtask name
+``auto_pa_line_calib_mode`` with no ``/usr/`` path attached. The only guard
+Bambuddy had tested ``filename.startswith("/usr/")``, so the calibration sailed
+past it, found no 3MF anywhere on the printer (there is none to find), and left
+a no-3MF archive named after itself in the user's history.
+
+The same name is already known to the completion guard: #2829's capture of
+queue item 649 has ``auto_pa_line_calib_mode`` arriving as the subtask name of a
+completion that had to be refused against a running job.
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.utils.print_jobs import is_internal_printer_job
+
+
+class TestTheCalibrationIsRecognised:
+    def test_the_pressure_advance_line_by_subtask_name(self):
+        """How it actually arrives: a bare subtask name, no filename at all."""
+        assert is_internal_printer_job("", "auto_pa_line_calib_mode")
+
+    def test_the_pressure_advance_line_by_filename(self):
+        """Both fields are tested, because which one carries it is not fixed."""
+        assert is_internal_printer_job("auto_pa_line_calib_mode", None)
+
+    def test_the_levelling_run_by_its_system_path(self):
+        assert is_internal_printer_job("/usr/etc/print/auto_cali_for_user.gcode", "auto_cali_for_user")
+
+    def test_the_levelling_run_by_name_alone(self):
+        """The /usr/ path is not guaranteed, so the name is listed too."""
+        assert is_internal_printer_job(None, "auto_cali_for_user")
+
+    @pytest.mark.parametrize(
+        "reported",
+        [
+            "auto_pa_line_calib_mode",
+            "auto_pa_line_calib_mode.gcode",
+            "auto_pa_line_calib_mode.3mf",
+            "auto_pa_line_calib_mode.gcode.3mf",
+            "AUTO_PA_LINE_CALIB_MODE",
+            "/data/auto_pa_line_calib_mode.gcode.3mf",
+        ],
+    )
+    def test_however_the_name_is_dressed_up(self, reported):
+        """Path, suffix and case all vary between the fields and firmwares."""
+        assert is_internal_printer_job(reported, None)
+
+    def test_anything_under_usr_counts(self):
+        """Nothing a user can print lives on the read-only system partition."""
+        assert is_internal_printer_job("/usr/bin/firmware_test.gcode", "test")
+
+
+class TestItLeavesRealPrintsAlone:
+    """The failure that matters: swallowing somebody's actual print."""
+
+    def test_an_ordinary_print(self):
+        assert not is_internal_printer_job("Benchy.gcode.3mf", "Benchy")
+
+    def test_nothing_reported_at_all(self):
+        assert not is_internal_printer_job(None, None)
+        assert not is_internal_printer_job("", "")
+
+    @pytest.mark.parametrize(
+        "reported",
+        [
+            "auto_pa_line_calib_mode_v2.3mf",
+            "my_auto_pa_line_calib_mode.3mf",
+            "auto_cali_for_user_test.gcode.3mf",
+        ],
+    )
+    def test_a_users_file_that_merely_contains_the_name(self, reported):
+        """Exact match after normalising, so no prefix or substring rule can
+        eat a file somebody deliberately named after the calibration."""
+        assert not is_internal_printer_job(reported, None)
+
+    def test_a_calibration_cube(self):
+        """The obvious false positive for any rule built on the word 'calib'."""
+        assert not is_internal_printer_job("Calibration_Cube.gcode.3mf", "Calibration Cube")
+
+
+def _mocked_print_start():
+    """Patch set for driving on_print_start without a printer or database."""
+    return (
+        patch("backend.app.main.async_session"),
+        patch("backend.app.main.notification_service"),
+        patch("backend.app.main.smart_plug_manager"),
+        patch("backend.app.main.ws_manager"),
+        patch("backend.app.main.printer_manager"),
+        patch("backend.app.main.mqtt_relay"),
+    )
+
+
+class TestPrintStartSkipsTheCalibration:
+    @pytest.mark.asyncio
+    async def test_no_archive_and_no_notification(self, capture_logs):
+        sess, notif, plug, ws, pm, relay = _mocked_print_start()
+        with sess as mock_session_maker, notif as mock_notif, plug as mock_plug, ws as mock_ws, pm as mock_pm, relay:
+            mock_notif.on_print_start = AsyncMock()
+            mock_plug.on_print_start = AsyncMock()
+            mock_ws.send_print_start = AsyncMock()
+            mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
+
+            mock_printer = MagicMock()
+            mock_printer.auto_archive = True
+            mock_printer.id = 1
+
+            mock_session = AsyncMock()
+            mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+            mock_session.__aexit__ = AsyncMock()
+            mock_session.execute = AsyncMock(
+                return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=mock_printer))
+            )
+            mock_session_maker.return_value = mock_session
+
+            with patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock) as mock_notify:
+                from backend.app.main import on_print_start
+
+                # No filename: exactly what the printer reports for this run,
+                # and the reason the old /usr/ prefix test never fired.
+                await on_print_start(1, {"filename": "", "subtask_name": "auto_pa_line_calib_mode"})
+
+                mock_notify.assert_not_called()
+
+        skipped = [r for r in capture_logs.records if "internal printer job" in str(r.message)]
+        assert skipped, "Should log that the calibration run was skipped"
+
+
+class TestPrintCompleteStaysQuiet:
+    @pytest.mark.asyncio
+    async def test_no_orphan_notification_when_the_calibration_finishes(self):
+        """With no archive to close, the completion would otherwise fall into
+        the no-archive notification path -- which attributes an unmatched
+        completion to any queue item this printer finished in the last five
+        minutes. For a calibration running alongside a real print that means
+        telling its owner their print is done, early and twice.
+        """
+        with (
+            patch("backend.app.main.async_session") as mock_session_maker,
+            patch("backend.app.main.ws_manager") as mock_ws,
+            patch("backend.app.main.printer_manager") as mock_pm,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.spawn_background_task") as mock_spawn,
+            patch("backend.app.main.clear_3mf_cache"),
+        ):
+            mock_ws.send_print_complete = AsyncMock()
+            mock_relay.on_print_complete = AsyncMock()
+            mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
+            mock_pm.get_current_print_user = MagicMock(return_value=None)
+            mock_pm.clear_current_print_user = MagicMock()
+            mock_pm.set_awaiting_plate_clear = MagicMock()
+
+            mock_session = AsyncMock()
+            mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+            mock_session.__aexit__ = AsyncMock()
+            mock_session.execute = AsyncMock(
+                return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=None), scalars=MagicMock())
+            )
+            mock_session_maker.return_value = mock_session
+
+            from backend.app.main import on_print_complete
+
+            await on_print_complete(
+                1,
+                {"filename": "", "subtask_name": "auto_pa_line_calib_mode", "status": "completed"},
+            )
+
+            spawned = [c for c in mock_spawn.call_args_list if "notify-no-archive" in str(c)]
+            assert not spawned, "No completion notification should be spawned for a calibration run"
+
+    @pytest.mark.asyncio
+    async def test_a_real_orphan_print_still_notifies(self):
+        """The no-archive path exists for prints started outside Bambuddy. The
+        guard must not take those down with it.
+        """
+        with (
+            patch("backend.app.main.async_session") as mock_session_maker,
+            patch("backend.app.main.ws_manager") as mock_ws,
+            patch("backend.app.main.printer_manager") as mock_pm,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.spawn_background_task") as mock_spawn,
+            patch("backend.app.main.clear_3mf_cache"),
+        ):
+            mock_ws.send_print_complete = AsyncMock()
+            mock_relay.on_print_complete = AsyncMock()
+            mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
+            mock_pm.get_current_print_user = MagicMock(return_value=None)
+            mock_pm.clear_current_print_user = MagicMock()
+            mock_pm.set_awaiting_plate_clear = MagicMock()
+
+            mock_session = AsyncMock()
+            mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+            mock_session.__aexit__ = AsyncMock()
+            mock_session.execute = AsyncMock(
+                return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=None), scalars=MagicMock())
+            )
+            mock_session_maker.return_value = mock_session
+
+            from backend.app.main import on_print_complete
+
+            await on_print_complete(
+                1,
+                {"filename": "", "subtask_name": "Benchy", "status": "completed"},
+            )
+
+            spawned = [c for c in mock_spawn.call_args_list if "notify-no-archive" in str(c)]
+            assert spawned, "An unmatched real print must still notify"