Do not close a queue item on a completion for another print

    on_print_complete finds the row to close by printer and status='printing'
    alone. The MQTT payload carries a subtask name but no run identifier, so
    nothing tied the event to the row: any completion delivered for a printer
    closed whichever job was printing on it. A job closed that way is marked
    completed while the printer is still working, leaves the queue for
    history, and strands the rest of its batch, because the queue correctly
    refuses to dispatch onto a busy printer.

    The handler now checks the completion against the file the row was
    dispatched with, recovered from its archive, and leaves the row alone
    when they disagree. Only a positive disagreement refuses: no archive, no
    file name or no subtask name is unverifiable rather than wrong, and
    refusing those would strand the item in 'printing' and wedge the queue --
    the failure the loose lookup was avoiding in the first place.

    This surfaced through the test suite, which could reach a real database.
    conftest built its own SQLite engine, but core/config.py snapshots
    DATABASE_URL at import time and core/database.py builds the module-level
    engine and async_session from it. Tests reaching code that opens its own
    session -- run_with_retry, which the completion path uses, takes its
    sessions from core.database and so is untouched by the widespread
    patch("backend.app.main.async_session") -- therefore talked to whatever
    database .env named: the developer's own SQLite file on a plain checkout,
    a live install with a PostgreSQL .env. DATABASE_URL is now redirected to
    a throwaway file before any app import, and the run aborts rather than
    starts if that did not take.

backend/app/main.py

@@ -8,7 +8,7 @@ import time
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
 from logging.handlers import RotatingFileHandler
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from urllib.parse import urlparse
 
 from fastapi import FastAPI
@@ -5033,6 +5033,65 @@ async def on_finish_photo_moment(printer_id: int, data: dict):
         producer_done.set()
 
 
+def _subtask_name_from_filename(filename: str) -> str:
+    """Recover the subtask name a print command would have carried for *filename*.
+
+    The dispatcher derives the printer-facing subtask name from the archive's
+    file name, so stripping the extensions back off gives the value MQTT echoes
+    on completion. Only the two extensions Bambuddy actually stores are removed,
+    and in the order they nest (``.gcode.3mf``), so a model whose own name
+    contains a dot -- ``My.Model.3mf`` -- keeps it.
+    """
+    name = PurePosixPath(filename).name
+    for suffix in (".3mf", ".gcode"):
+        if name.lower().endswith(suffix):
+            name = name[: -len(suffix)]
+    return name
+
+
+async def _completion_belongs_to_queue_item(db, item, data: dict) -> bool:
+    """Whether this completion event is plausibly about *item*'s print.
+
+    The caller finds its queue row by printer and ``status='printing'`` alone,
+    which is all a completion event gives it -- there is no run identifier in
+    the MQTT payload to match on. That makes the lookup indiscriminate: any
+    completion delivered for this printer closes whichever row happens to be
+    printing, however unrelated. Comparing the subtask name against the archive
+    the row was dispatched with costs one primary-key load and rules that out.
+
+    Deliberately permissive: it answers False only on a positive disagreement
+    between two names we actually have. A row with no archive, an archive with
+    no file name, or an event with no subtask name is unverifiable rather than
+    wrong, and refusing those would strand the item in ``printing`` and wedge
+    the printer's queue -- a worse failure than the one being prevented.
+    """
+    observed = (data.get("subtask_name") or "").strip()
+    if not observed or item.archive_id is None:
+        return True
+
+    from backend.app.models.archive import PrintArchive
+
+    archive = await db.get(PrintArchive, item.archive_id)
+    if archive is None or not archive.filename:
+        return True
+
+    expected = _subtask_name_from_filename(archive.filename)
+    if not expected or expected.casefold() == observed.casefold():
+        return True
+
+    logging.getLogger(__name__).warning(
+        "Ignoring print completion for queue item %s: it was dispatched as %r "
+        "(archive %s, %s) but the completion reports subtask %r. Leaving the item "
+        "printing rather than closing a run this event is not about.",
+        item.id,
+        expected,
+        archive.id,
+        archive.filename,
+        observed,
+    )
+    return False
+
+
 async def on_print_complete(printer_id: int, data: dict):
     """Handle print completion - update the archive status."""
     import time
@@ -5347,6 +5406,8 @@ async def on_print_complete(printer_id: int, data: dict):
                     [(i.id, i.archive_id, i.library_file_id) for i in printing_items],
                 )
             item = printing_items[0] if printing_items else None
+            if item is not None and not await _completion_belongs_to_queue_item(db, item, data):
+                return
             if item:
                 queue_status = data.get("status", "completed")
                 # MQTT sends "aborted" for cancelled prints; normalise to

backend/tests/conftest.py

@@ -18,6 +18,52 @@ import pytest
 os.environ["LOG_TO_FILE"] = "false"
 os.environ["DEBUG"] = "false"
 
+# Point the app's own engine at a throwaway database before anything reads
+# DATABASE_URL.
+#
+# The fixtures below build their own SQLite engine, but that is not the only
+# engine in play: `core/config.py` snapshots ``DATABASE_URL`` at import time and
+# `core/database.py` builds a module-level ``engine`` / ``async_session`` from
+# it. Any app code that opens its own session rather than receiving the fixture
+# one therefore talks to whatever database the developer's `.env` names. The
+# clearest example is ``run_with_retry`` (used by the print-completion path),
+# whose sessions come from ``backend.app.core.database`` — so the widespread
+# ``patch("backend.app.main.async_session")`` does not intercept them.
+#
+# Left alone that is not a hypothetical: on a plain checkout it means the suite
+# writes to the developer's real SQLite file, and with a PostgreSQL `.env` it
+# means a live install. A completion test calling ``on_print_complete(1, ...)``
+# closed a queue item belonging to an actual running print that way.
+_TEST_APP_DB_DIR = Path(tempfile.mkdtemp(prefix="bambuddy_test_appdb_"))
+APP_DATABASE_URL = f"sqlite+aiosqlite:///{_TEST_APP_DB_DIR / 'app.db'}"
+os.environ["DATABASE_URL"] = APP_DATABASE_URL
+
+
+def _cleanup_test_app_db_dir():
+    shutil.rmtree(_TEST_APP_DB_DIR, ignore_errors=True)
+
+
+atexit.register(_cleanup_test_app_db_dir)
+
+
+def _assert_disposable_database(url, source: str) -> None:
+    """Abort the run unless *url* is the throwaway database created above.
+
+    A guard rather than a comment because the failure it prevents is silent and
+    destructive: the suite would appear to pass while having mutated real print
+    history. Anything that reintroduces a real ``DATABASE_URL`` — an `.env` read
+    later in the import order, a fixture rebuilding the engine — trips this
+    instead of reaching the database.
+    """
+    database = str(getattr(url, "database", "") or "")
+    if not str(getattr(url, "drivername", "")).startswith("sqlite") or not database.startswith(str(_TEST_APP_DB_DIR)):
+        raise RuntimeError(
+            f"Refusing to run tests: {source} resolves to {url!r}, which is not the "
+            f"disposable SQLite database under {_TEST_APP_DB_DIR}. Tests must never "
+            f"open a session against a real Bambuddy database."
+        )
+
+
 from httpx import ASGITransport, AsyncClient  # noqa: E402
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine  # noqa: E402
 
@@ -25,6 +71,12 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
 from backend.app.core.config import settings  # noqa: E402
 
 settings.log_to_file = False
+if settings.database_url != APP_DATABASE_URL:
+    raise RuntimeError(
+        f"Refusing to run tests: settings.database_url is {settings.database_url!r} "
+        f"rather than the disposable test database. Something read DATABASE_URL "
+        f"before conftest could override it."
+    )
 
 # Use a temp directory for plate calibration to avoid deleting real calibration files
 _test_plate_cal_dir = Path(tempfile.mkdtemp(prefix="bambuddy_test_plate_cal_"))
@@ -39,7 +91,11 @@ def _cleanup_test_plate_cal_dir():
 
 atexit.register(_cleanup_test_plate_cal_dir)
 
-from backend.app.core.database import Base  # noqa: E402
+from backend.app.core.database import Base, engine as _app_engine  # noqa: E402
+
+# The engine is built at import time from the URL above, so this catches the
+# case where that override did not take effect for whatever reason.
+_assert_disposable_database(_app_engine.url, "backend.app.core.database.engine")
 
 # Use in-memory SQLite for tests
 TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

backend/tests/integration/test_print_queue_api.py

@@ -2177,14 +2177,20 @@ class TestAbortedStatusNormalisation:
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_on_print_complete_normalises_aborted_to_cancelled(self, queue_item_factory, db_session):
+    async def test_on_print_complete_normalises_aborted_to_cancelled(
+        self, queue_item_factory, archive_factory, db_session
+    ):
         """Verify the completion handler maps 'aborted' → 'cancelled' for queue items."""
         import asyncio
         from unittest.mock import AsyncMock, MagicMock, patch
 
-        item = await queue_item_factory(status="printing")
+        archive = await archive_factory(filename="Abort_Me.gcode.3mf")
+        item = await queue_item_factory(status="printing", archive_id=archive.id)
 
-        # Build a mock session whose execute returns our item
+        # Build a mock session whose execute returns our item. `get` has to
+        # answer with the item's real archive: the handler checks the completion
+        # is about this row's print before closing it, and an archive that names
+        # a different subtask is exactly what it refuses.
         mock_result = MagicMock()
         mock_result.scalars.return_value.all.return_value = [item]
 
@@ -2192,6 +2198,7 @@ class TestAbortedStatusNormalisation:
         mock_session.__aenter__ = AsyncMock(return_value=mock_session)
         mock_session.__aexit__ = AsyncMock(return_value=False)
         mock_session.execute = AsyncMock(return_value=mock_result)
+        mock_session.get = AsyncMock(return_value=archive)
         mock_session.commit = AsyncMock()
 
         tasks_before = set(asyncio.all_tasks())
@@ -2220,7 +2227,7 @@ class TestAbortedStatusNormalisation:
                 {
                     "status": "aborted",
                     "filename": "test.gcode",
-                    "subtask_name": "Test",
+                    "subtask_name": "Abort_Me",
                     "timelapse_was_active": False,
                 },
             )
@@ -2274,12 +2281,84 @@ class TestAbortedStatusNormalisation:
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_completed_status_passes_through_unchanged(self, queue_item_factory, db_session):
+    async def test_leaves_a_printing_item_alone_when_the_completion_is_for_another_print(
+        self, queue_item_factory, archive_factory, db_session
+    ):
+        """A completion naming a different subtask must not close this row.
+
+        The handler looks its row up by printer and ``status='printing'`` only,
+        so before this guard any completion delivered for the printer closed
+        whatever was printing on it. That is how a live 14-hour job was marked
+        completed 18 minutes in -- by an event that had nothing to do with it --
+        which also stranded the second plate of its batch, since the queue will
+        not dispatch while the printer is still running.
+        """
+        import asyncio
+        from unittest.mock import AsyncMock, MagicMock, patch
+
+        archive = await archive_factory(filename="AMS_Rack.gcode.3mf")
+        item = await queue_item_factory(status="printing", archive_id=archive.id)
+
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.all.return_value = [item]
+
+        mock_session = AsyncMock()
+        mock_session.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_session.__aexit__ = AsyncMock(return_value=False)
+        mock_session.execute = AsyncMock(return_value=mock_result)
+        mock_session.get = AsyncMock(return_value=archive)
+        mock_session.commit = AsyncMock()
+
+        tasks_before = set(asyncio.all_tasks())
+
+        with (
+            patch("backend.app.main.async_session", return_value=mock_session),
+            patch("backend.app.core.database.async_session", return_value=mock_session),
+            patch("backend.app.main.ws_manager") as mock_ws,
+            patch("backend.app.main.mqtt_relay") as mock_relay,
+            patch("backend.app.main.notification_service") as mock_notif,
+            patch("backend.app.main.smart_plug_manager") as mock_plug,
+            patch("backend.app.main.printer_manager") as mock_pm,
+        ):
+            mock_ws.send_print_complete = AsyncMock()
+            mock_ws.broadcast = AsyncMock()
+            mock_relay.on_print_complete = AsyncMock()
+            mock_relay.on_queue_job_completed = AsyncMock()
+            mock_notif.on_print_complete = AsyncMock()
+            mock_plug.on_print_complete = AsyncMock()
+            mock_pm.get_printer.return_value = None
+
+            from backend.app.main import on_print_complete
+
+            await on_print_complete(
+                item.printer_id,
+                {
+                    "status": "completed",
+                    "filename": "/data/Metadata/plate_1.gcode",
+                    "subtask_name": "Some_Other_Print",
+                    "timelapse_was_active": False,
+                },
+            )
+
+            for task in asyncio.all_tasks() - tasks_before:
+                task.cancel()
+                try:
+                    await task
+                except (asyncio.CancelledError, Exception):
+                    pass
+
+        assert item.status == "printing"
+        assert item.completed_at is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_completed_status_passes_through_unchanged(self, queue_item_factory, archive_factory, db_session):
         """Verify normal statuses like 'completed' are not affected by normalisation."""
         import asyncio
         from unittest.mock import AsyncMock, MagicMock, patch
 
-        item = await queue_item_factory(status="printing")
+        archive = await archive_factory(filename="Finish_Me.gcode.3mf")
+        item = await queue_item_factory(status="printing", archive_id=archive.id)
 
         mock_result = MagicMock()
         mock_result.scalars.return_value.all.return_value = [item]
@@ -2288,6 +2367,7 @@ class TestAbortedStatusNormalisation:
         mock_session.__aenter__ = AsyncMock(return_value=mock_session)
         mock_session.__aexit__ = AsyncMock(return_value=False)
         mock_session.execute = AsyncMock(return_value=mock_result)
+        mock_session.get = AsyncMock(return_value=archive)
         mock_session.commit = AsyncMock()
 
         tasks_before = set(asyncio.all_tasks())
@@ -2316,7 +2396,7 @@ class TestAbortedStatusNormalisation:
                 {
                     "status": "completed",
                     "filename": "test.gcode",
-                    "subtask_name": "Test",
+                    "subtask_name": "Finish_Me",
                     "timelapse_was_active": False,
                 },
             )

backend/tests/unit/test_completion_queue_item_match.py

@@ -0,0 +1,153 @@
+"""A print completion must not close a queue row belonging to another print.
+
+``on_print_complete`` finds its queue row by printer and ``status='printing'``
+alone -- the MQTT payload carries no run identifier to match on -- so any
+completion delivered for a printer closes whichever row happens to be printing.
+That is fine while the only source of completions is the printer itself, and
+wrong the moment one arrives from anywhere else: a live 14-hour print was closed
+18 minutes in, and its plate 2 never dispatched, because a completion for an
+unrelated subtask reached the same lookup.
+
+These cover the guard that rules that out, and the deliberate decision to let
+the unverifiable cases through rather than strand an item in ``printing``.
+"""
+
+import pytest
+
+from backend.app.main import _completion_belongs_to_queue_item, _subtask_name_from_filename
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_queue import PrintQueueItem
+
+
+class TestSubtaskNameFromFilename:
+    """The dispatcher builds the subtask name off the archive file name, so
+    stripping the extensions back off has to land on exactly what MQTT echoes."""
+
+    @pytest.mark.parametrize(
+        ("filename", "expected"),
+        [
+            ("AMS_Rack.gcode.3mf", "AMS_Rack"),
+            ("AMS_Rack.3mf", "AMS_Rack"),
+            ("plate.gcode", "plate"),
+            # A dot in the model's own name is not an extension. Path.stem would
+            # eat it and produce "My", which matches nothing.
+            ("My.Model.3mf", "My.Model"),
+            ("My.Model.gcode.3mf", "My.Model"),
+            # Extensions are matched case-insensitively; the name is not.
+            ("Cover.GCODE.3MF", "Cover"),
+            # Only the file name matters -- archives store a path.
+            ("archive/1/20260811_112435_AMS_Rack/AMS_Rack.gcode.3mf", "AMS_Rack"),
+            # Nothing to strip.
+            ("AMS_Rack", "AMS_Rack"),
+        ],
+    )
+    def test_recovers_the_dispatched_subtask_name(self, filename, expected):
+        assert _subtask_name_from_filename(filename) == expected
+
+
+async def _seed(db, *, archive_filename: str | None) -> PrintQueueItem:
+    """A printing queue item, optionally linked to an archive."""
+    archive_id = None
+    if archive_filename is not None:
+        archive = PrintArchive(
+            printer_id=1,
+            filename=archive_filename,
+            file_path=f"archive/1/{archive_filename}",
+            file_size=1,
+            status="printing",
+        )
+        db.add(archive)
+        await db.flush()
+        archive_id = archive.id
+
+    item = PrintQueueItem(printer_id=1, status="printing", archive_id=archive_id)
+    db.add(item)
+    await db.flush()
+    return item
+
+
+@pytest.mark.asyncio
+class TestCompletionBelongsToQueueItem:
+    async def test_accepts_the_completion_for_its_own_print(self, db_session):
+        item = await _seed(db_session, archive_filename="AMS_Rack.gcode.3mf")
+
+        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "AMS_Rack"}) is True
+
+    async def test_rejects_a_completion_for_a_different_print(self, db_session):
+        # The exact shape of the incident: the row was dispatched as AMS_Rack and
+        # a completion for "Test" arrived on the same printer.
+        item = await _seed(db_session, archive_filename="AMS_Rack.gcode.3mf")
+
+        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "Test"}) is False
+
+    async def test_matches_regardless_of_case(self, db_session):
+        item = await _seed(db_session, archive_filename="AMS_Rack.gcode.3mf")
+
+        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "ams_rack"}) is True
+
+    @pytest.mark.parametrize("subtask", [None, "", "   "])
+    async def test_lets_an_unidentified_completion_through(self, db_session, subtask):
+        # No subtask name to compare means unverifiable, not wrong. Refusing here
+        # would leave the item printing forever and wedge the printer's queue,
+        # which is the failure the indiscriminate lookup existed to avoid.
+        item = await _seed(db_session, archive_filename="AMS_Rack.gcode.3mf")
+
+        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": subtask}) is True
+
+    async def test_lets_an_archiveless_item_through(self, db_session):
+        # Library-file dispatch links the archive after the fact; there is
+        # nothing to compare against yet.
+        item = await _seed(db_session, archive_filename=None)
+
+        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "Anything"}) is True
+
+    async def test_lets_an_archive_without_a_filename_through(self, db_session):
+        # `filename` is NOT NULL, but nothing stops it being empty.
+        item = await _seed(db_session, archive_filename="")
+
+        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "Anything"}) is True
+
+    async def test_lets_a_dangling_archive_reference_through(self, db_session):
+        item = await _seed(db_session, archive_filename=None)
+        item.archive_id = 999999
+
+        assert await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": "Anything"}) is True
+
+
+class TestDisposableDatabaseGuard:
+    """The suite must never be able to open a session against a real database.
+
+    ``run_with_retry`` takes its session from ``backend.app.core.database``, not
+    from the ``backend.app.main.async_session`` that most tests patch, so an
+    unmocked completion path reaches the app's module-level engine. That engine
+    is built from ``DATABASE_URL``; conftest redirects it to a throwaway SQLite
+    file and asserts the redirect took, because the alternative is a suite that
+    passes while having edited someone's live print history.
+    """
+
+    def test_the_app_engine_points_at_a_throwaway_sqlite_file(self):
+        from backend.app.core.database import engine
+        from backend.tests.conftest import _TEST_APP_DB_DIR
+
+        assert engine.url.drivername.startswith("sqlite")
+        assert str(engine.url.database).startswith(str(_TEST_APP_DB_DIR))
+
+    def test_the_guard_rejects_a_real_database(self):
+        from sqlalchemy.engine import make_url
+
+        from backend.tests.conftest import _assert_disposable_database
+
+        with pytest.raises(RuntimeError, match="Refusing to run tests"):
+            _assert_disposable_database(
+                make_url("postgresql+asyncpg://user:pw@192.168.0.2:5432/bambuddy"),
+                "test",
+            )
+
+    def test_the_guard_rejects_another_sqlite_file(self):
+        # A developer's own data/bambuddy.db is just as real as a server.
+        from sqlalchemy.engine import make_url
+
+        from backend.tests.conftest import _assert_disposable_database
+
+        with pytest.raises(RuntimeError, match="Refusing to run tests"):
+            _assert_disposable_database(make_url("sqlite+aiosqlite:///data/bambuddy.db"), "test")