Match a print completion to its queue job the way the printer names it (#2829)

    Bambuddy has no run identifier to tie a completion to a queue row, so it
    finds the row by printer and status='printing' alone. b5a34b7ba added a
    check that the completion's subtask name agrees with the file the row was
    dispatched with, so the printer's own calibration runs cannot close
    someone's job early. It compared the two names verbatim.

    The printer does not echo them verbatim. It substitutes underscores for
    spaces, so 'H2D_Carbon_Filter_(V2)_Body & Solid Lid' came back as
    'H2D_Carbon_Filter_(V2)_Body_&_Solid_Lid', the check refused it, and the
    row stayed printing. check_queue counts every printing row as a busy
    printer and nothing else ever closes one, so the printer's queue stopped
    until someone cancelled by hand. It also truncates long names and marks
    the cut with '...', which would have done the same to any long title.

    Compare on the canonical form instead -- case, spaces and underscores --
    which is the rule the 3MF lookup in this module has always used for the
    same names, and treat a truncation marker as a prefix match. The check
    keeps its purpose: the same printer the same day correctly refused a
    completion for auto_pa_line_calib_mode.

    One comparison being stricter than reality should not be able to stop a
    queue indefinitely, so the scheduler now closes a row itself when it has
    been printing for five minutes after its printer went terminal, with the
    status that state implies. A real completion arrives within seconds, so
    this only sees rows that were already stranded, and a disconnected
    printer never qualifies. It restores the queue only -- notifications,
    billing and auto-off are not replayed minutes late.

backend/app/main.py

@@ -5230,6 +5230,50 @@ def _subtask_name_from_filename(filename: str) -> str:
     return name
 
 
+# How the printer marks a subtask name it had to cut short. Observed on real
+# hardware at ~100 characters, but the cut-off is not a fixed character count
+# (a name with multibyte characters came back at 98), so match the marker
+# rather than a length.
+_SUBTASK_TRUNCATION_MARKER = "..."
+
+
+def _normalise_subtask_name(name: str) -> str:
+    """Canonical form for comparing a dispatched name against MQTT's echo.
+
+    The printer does not echo the name back verbatim: it substitutes
+    underscores for spaces. ``H2D_Carbon_Filter_(V2)_Body & Solid Lid`` is
+    dispatched and ``H2D_Carbon_Filter_(V2)_Body_&_Solid_Lid`` comes back.
+
+    The 3MF lookup in this module has always known that -- it builds
+    space-to-underscore variants of every candidate filename, and its
+    directory search normalises both sides before comparing. This exists so
+    the completion check reads the same rule from the same place instead of
+    growing its own, which is exactly how it came to disagree (#2829).
+    """
+    return name.strip().replace(" ", "_").casefold()
+
+
+def _subtask_names_match(expected: str, observed: str) -> bool:
+    """Whether two subtask names describe the same print.
+
+    Beyond the space/underscore substitution, the printer truncates long names
+    and marks the cut with ``...``. A truncated echo has to count as a match or
+    every print with a long name strands its queue item the same way.
+    """
+    expected_n = _normalise_subtask_name(expected)
+    observed_n = _normalise_subtask_name(observed)
+    if expected_n == observed_n:
+        return True
+
+    # Either side can be the truncated one: the printer truncates what it
+    # echoes, and an archive whose own filename was recorded from a previous
+    # truncated echo carries the marker too.
+    for full, cut in ((expected_n, observed_n), (observed_n, expected_n)):
+        if cut.endswith(_SUBTASK_TRUNCATION_MARKER) and full.startswith(cut[: -len(_SUBTASK_TRUNCATION_MARKER)]):
+            return True
+    return False
+
+
 async def _completion_belongs_to_queue_item(db, item, data: dict) -> bool:
     """Whether this completion event is plausibly about *item*'s print.
 
@@ -5257,7 +5301,7 @@ async def _completion_belongs_to_queue_item(db, item, data: dict) -> bool:
         return True
 
     expected = _subtask_name_from_filename(archive.filename)
-    if not expected or expected.casefold() == observed.casefold():
+    if not expected or _subtask_names_match(expected, observed):
         return True
 
     logging.getLogger(__name__).warning(

backend/app/services/print_scheduler.py

@@ -114,6 +114,39 @@ _PREHEAT_CANCEL_CHECK_SECONDS = 10.0
 _AIRDUCT_MODE_COOLING = 0
 _AIRDUCT_MODE_HEATING = 1
 
+# How long a queue row may stay 'printing' while its printer sits in a terminal
+# state before the scheduler closes it itself (#2829).
+#
+# A real completion arrives within seconds of the printer going terminal, so
+# five minutes is far outside the normal path — this only ever sees a row whose
+# completion was refused or never delivered. It is the whole cost of the
+# failure to the user, though: a stranded row blocks every later job for that
+# printer, so it should not be raised without reason.
+_STRANDED_PRINTING_GRACE_SECONDS = 300.0
+
+# gcode_state values that mean the print is over, mapped to the queue status
+# they imply. Mirrors the mapping in bambu_mqtt's completion detection
+# (FINISH -> completed, FAILED -> failed, anything else terminal -> aborted,
+# which the queue calls cancelled) so a recovered row cannot disagree with one
+# closed by the normal path.
+_TERMINAL_STATE_QUEUE_STATUS = {
+    "FINISH": "completed",
+    "FAILED": "failed",
+    "IDLE": "cancelled",
+}
+
+
+def _terminal_queue_status(state) -> str | None:
+    """Queue status implied by *state*, or None if the print is not over.
+
+    None for a disconnected printer as well as a busy one: a printer we are not
+    talking to has a stale ``state`` field that proves nothing about what it is
+    doing now.
+    """
+    if state is None or not getattr(state, "connected", False):
+        return None
+    return _TERMINAL_STATE_QUEUE_STATUS.get(getattr(state, "state", None))
+
 
 @dataclass
 class _KeepWarmEntry:
@@ -639,6 +672,12 @@ class PrintScheduler:
         # `notify_dispatch_cancelled` from the queue routes, consumed by
         # `_preheat_sleep`, and cleared when the dispatch exits.
         self._cancelled_dispatches: set[int] = set()
+        # printer_id -> monotonic time it was first seen terminal while one of
+        # its queue rows was still 'printing'. Reset by any non-terminal
+        # observation, so it measures an unbroken run rather than a total.
+        # In-memory on purpose: a restart re-arms the grace period, which only
+        # delays a recovery that is already the exceptional path (#2829).
+        self._terminal_since: dict[int, float] = {}
 
     async def run(self):
         """Main loop - check queue every interval."""
@@ -656,6 +695,7 @@ class PrintScheduler:
                 # briefly unreachable), instead of leaving the row wedged until
                 # the next restart.
                 await self._clear_stale_dispatch_claims()
+                await self._close_stranded_printing_items()
                 dispatched = await self.check_queue()
             except Exception as e:
                 logger.error("Scheduler error: %s", e)
@@ -664,6 +704,90 @@ class PrintScheduler:
             # not stall behind the idle interval; otherwise sleep normally (#2555).
             await asyncio.sleep(self._fast_check_interval if dispatched else self._check_interval)
 
+    async def _close_stranded_printing_items(self) -> None:
+        """Close a ``printing`` row the completion event never closed (#2829).
+
+        ``on_print_complete`` refuses to close a row when the completion's
+        subtask name disagrees with the file the row was dispatched with, so a
+        completion meant for something else (the printer's own
+        ``auto_pa_line_calib_mode`` run, say) cannot end someone's job early.
+        The refusal has no way back, though: nothing else ever closes the row,
+        and ``check_queue`` treats every ``printing`` row as a busy printer, so
+        one bad comparison wedges that printer's queue until a human presses
+        cancel. That is what #2829's reporters hit, and the guard's own
+        docstring already called stranding the worse of the two failures.
+
+        This is the way back. When a row has been ``printing`` while its
+        printer sat in a terminal state for the whole grace period, the print
+        is over however the event was read, and the row is closed with the
+        status the printer's own state implies.
+
+        Deliberately conservative:
+
+        * Only a connected printer counts. A disconnected one has a stale
+          ``state`` and proves nothing.
+        * The clock is reset by any non-terminal observation, so this cannot
+          fire on a printer that is merely between stages.
+        * The grace period is far longer than the gap between a printer
+          finishing and its completion arriving, so the normal path always
+          wins the race and this only ever sees genuine strandings.
+
+        What it does *not* do is replay the completion's side effects --
+        notifications, billing, auto-off. It restores the queue, which is the
+        harm being undone; the archive was updated by the normal path
+        regardless, since only the queue block refuses. A recovery that
+        silently re-fired notifications minutes late would be its own bug.
+        """
+        try:
+            async with async_session() as db:
+                result = await db.execute(
+                    select(PrintQueueItem)
+                    .where(PrintQueueItem.status == "printing")
+                    .where(PrintQueueItem.printer_id.is_not(None))
+                )
+                items = list(result.scalars().all())
+                if not items:
+                    self._terminal_since.clear()
+                    return
+
+                now = time.monotonic()
+                seen_printers: set[int] = set()
+                closed = False
+                for item in items:
+                    printer_id = item.printer_id
+                    seen_printers.add(printer_id)
+                    state = printer_manager.get_status(printer_id)
+                    status = _terminal_queue_status(state)
+                    if status is None:
+                        self._terminal_since.pop(printer_id, None)
+                        continue
+                    since = self._terminal_since.setdefault(printer_id, now)
+                    if now - since < _STRANDED_PRINTING_GRACE_SECONDS:
+                        continue
+
+                    item.status = status
+                    item.completed_at = datetime.now(timezone.utc)
+                    closed = True
+                    logger.warning(
+                        "Queue item %s was still 'printing' after printer %s reported %s for %.0fs — "
+                        "closing it as %s. Its completion event was never matched to it, which blocks "
+                        "every later job for this printer (#2829).",
+                        item.id,
+                        printer_id,
+                        getattr(state, "state", None),
+                        now - since,
+                        status,
+                    )
+                if closed:
+                    await db.commit()
+                for printer_id in list(self._terminal_since):
+                    if printer_id not in seen_printers:
+                        del self._terminal_since[printer_id]
+        except Exception as e:
+            # Best-effort, same as the claim sweep beside it: a recovery path
+            # that can itself break the scheduler loop is worse than the strand.
+            logger.error("Stranded-item sweep failed: %s", e)
+
     async def _clear_stale_dispatch_claims(self, *, at_startup: bool = False) -> None:
         """Clear dispatch claims with no live dispatch coroutine behind them (#2615).
 

backend/tests/integration/test_completion_guard_wiring_2829.py

@@ -0,0 +1,106 @@
+"""The completion guard itself, not just the name comparison (#2829).
+
+There is a unit test for ``_subtask_names_match``. It is not enough on its own:
+reverting ``_completion_belongs_to_queue_item`` to the strict equality that
+caused the bug leaves every one of those tests green, because they never touch
+the guard. So these drive the guard, with a real archive row behind a real
+queue row, on the exact strings that stranded the maintainer's H2D.
+"""
+
+import pytest
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+
+async def _archive(db_session, printer, filename):
+    from backend.app.models.archive import PrintArchive
+
+    archive = PrintArchive(
+        printer_id=printer.id,
+        filename=filename,
+        file_path=f"archives/{filename}",
+        file_size=1024,
+        status="printing",
+    )
+    db_session.add(archive)
+    await db_session.commit()
+    await db_session.refresh(archive)
+    return archive
+
+
+async def _item(db_session, printer, archive):
+    from backend.app.models.print_queue import PrintQueueItem
+
+    item = PrintQueueItem(printer_id=printer.id, status="printing", archive_id=archive.id)
+    db_session.add(item)
+    await db_session.commit()
+    await db_session.refresh(item)
+    return item
+
+
+async def _belongs(db_session, item, subtask_name):
+    from backend.app.main import _completion_belongs_to_queue_item
+
+    return await _completion_belongs_to_queue_item(db_session, item, {"subtask_name": subtask_name})
+
+
+class TestTheReportedStranding:
+    async def test_the_completion_for_its_own_print_is_accepted(self, db_session, printer_factory):
+        """Queue item 649 on the maintainer's H2D, verbatim. The printer echoes
+        the name back with underscores where the file has spaces; the guard
+        read that as a different print and left the row printing forever."""
+        printer = await printer_factory()
+        archive = await _archive(db_session, printer, "H2D_Carbon_Filter_(V2)_Body & Solid Lid.gcode.3mf")
+        item = await _item(db_session, printer, archive)
+
+        assert await _belongs(db_session, item, "H2D_Carbon_Filter_(V2)_Body_&_Solid_Lid")
+
+    async def test_a_truncated_echo_is_accepted(self, db_session, printer_factory):
+        printer = await printer_factory()
+        name = "169356_204314.STEP + 169356_204314.STEP + 169356_204314.STEP + 169356_204314.STEP + 169356_204314"
+        archive = await _archive(db_session, printer, f"{name}.gcode.3mf")
+        item = await _item(db_session, printer, archive)
+
+        assert await _belongs(db_session, item, f"{name[:70]}...")
+
+
+class TestItStillRefuses:
+    """The guard's reason for existing: a completion for something else must
+    not close a job that is still running."""
+
+    async def test_the_printers_own_calibration_run(self, db_session, printer_factory):
+        printer = await printer_factory()
+        archive = await _archive(db_session, printer, "H2D_Carbon_Filter_(V2)_Body & Solid Lid.gcode.3mf")
+        item = await _item(db_session, printer, archive)
+
+        assert not await _belongs(db_session, item, "auto_pa_line_calib_mode")
+
+    async def test_an_unrelated_print(self, db_session, printer_factory):
+        printer = await printer_factory()
+        archive = await _archive(db_session, printer, "Benchy.gcode.3mf")
+        item = await _item(db_session, printer, archive)
+
+        assert not await _belongs(db_session, item, "Calibration Cube")
+
+
+class TestUnverifiableIsNotWrong:
+    """Refusing what cannot be checked would strand the queue, which is the
+    worse of the two failures and the one this issue is about."""
+
+    async def test_no_subtask_name_in_the_event(self, db_session, printer_factory):
+        printer = await printer_factory()
+        archive = await _archive(db_session, printer, "Benchy.gcode.3mf")
+        item = await _item(db_session, printer, archive)
+
+        assert await _belongs(db_session, item, "")
+
+    async def test_a_row_with_no_archive(self, db_session, printer_factory):
+        from backend.app.models.print_queue import PrintQueueItem
+
+        printer = await printer_factory()
+        item = PrintQueueItem(printer_id=printer.id, status="printing")
+        db_session.add(item)
+        await db_session.commit()
+        await db_session.refresh(item)
+
+        assert await _belongs(db_session, item, "Anything At All")

backend/tests/integration/test_stranded_printing_recovery_2829.py

@@ -0,0 +1,228 @@
+"""A queue row that was never closed must not block the printer forever (#2829).
+
+``on_print_complete`` refuses to close a row when the completion's subtask name
+disagrees with the file it was dispatched with, so another print's completion
+cannot end someone's job early. Nothing took the refusal back, though, and
+``check_queue`` counts every ``printing`` row as a busy printer -- so one bad
+comparison wedged that printer's queue until a human pressed cancel.
+
+The name comparison is fixed separately; this is the net under it, for the next
+name format nobody predicted.
+"""
+
+import types
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.services.print_scheduler import _STRANDED_PRINTING_GRACE_SECONDS, _terminal_queue_status
+
+pytestmark = pytest.mark.integration
+
+
+def _state(state="FINISH", connected=True):
+    return types.SimpleNamespace(state=state, connected=connected)
+
+
+async def _noop():
+    return None
+
+
+async def _noop_arg(*_args, **_kwargs):
+    return None
+
+
+class TestTerminalStatusMapping:
+    @pytest.mark.parametrize(
+        "printer_state,expected",
+        [("FINISH", "completed"), ("FAILED", "failed"), ("IDLE", "cancelled")],
+    )
+    def test_terminal_states_imply_a_queue_status(self, printer_state, expected):
+        """Mirrors what the MQTT completion path would have set, so a recovered
+        row cannot disagree with one closed normally."""
+        assert _terminal_queue_status(_state(printer_state)) == expected
+
+    @pytest.mark.parametrize("printer_state", ["RUNNING", "PREPARE", "PAUSE", "SLICING", None])
+    def test_a_busy_printer_implies_nothing(self, printer_state):
+        assert _terminal_queue_status(_state(printer_state)) is None
+
+    def test_a_disconnected_printer_implies_nothing(self):
+        """Its `state` is whatever we last heard, which proves nothing about
+        what the printer is doing now -- closing on it would be a guess."""
+        assert _terminal_queue_status(_state("FINISH", connected=False)) is None
+
+    def test_no_state_at_all_implies_nothing(self):
+        assert _terminal_queue_status(None) is None
+
+
+@pytest.mark.asyncio
+class TestTheSweep:
+    """Drives the real sweep against a real database."""
+
+    @pytest.fixture
+    def scheduler(self, test_engine):
+        """The sweep opens its own session from the scheduler module, which the
+        widespread `patch("backend.app.main.async_session")` does not reach --
+        the same trap b5a34b7ba's own commit message describes. Patch it at the
+        module, as the other scheduler integration tests do."""
+        import backend.app.services.print_scheduler as scheduler_module
+        from backend.app.services.print_scheduler import PrintScheduler
+
+        maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+        with patch.object(scheduler_module, "async_session", maker):
+            yield PrintScheduler()
+
+    async def _item(self, db_session, printer, status="printing"):
+        from backend.app.models.print_queue import PrintQueueItem
+
+        item = PrintQueueItem(printer_id=printer.id, status=status)
+        db_session.add(item)
+        await db_session.commit()
+        await db_session.refresh(item)
+        return item
+
+    async def _status_of(self, db_session, item_id):
+        from backend.app.models.print_queue import PrintQueueItem
+
+        db_session.expire_all()
+        return (await db_session.get(PrintQueueItem, item_id)).status
+
+    async def test_a_row_is_left_alone_inside_the_grace_period(
+        self, scheduler, db_session, printer_factory, monkeypatch
+    ):
+        """A real completion arrives seconds after the printer goes terminal.
+        Closing early would race the normal path and beat it to the row."""
+        printer = await printer_factory()
+        item = await self._item(db_session, printer)
+        monkeypatch.setattr(
+            "backend.app.services.print_scheduler.printer_manager.get_status", lambda _pid: _state("FINISH")
+        )
+
+        await scheduler._close_stranded_printing_items()
+
+        assert await self._status_of(db_session, item.id) == "printing"
+
+    async def test_a_row_is_closed_once_the_grace_period_passes(
+        self, scheduler, db_session, printer_factory, monkeypatch
+    ):
+        printer = await printer_factory()
+        item = await self._item(db_session, printer)
+        monkeypatch.setattr(
+            "backend.app.services.print_scheduler.printer_manager.get_status", lambda _pid: _state("FINISH")
+        )
+
+        await scheduler._close_stranded_printing_items()
+        # Age the clock rather than sleeping five minutes.
+        scheduler._terminal_since[printer.id] -= _STRANDED_PRINTING_GRACE_SECONDS + 1
+        await scheduler._close_stranded_printing_items()
+
+        assert await self._status_of(db_session, item.id) == "completed"
+
+    async def test_the_clock_restarts_when_the_printer_goes_busy_again(
+        self, scheduler, db_session, printer_factory, monkeypatch
+    ):
+        """The grace period has to measure one unbroken terminal run. A printer
+        that finished, started something else, and finished again must not have
+        the two stretches added together."""
+        printer = await printer_factory()
+        item = await self._item(db_session, printer)
+        state = _state("FINISH")
+        monkeypatch.setattr("backend.app.services.print_scheduler.printer_manager.get_status", lambda _pid: state)
+
+        await scheduler._close_stranded_printing_items()
+        scheduler._terminal_since[printer.id] -= _STRANDED_PRINTING_GRACE_SECONDS + 1
+
+        state.state = "RUNNING"
+        await scheduler._close_stranded_printing_items()
+        assert printer.id not in scheduler._terminal_since
+
+        state.state = "FINISH"
+        await scheduler._close_stranded_printing_items()
+
+        assert await self._status_of(db_session, item.id) == "printing"
+
+    async def test_a_disconnected_printer_is_never_closed_on(self, scheduler, db_session, printer_factory, monkeypatch):
+        printer = await printer_factory()
+        item = await self._item(db_session, printer)
+        monkeypatch.setattr(
+            "backend.app.services.print_scheduler.printer_manager.get_status",
+            lambda _pid: _state("FINISH", connected=False),
+        )
+
+        await scheduler._close_stranded_printing_items()
+        scheduler._terminal_since[printer.id] = 0.0  # as if it had been ages
+        await scheduler._close_stranded_printing_items()
+
+        assert await self._status_of(db_session, item.id) == "printing"
+
+    async def test_the_failure_status_is_carried_over(self, scheduler, db_session, printer_factory, monkeypatch):
+        printer = await printer_factory()
+        item = await self._item(db_session, printer)
+        monkeypatch.setattr(
+            "backend.app.services.print_scheduler.printer_manager.get_status", lambda _pid: _state("FAILED")
+        )
+
+        await scheduler._close_stranded_printing_items()
+        scheduler._terminal_since[printer.id] -= _STRANDED_PRINTING_GRACE_SECONDS + 1
+        await scheduler._close_stranded_printing_items()
+
+        assert await self._status_of(db_session, item.id) == "failed"
+
+    async def test_rows_that_are_not_printing_are_ignored(self, scheduler, db_session, printer_factory, monkeypatch):
+        printer = await printer_factory()
+        pending = await self._item(db_session, printer, status="pending")
+        monkeypatch.setattr(
+            "backend.app.services.print_scheduler.printer_manager.get_status", lambda _pid: _state("FINISH")
+        )
+
+        await scheduler._close_stranded_printing_items()
+        scheduler._terminal_since[printer.id] = 0.0
+        await scheduler._close_stranded_printing_items()
+
+        assert await self._status_of(db_session, pending.id) == "pending"
+
+    async def test_a_completed_row_clears_the_clock(self, scheduler, db_session, printer_factory, monkeypatch):
+        """Nothing printing means nothing to time, and a stale entry would give
+        the next print a head start on its own grace period."""
+        printer = await printer_factory()
+        monkeypatch.setattr(
+            "backend.app.services.print_scheduler.printer_manager.get_status", lambda _pid: _state("FINISH")
+        )
+        scheduler._terminal_since[printer.id] = 0.0
+
+        await scheduler._close_stranded_printing_items()
+
+        assert scheduler._terminal_since == {}
+
+    async def test_the_scheduler_loop_actually_runs_the_sweep(self, scheduler, monkeypatch):
+        """The sweep is only worth anything if the loop calls it.
+
+        Without this, every test above passes against a build where the call
+        was never wired in -- which is exactly what a mutation check caught.
+        """
+        called = []
+        monkeypatch.setattr(scheduler, "_close_stranded_printing_items", lambda: called.append(True) or _noop())
+        monkeypatch.setattr(scheduler, "_clear_stale_dispatch_claims", lambda **_kw: _noop())
+        monkeypatch.setattr(scheduler, "_sample_chamber_temps", lambda: None)
+
+        async def stop_after_one_pass():
+            scheduler._running = False
+            return False
+
+        monkeypatch.setattr(scheduler, "check_queue", stop_after_one_pass)
+        monkeypatch.setattr("backend.app.services.print_scheduler.asyncio.sleep", _noop_arg)
+
+        await scheduler.run()
+
+        assert called, "the scheduler loop never called the stranded-item sweep"
+
+    async def test_a_broken_sweep_does_not_break_the_scheduler_loop(self, scheduler, monkeypatch):
+        """It runs beside the dispatch-claim sweep on every tick. A recovery
+        path that can take the loop down is worse than the strand it fixes."""
+        monkeypatch.setattr(
+            "backend.app.services.print_scheduler.printer_manager.get_status",
+            lambda _pid: (_ for _ in ()).throw(RuntimeError("boom")),
+        )
+
+        await scheduler._close_stranded_printing_items()  # must not raise

backend/tests/unit/test_completion_subtask_match_2829.py

@@ -0,0 +1,127 @@
+"""Matching a completion event to the queue row it belongs to (#2829).
+
+``on_print_complete`` finds its row by printer and ``status='printing'`` alone,
+so #b5a34b7ba added a check that the completion's subtask name agrees with the
+file the row was dispatched with -- otherwise the printer's own calibration
+runs close whoever's job happens to be printing.
+
+The check compared the two names with plain equality, and the printer does not
+echo the name back verbatim. Three days later two users had queues that would
+not advance: the row stayed ``printing``, ``check_queue`` counts every such row
+as a busy printer, and nothing anywhere ever closes it. Cancelling by hand was
+the only way out.
+
+The strings below are the real ones from the maintainer's own H2D, queue item
+649, and from a support bundle showing the truncation case.
+"""
+
+import pytest
+
+from backend.app.main import _normalise_subtask_name, _subtask_name_from_filename, _subtask_names_match
+
+pytestmark = pytest.mark.unit
+
+
+class TestTheReportedCase:
+    def test_spaces_come_back_as_underscores(self):
+        """Queue item 649, verbatim from the warning it logged twice."""
+        dispatched = "H2D_Carbon_Filter_(V2)_Body & Solid Lid"
+        reported = "H2D_Carbon_Filter_(V2)_Body_&_Solid_Lid"
+
+        assert _subtask_names_match(dispatched, reported)
+
+    def test_from_the_archive_filename_it_was_dispatched_with(self):
+        """End to end from the stored filename, which is where the check
+        actually gets its side of the comparison."""
+        expected = _subtask_name_from_filename("H2D_Carbon_Filter_(V2)_Body & Solid Lid.gcode.3mf")
+
+        assert _subtask_names_match(expected, "H2D_Carbon_Filter_(V2)_Body_&_Solid_Lid")
+
+
+class TestTruncation:
+    """The printer cuts long names and marks the cut with '...'.
+
+    Observed at ~100 characters, but not a fixed count -- a name with multibyte
+    characters came back at 98 -- so the marker is what is matched, not a
+    length. Without this every print with a long name strands its row the same
+    way the space substitution did.
+    """
+
+    def test_a_truncated_echo_matches_the_full_name(self):
+        full = (
+            "169356_204314.STEP + 169356_204314.STEP + 169356_204314.STEP + "
+            "169356_204314.STEP + 169356_204314.STEP + 169356_204314.STEP"
+        )
+        truncated = (
+            "169356_204314.STEP + 169356_204314.STEP + 169356_204314.STEP + 169356_204314.STEP + 169356_204314..."
+        )
+
+        assert _subtask_names_match(full, truncated)
+
+    def test_a_truncated_name_on_the_archive_side_matches_too(self):
+        """An archive whose filename was recorded from an earlier truncated
+        echo carries the marker itself, so the cut can be on either side."""
+        stored = "EXXXX-A001-Barriere Mundstück.STEP + EXXXX-A001-Barriere M..."
+        reported = "EXXXX-A001-Barriere_Mundstück.STEP_+_EXXXX-A001-Barriere_Mundstück.STEP"
+
+        assert _subtask_names_match(stored, reported)
+
+    def test_truncation_does_not_match_a_different_print(self):
+        """The prefix still has to agree -- '...' is not a wildcard."""
+        assert not _subtask_names_match("Benchy_Calibration_Cube_Large", "Something_Else_Entirely...")
+
+
+class TestItStillRefusesADifferentPrint:
+    """The check has to keep doing its job, or #b5a34b7ba's bug comes back:
+    a completion for another print closing a job that is still running.
+    """
+
+    def test_the_printers_own_calibration_run(self):
+        """The second rejection on queue item 649, and a correct one."""
+        assert not _subtask_names_match("H2D_Carbon_Filter_(V2)_Body & Solid Lid", "auto_pa_line_calib_mode")
+
+    def test_an_unrelated_print(self):
+        assert not _subtask_names_match("Benchy", "Calibration Cube")
+
+    def test_a_name_that_merely_starts_the_same(self):
+        assert not _subtask_names_match("Bracket_v1", "Bracket_v2")
+
+
+class TestNormalisation:
+    def test_case_is_ignored(self):
+        assert _subtask_names_match("BENCHY BOAT", "benchy_boat")
+
+    def test_surrounding_whitespace_is_ignored(self):
+        assert _subtask_names_match("  Benchy  ", "Benchy")
+
+    @pytest.mark.parametrize(
+        "raw,expected",
+        [
+            ("A B", "a_b"),
+            ("A_B", "a_b"),
+            (" A  B ", "a__b"),
+            ("Mundstück", "mundstück"),
+        ],
+    )
+    def test_canonical_form(self, raw, expected):
+        assert _normalise_subtask_name(raw) == expected
+
+    def test_spaces_and_underscores_are_the_same_rule_the_3mf_lookup_uses(self):
+        """The 3MF lookup has always built space-to-underscore variants of its
+        candidates. The completion check growing its own comparison instead of
+        reading the same rule is how the two came to disagree."""
+        assert _normalise_subtask_name("My Model") == _normalise_subtask_name("My_Model")
+
+
+class TestFilenameDerivation:
+    @pytest.mark.parametrize(
+        "filename,expected",
+        [
+            ("Benchy.gcode.3mf", "Benchy"),
+            ("Benchy.3mf", "Benchy"),
+            ("My.Model.3mf", "My.Model"),
+            ("/cache/Nested Path/Benchy.gcode.3mf", "Benchy"),
+        ],
+    )
+    def test_extensions_come_off_and_nothing_else_does(self, filename, expected):
+        assert _subtask_name_from_filename(filename) == expected