fix: add logging and harden archive matching for phantom print investigation (#374)

Suppress SQL/aiosqlite debug noise (~90% log volume reduction), add
caller-traced PRINT COMMAND logging to start_print(), log scheduler
queue checks, tighten stale archive ilike match to exact match, and
warn on multiple queue items in "printing" status. Includes 18 new
unit tests.

CHANGELOG.md

@@ -31,8 +31,10 @@ All notable changes to Bambuddy will be documented in this file.
 - **Archive List View Not Labeling Failed Prints** ([#365](https://github.com/maziggy/bambuddy/issues/365)) — The archive grid view displayed a red "Failed" / "Cancelled" badge on failed and aborted prints, but the list view had no equivalent indicator. Now shows an inline status badge next to the print name in list view.
 - **Reprint Fails with SD Card Error for Archives Without 3MF File** ([#376](https://github.com/maziggy/bambuddy/issues/376)) — When a print was sent from an external slicer and Bambuddy couldn't download the 3MF from the printer during auto-archiving, the fallback archive had no file. Attempting to reprint such an archive tried to upload the data directory as a file, causing a confusing "SD card error." The backend now returns a clear error for file-less archives, and the frontend disables Print/Schedule/Open in Slicer buttons with a tooltip explaining that the 3MF file is unavailable.
 - **Inventory Spool Weight Resets After Print Completes** — After a print, the usage tracker correctly updated `weight_used` (e.g., +1.6g), but periodic AMS status updates recalculated `weight_used` from the AMS remain% sensor and overwrote the precise value. For small prints on large spools (e.g., 1.6g on 1000g), the AMS remain% stays at 100% (integer resolution = 10g steps), resetting `weight_used` back to 0. The AMS weight sync now only increases `weight_used`, never decreases it, preserving precise values from the usage tracker.
+- **Loose Archive Name Matching Could Cause Wrong Archive Reuse** ([#374](https://github.com/maziggy/bambuddy/issues/374)) — The `on_print_start` callback used `ilike('%{name}%')` to find existing "printing" archives, which meant a print named "Clip" could incorrectly match "Cable Clip" or "Clip Stand". This could cause a new print to reuse the wrong archive or skip creating one. Tightened to exact `print_name` match or exact filename variants (`.3mf`, `.gcode.3mf`).
 
 ### Improved
+- **Phantom Print Investigation — Logging & Hardening** ([#374](https://github.com/maziggy/bambuddy/issues/374)) — Added targeted logging and hardening to help diagnose reports of prints starting automatically without user input. Debug log volume reduced ~90% by suppressing `sqlalchemy.engine` (changed from INFO to WARNING) and `aiosqlite` (new WARNING suppression) noise that previously filled 2.5MB in 16 minutes. Every `start_print()` call now logs a `PRINT COMMAND` trace with the caller's file, line, and function name. The print scheduler logs pending queue items when found. `on_print_complete` warns when multiple queue items are in "printing" status for the same printer, which signals a state inconsistency.
 - **Reduce Log Noise from MQTT Diagnostics** ([#365](https://github.com/maziggy/bambuddy/issues/365)) — Downgraded 58 high-frequency MQTT diagnostic messages from INFO to DEBUG level. Payload dumps, detector state changes, field discovery logs, H2D disambiguation, and periodic status updates no longer flood the log at the default INFO level. Also suppresses paho-mqtt library INFO messages in production. User-initiated actions (print start/stop, AMS load/unload, calibration) remain at INFO. All diagnostic detail is still available when debug logging is enabled.
 - **SQLite WAL Mode for Database Reliability** — Database now uses Write-Ahead Logging (WAL) mode with a 5-second busy timeout, reducing "database is locked" errors under concurrent access. WAL mode allows simultaneous reads during writes, improving responsiveness for multi-printer setups. Automatically enabled on startup.
 - **External Camera Not Used for Snapshot + Stream Dropping** ([#325](https://github.com/maziggy/bambuddy/issues/325)) — The snapshot endpoint (`/camera/snapshot`) always used the internal printer camera even when an external camera was configured. Now checks for external camera first, matching the existing stream endpoint behavior. Also fixed external MJPEG and RTSP streams silently dropping every ~60 seconds due to missing reconnect logic — the underlying stream generators exit on read timeout, and the caller now retries up to 3 times with a 2-second delay instead of ending the stream.

backend/app/api/routes/support.py

@@ -103,7 +103,8 @@ def _apply_log_level(debug: bool):
 
     # Also adjust third-party loggers
     if debug:
-        logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)
+        logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
+        logging.getLogger("aiosqlite").setLevel(logging.WARNING)
         logging.getLogger("httpcore").setLevel(logging.DEBUG)
         logging.getLogger("httpx").setLevel(logging.DEBUG)
         logging.getLogger("paho.mqtt").setLevel(logging.DEBUG)

backend/app/main.py

@@ -1274,7 +1274,17 @@ async def on_print_start(printer_id: int, data: dict):
             select(PrintArchive)
             .where(PrintArchive.printer_id == printer_id)
             .where(PrintArchive.status == "printing")
-            .where(PrintArchive.print_name.ilike(f"%{check_name}%"))
+            .where(
+                or_(
+                    PrintArchive.print_name == check_name,
+                    PrintArchive.filename.in_(
+                        [
+                            f"{check_name}.3mf",
+                            f"{check_name}.gcode.3mf",
+                        ]
+                    ),
+                )
+            )
             .order_by(PrintArchive.created_at.desc())
             .limit(1)
         )
@@ -2633,7 +2643,14 @@ async def on_print_complete(printer_id: int, data: dict):
                 .where(PrintQueueItem.printer_id == printer_id)
                 .where(PrintQueueItem.status == "printing")
             )
-            queue_item = result.scalar_one_or_none()
+            printing_items = list(result.scalars().all())
+            if len(printing_items) > 1:
+                logger.warning(
+                    "BUG: Multiple queue items in 'printing' status for printer %s: %s",
+                    printer_id,
+                    [(i.id, i.archive_id, i.library_file_id) for i in printing_items],
+                )
+            queue_item = printing_items[0] if printing_items else None
             if queue_item:
                 status = data.get("status", "completed")
                 queue_item.status = status

backend/app/services/print_scheduler.py

@@ -69,6 +69,12 @@ class PrintScheduler:
             if not items:
                 return
 
+            logger.info(
+                "Queue check: found %d pending items: %s",
+                len(items),
+                [(i.id, i.printer_id, i.archive_id, i.library_file_id) for i in items],
+            )
+
             # Track busy printers to avoid assigning multiple items to same printer
             busy_printers: set[int] = set()
 

backend/app/services/printer_manager.py

@@ -1,4 +1,6 @@
 import asyncio
+import logging
+import traceback
 from collections.abc import Callable
 
 from sqlalchemy import select
@@ -7,6 +9,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from backend.app.models.printer import Printer
 from backend.app.services.bambu_mqtt import BambuMQTTClient, MQTTLogEntry, PrinterState, get_stage_name
 
+logger = logging.getLogger(__name__)
+
 # Models that have a real chamber temperature sensor
 # Based on Home Assistant Bambu Lab integration
 # P1P/P1S and A1/A1Mini do NOT have chamber temp sensors
@@ -304,6 +308,15 @@ class PrinterManager:
         use_ams: bool = True,
     ) -> bool:
         """Start a print on a connected printer."""
+        caller = traceback.extract_stack(limit=3)[0]
+        logger.info(
+            "PRINT COMMAND: printer=%s, file=%s, caller=%s:%s:%s",
+            printer_id,
+            filename,
+            caller.filename.split("/")[-1],
+            caller.lineno,
+            caller.name,
+        )
         if printer_id in self._clients:
             return self._clients[printer_id].start_print(
                 filename,

backend/tests/unit/services/test_printer_manager.py

@@ -3,6 +3,7 @@
 Tests printer connection management, status tracking, and print control.
 """
 
+import logging
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -382,6 +383,30 @@ class TestPrinterManager:
         result = manager.start_print(999, "test.gcode")
         assert result is False
 
+    def test_start_print_logs_print_command_with_caller(self, manager, mock_client, caplog):
+        """Verify start_print logs PRINT COMMAND with caller info (#374)."""
+        mock_client.start_print.return_value = True
+        manager._clients[1] = mock_client
+
+        with caplog.at_level(logging.INFO, logger="backend.app.services.printer_manager"):
+            manager.start_print(1, "benchy.3mf")
+
+        print_cmd_logs = [r for r in caplog.records if "PRINT COMMAND" in r.message]
+        assert len(print_cmd_logs) == 1
+        log_msg = print_cmd_logs[0].message
+        assert "printer=1" in log_msg
+        assert "file=benchy.3mf" in log_msg
+        assert "caller=" in log_msg
+
+    def test_start_print_logs_even_when_printer_unknown(self, manager, caplog):
+        """Verify PRINT COMMAND is logged even for unknown printers (#374)."""
+        with caplog.at_level(logging.INFO, logger="backend.app.services.printer_manager"):
+            result = manager.start_print(999, "ghost.3mf")
+
+        assert result is False
+        print_cmd_logs = [r for r in caplog.records if "PRINT COMMAND" in r.message]
+        assert len(print_cmd_logs) == 1
+
     # ========================================================================
     # Tests for stop_print
     # ========================================================================

backend/tests/unit/test_phantom_print_hardening.py

@@ -0,0 +1,211 @@
+"""Tests for phantom print investigation hardening (#374).
+
+Tests the tightened archive matching (no ilike) and the
+multiple-printing-items warning logic.
+
+These are pure unit tests that test the changed logic directly,
+NOT by calling the full on_print_start/on_print_complete callbacks
+(which spawn background tasks and require heavy mocking).
+"""
+
+import logging
+
+import pytest
+from sqlalchemy import or_, select
+from sqlalchemy.sql import ClauseElement
+
+from backend.app.models.archive import PrintArchive
+
+
+class TestArchiveMatchQueryShape:
+    """Tests that the archive duplicate lookup query uses exact match, not ilike (#374).
+
+    The old query used `ilike('%{name}%')` which caused "Clip" to match
+    "Cable Clip", "Clip Stand", etc. The new query uses exact print_name
+    match OR exact filename variants (.3mf, .gcode.3mf).
+    """
+
+    def _build_archive_query(self, check_name: str, printer_id: int = 1) -> ClauseElement:
+        """Build the exact query used in on_print_start for archive dedup."""
+        return (
+            select(PrintArchive)
+            .where(PrintArchive.printer_id == printer_id)
+            .where(PrintArchive.status == "printing")
+            .where(
+                or_(
+                    PrintArchive.print_name == check_name,
+                    PrintArchive.filename.in_(
+                        [
+                            f"{check_name}.3mf",
+                            f"{check_name}.gcode.3mf",
+                        ]
+                    ),
+                )
+            )
+            .order_by(PrintArchive.created_at.desc())
+            .limit(1)
+        )
+
+    def test_query_does_not_contain_ilike(self):
+        """Verify the compiled query does NOT use LIKE/ILIKE."""
+        query = self._build_archive_query("Clip")
+        query_str = str(query.compile(compile_kwargs={"literal_binds": True}))
+
+        assert "LIKE" not in query_str.upper(), f"Query should not use LIKE: {query_str}"
+
+    def test_query_uses_exact_equality(self):
+        """Verify the query uses = for print_name comparison."""
+        query = self._build_archive_query("Benchy")
+        query_str = str(query.compile(compile_kwargs={"literal_binds": True}))
+
+        assert "print_name = " in query_str or "print_name ='" in query_str or "print_name =" in query_str
+
+    def test_query_uses_in_for_filename_variants(self):
+        """Verify the query uses IN for filename matching with .3mf variants."""
+        query = self._build_archive_query("MyPrint")
+        query_str = str(query.compile(compile_kwargs={"literal_binds": True}))
+
+        assert "IN" in query_str.upper()
+        assert "MyPrint.3mf" in query_str
+        assert "MyPrint.gcode.3mf" in query_str
+
+    def test_partial_name_not_in_query(self):
+        """Verify 'Clip' does not produce a wildcard pattern."""
+        query = self._build_archive_query("Clip")
+        query_str = str(query.compile(compile_kwargs={"literal_binds": True}))
+
+        # Should NOT contain %Clip% wildcard
+        assert "%Clip%" not in query_str
+
+    def test_check_name_derivation_from_subtask(self):
+        """Verify check_name is derived correctly from subtask_name."""
+        # Simulates: check_name = subtask_name or filename.split("/")[-1].replace(...)
+        subtask_name = "Cable Clip"
+        filename = "/sdcard/Cable Clip.gcode"
+        check_name = subtask_name or filename.split("/")[-1].replace(".gcode", "").replace(".3mf", "")
+        assert check_name == "Cable Clip"
+
+        query = self._build_archive_query(check_name)
+        query_str = str(query.compile(compile_kwargs={"literal_binds": True}))
+
+        # Exact match should contain the full name, not a partial
+        assert "Cable Clip" in query_str
+        assert "%Cable Clip%" not in query_str
+
+    def test_check_name_derivation_from_filename(self):
+        """Verify check_name strips extensions correctly from filename."""
+        subtask_name = None
+        filename = "/sdcard/MyPrint.gcode"
+        check_name = subtask_name or filename.split("/")[-1].replace(".gcode", "").replace(".3mf", "")
+        assert check_name == "MyPrint"
+
+
+class TestMultiplePrintingQueueItemsWarning:
+    """Tests for the multiple-printing-items warning logic (#374).
+
+    The code in on_print_complete now detects when multiple queue items
+    are in 'printing' status for the same printer, which signals a bug.
+    """
+
+    def test_single_item_returns_item_no_warning(self, caplog):
+        """Verify single item is returned without warning."""
+        from unittest.mock import MagicMock
+
+        items = [MagicMock(id=1, archive_id=10, library_file_id=None)]
+
+        # Simulate the exact code from on_print_complete
+        with caplog.at_level(logging.WARNING, logger="backend.app.main"):
+            logger = logging.getLogger("backend.app.main")
+            printer_id = 1
+            printing_items = list(items)
+
+            if len(printing_items) > 1:
+                logger.warning(
+                    "BUG: Multiple queue items in 'printing' status for printer %s: %s",
+                    printer_id,
+                    [(i.id, i.archive_id, i.library_file_id) for i in printing_items],
+                )
+            queue_item = printing_items[0] if printing_items else None
+
+        assert queue_item is not None
+        assert queue_item.id == 1
+        bug_warnings = [r for r in caplog.records if "BUG: Multiple queue items" in r.message]
+        assert len(bug_warnings) == 0
+
+    def test_multiple_items_warns_and_returns_first(self, caplog):
+        """Verify warning is logged and first item is returned when multiple exist."""
+        from unittest.mock import MagicMock
+
+        items = [
+            MagicMock(id=1, archive_id=10, library_file_id=None),
+            MagicMock(id=2, archive_id=20, library_file_id=None),
+        ]
+
+        with caplog.at_level(logging.WARNING, logger="backend.app.main"):
+            logger = logging.getLogger("backend.app.main")
+            printer_id = 1
+            printing_items = list(items)
+
+            if len(printing_items) > 1:
+                logger.warning(
+                    "BUG: Multiple queue items in 'printing' status for printer %s: %s",
+                    printer_id,
+                    [(i.id, i.archive_id, i.library_file_id) for i in printing_items],
+                )
+            queue_item = printing_items[0] if printing_items else None
+
+        assert queue_item is not None
+        assert queue_item.id == 1  # First item is used
+        bug_warnings = [r for r in caplog.records if "BUG: Multiple queue items" in r.message]
+        assert len(bug_warnings) == 1
+        assert "printer 1" in bug_warnings[0].message
+        # Warning should include item details
+        assert "10" in bug_warnings[0].message  # archive_id of item 1
+        assert "20" in bug_warnings[0].message  # archive_id of item 2
+
+    def test_empty_list_returns_none_no_warning(self, caplog):
+        """Verify None is returned and no warning when no items exist."""
+        with caplog.at_level(logging.WARNING, logger="backend.app.main"):
+            logger = logging.getLogger("backend.app.main")
+            printer_id = 1
+            printing_items = []
+
+            if len(printing_items) > 1:
+                logger.warning(
+                    "BUG: Multiple queue items in 'printing' status for printer %s: %s",
+                    printer_id,
+                    [(i.id, i.archive_id, i.library_file_id) for i in printing_items],
+                )
+            queue_item = printing_items[0] if printing_items else None
+
+        assert queue_item is None
+        bug_warnings = [r for r in caplog.records if "BUG: Multiple queue items" in r.message]
+        assert len(bug_warnings) == 0
+
+    def test_three_items_warns_with_all_details(self, caplog):
+        """Verify warning includes all item details when three items found."""
+        from unittest.mock import MagicMock
+
+        items = [
+            MagicMock(id=1, archive_id=10, library_file_id=None),
+            MagicMock(id=2, archive_id=None, library_file_id=5),
+            MagicMock(id=3, archive_id=30, library_file_id=None),
+        ]
+
+        with caplog.at_level(logging.WARNING, logger="backend.app.main"):
+            logger = logging.getLogger("backend.app.main")
+            printer_id = 7
+            printing_items = list(items)
+
+            if len(printing_items) > 1:
+                logger.warning(
+                    "BUG: Multiple queue items in 'printing' status for printer %s: %s",
+                    printer_id,
+                    [(i.id, i.archive_id, i.library_file_id) for i in printing_items],
+                )
+            queue_item = printing_items[0] if printing_items else None
+
+        assert queue_item.id == 1
+        bug_warnings = [r for r in caplog.records if "BUG: Multiple queue items" in r.message]
+        assert len(bug_warnings) == 1
+        assert "printer 7" in bug_warnings[0].message

backend/tests/unit/test_scheduler_clear_plate.py

@@ -1,6 +1,7 @@
 """Tests for the clear plate queue flow in the print scheduler."""
 
-from unittest.mock import MagicMock, patch
+import logging
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -120,3 +121,66 @@ class TestSchedulerIdleCheckWithPlateCleared:
         mock_pm.is_connected.return_value = True
         mock_pm.get_status.return_value = None
         assert scheduler._is_printer_idle(1) is False
+
+
+class TestSchedulerQueueCheckLogging:
+    """Test queue check logging when pending items are found (#374)."""
+
+    @pytest.fixture
+    def scheduler(self):
+        return PrintScheduler()
+
+    @pytest.mark.asyncio
+    @patch("backend.app.services.print_scheduler.printer_manager")
+    async def test_check_queue_logs_pending_items(self, mock_pm, scheduler, caplog):
+        """Verify pending items are logged when found in check_queue."""
+        mock_item = MagicMock()
+        mock_item.id = 42
+        mock_item.printer_id = 1
+        mock_item.archive_id = 100
+        mock_item.library_file_id = None
+        mock_item.scheduled_time = None
+        mock_item.manual_start = False
+        mock_item.target_model = None
+
+        mock_pm.is_connected.return_value = True
+        mock_pm.get_status.return_value = MagicMock(state="RUNNING")
+
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.all.return_value = [mock_item]
+
+        with (
+            patch("backend.app.services.print_scheduler.async_session") as mock_session_ctx,
+            caplog.at_level(logging.INFO, logger="backend.app.services.print_scheduler"),
+        ):
+            mock_db = AsyncMock()
+            mock_db.execute = AsyncMock(return_value=mock_result)
+            mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            mock_session_ctx.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            await scheduler.check_queue()
+
+        queue_logs = [r for r in caplog.records if "Queue check" in r.message]
+        assert len(queue_logs) == 1
+        assert "1 pending items" in queue_logs[0].message
+        assert "42" in queue_logs[0].message  # item ID
+
+    @pytest.mark.asyncio
+    async def test_check_queue_no_log_when_empty(self, scheduler, caplog):
+        """Verify no queue log when no pending items found."""
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.all.return_value = []
+
+        with (
+            patch("backend.app.services.print_scheduler.async_session") as mock_session_ctx,
+            caplog.at_level(logging.INFO, logger="backend.app.services.print_scheduler"),
+        ):
+            mock_db = AsyncMock()
+            mock_db.execute = AsyncMock(return_value=mock_result)
+            mock_session_ctx.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+            mock_session_ctx.return_value.__aexit__ = AsyncMock(return_value=False)
+
+            await scheduler.check_queue()
+
+        queue_logs = [r for r in caplog.records if "Queue check" in r.message]
+        assert len(queue_logs) == 0

backend/tests/unit/test_support_helpers.py

@@ -12,6 +12,53 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
+class TestApplyLogLevel:
+    """Tests for _apply_log_level() debug noise suppression."""
+
+    def test_debug_mode_suppresses_sqlalchemy_to_warning(self):
+        """Verify sqlalchemy.engine is set to WARNING (not INFO) in debug mode."""
+        import logging
+
+        from backend.app.api.routes.support import _apply_log_level
+
+        _apply_log_level(True)
+
+        assert logging.getLogger("sqlalchemy.engine").level == logging.WARNING
+
+    def test_debug_mode_suppresses_aiosqlite(self):
+        """Verify aiosqlite is set to WARNING in debug mode to prevent cursor noise."""
+        import logging
+
+        from backend.app.api.routes.support import _apply_log_level
+
+        _apply_log_level(True)
+
+        assert logging.getLogger("aiosqlite").level == logging.WARNING
+
+    def test_debug_mode_enables_httpcore_debug(self):
+        """Verify httpcore stays at DEBUG in debug mode."""
+        import logging
+
+        from backend.app.api.routes.support import _apply_log_level
+
+        _apply_log_level(True)
+
+        assert logging.getLogger("httpcore").level == logging.DEBUG
+
+    def test_non_debug_mode_suppresses_all_noisy_loggers(self):
+        """Verify all noisy loggers are set to WARNING in non-debug mode."""
+        import logging
+
+        from backend.app.api.routes.support import _apply_log_level
+
+        _apply_log_level(False)
+
+        assert logging.getLogger("sqlalchemy.engine").level == logging.WARNING
+        assert logging.getLogger("httpcore").level == logging.WARNING
+        assert logging.getLogger("httpx").level == logging.WARNING
+        assert logging.getLogger("paho.mqtt").level == logging.WARNING
+
+
 class TestAnonymizeMqttBroker:
     """Tests for _anonymize_mqtt_broker()."""