MQTT: announce each print's run (archive redesign P3)

- run/started, run/completed and run/discarded under
  printers/{serial}/: the run, its archive, printer and plate, and at
  its end its outcome, times (UTC) and figures; sent once per print,
  also for a stop while the printer is offline
- The existing print and archive events are unchanged: they go out
  before the print's run exists
- Notifications of a print's end have {archive_id} and {run_id}, the
  kill switch's stop notification too

backend/app/api/routes/print_queue.py

@@ -2975,6 +2975,13 @@ async def stop_queue_item(
     await refresh_batches_after([item.id])
     if closed_archive_id is not None:
         await ws_manager.send_archive_updated({"id": closed_archive_id, "status": "cancelled"})
+    if closed_archive_id is not None and printer_id is not None:
+        # Its run ended here, not at a print's end (archive redesign).
+        from backend.app.main import _publish_run_completed
+        from backend.app.services.print_log import run_being_ended
+
+        ended = await run_being_ended(db, closed_archive_id, printer_id, None)
+        await _publish_run_completed(ended.id if ended is not None else None, printer_id)
 
     logger.info("Stopped printing queue item %s (stop command sent: %s)", item_id, stop_sent)
 

backend/app/main.py

@@ -988,6 +988,47 @@ async def _archive_of_open_run(printer_id: int, subtask_id: str | None = None) -
         return None
 
 
+async def _announce_discarded_runs(discarded, printer_id: int) -> None:
+    """Publish the runs ``discard_open_runs`` dropped (archive redesign):
+    each announced by run/started, now gone without an end."""
+    if not discarded:
+        return
+    try:
+        from backend.app.models.printer import Printer
+
+        async with async_session() as db:
+            printer = (
+                await db.execute(select(Printer.name, Printer.serial_number).where(Printer.id == printer_id))
+            ).first()
+        if printer is not None:
+            for run in discarded:
+                await mqtt_relay.on_run_discarded(run, printer.serial_number, printer.name)
+    except Exception as e:
+        logging.getLogger(__name__).debug("[RUN] Could not announce discarded runs on printer %s: %s", printer_id, e)
+
+
+async def _publish_run_completed(run_id: int | None, printer_id: int) -> None:
+    """Publish the run a print's end just wrote on the MQTT relay (archive
+    redesign); nothing without one. Never in the way of the completion."""
+    if run_id is None:
+        return
+    try:
+        from backend.app.models.archive import PrintArchive
+        from backend.app.models.print_log import PrintLogEntry
+        from backend.app.models.printer import Printer
+
+        async with async_session() as db:
+            run = await db.get(PrintLogEntry, run_id)
+            printer = (
+                await db.execute(select(Printer.name, Printer.serial_number).where(Printer.id == printer_id))
+            ).first()
+            archive = await db.get(PrintArchive, run.archive_id) if run is not None and run.archive_id else None
+        if run is not None and printer is not None:
+            await mqtt_relay.on_run_completed(run, archive, printer.serial_number, printer.name)
+    except Exception as e:
+        logging.getLogger(__name__).debug("[RUN] Could not publish the end of run %s: %s", run_id, e)
+
+
 async def _open_run_for(archive_id: int, printer_id: int, subtask_id: str | None = None) -> None:
     """Write the run of the print that just started on the archive (archive
     redesign). In a session of its own and never in the way of the print
@@ -997,14 +1038,19 @@ async def _open_run_for(archive_id: int, printer_id: int, subtask_id: str | None
     try:
         from backend.app.models.archive import PrintArchive
         from backend.app.models.printer import Printer
-        from backend.app.services.print_log import open_run
+        from backend.app.services.print_log import open_run, open_run_on_printer, run_snapshot
 
         async with async_session() as db:
             archive = await db.get(PrintArchive, archive_id)
             if archive is None:
                 return
-            printer_name = await db.scalar(select(Printer.name).where(Printer.id == printer_id))
-            await open_run(
+            printer = (
+                await db.execute(select(Printer.name, Printer.serial_number).where(Printer.id == printer_id))
+            ).first()
+            printer_name = printer.name if printer is not None else None
+            already = await open_run_on_printer(db, printer_id)
+            before = run_snapshot(already) if already is not None else None
+            run = await open_run(
                 db,
                 archive,
                 printer_id=printer_id,
@@ -1013,6 +1059,21 @@ async def _open_run_for(archive_id: int, printer_id: int, subtask_id: str | None
                 subtask_id=subtask_id,
             )
             await db.commit()
+        # Once per print: a resume or a repeated start finds the run it opened.
+        # A new one replaced the run still open there (open_run): that one's
+        # print ended unseen.
+        # Compared by start too: SQLite can give the new run the id of the one
+        # it replaced.
+        replaced = (
+            run is not None and before is not None and (before.id != run.id or before.started_at != run.started_at)
+        )
+        if run is not None and printer is not None and (before is None or replaced):
+            if replaced:
+                await _announce_discarded_runs([before], printer_id)
+            try:
+                await mqtt_relay.on_run_started(run, archive, printer.serial_number, printer_name)
+            except Exception:
+                pass  # Don't fail the print start if MQTT fails
     except Exception as e:
         logging.getLogger(__name__).warning("[RUN] Could not open the run of archive %s: %s", archive_id, e)
 
@@ -1202,12 +1263,18 @@ async def _send_kill_switch_provider_notification(
     logger = logging.getLogger(__name__)
     try:
         async with async_session() as db:
+            from backend.app.services.print_log import open_run_on_printer
+
+            # The print it stopped, for {archive_id} / {run_id}: the completion
+            # that would name them skips its notification after this one.
+            run = await open_run_on_printer(db, printer_id)
             await notification_service.on_print_complete(
                 printer_id,
                 printer_name,
                 "stopped",
                 data,
                 db,
+                archive_data={"archive_id": run.archive_id, "run_id": run.id} if run is not None else None,
             )
         return True
     except Exception as e:
@@ -5572,8 +5639,9 @@ async def on_print_start(printer_id: int, data: dict):
                 # unseen. Left open, the next reconnect would close it again.
                 from backend.app.services.print_log import discard_open_runs
 
-                await discard_open_runs(db, printer_id)
+                discarded = await discard_open_runs(db, printer_id)
                 await db.commit()
+                await _announce_discarded_runs(discarded, printer_id)
                 # Its print keys go with it; the new archive registers its own.
                 for _key in [k for k, v in _active_prints.items() if v == existing_archive.id and k[0] == printer_id]:
                     del _active_prints[_key]
@@ -9726,8 +9794,9 @@ async def _close_out_print(printer_id: int, data: dict):
         archive = await db.get(PrintArchive, archive_id)
         if archive is None:
             # Deleted while it printed: its run has nothing left to end on.
-            await discard_open_runs(db, printer_id)
+            discarded = await discard_open_runs(db, printer_id)
             await db.commit()
+            await _announce_discarded_runs(discarded, printer_id)
         if archive:
             # Back-fill created_by_id on reprint (#730): reprint reuses the
             # source archive row rather than creating a new one, so an
@@ -9829,6 +9898,7 @@ async def _close_out_print(printer_id: int, data: dict):
         await run_with_retry(_write_run, label="print log entry")
     except Exception as e:
         logger.warning("[PRINT_LOG] Failed to write log entry for archive %s: %s", archive_id, e)
+    await _publish_run_completed(completed_run_id, printer_id)
 
     log_timing("Print log entry")
 
@@ -10321,6 +10391,9 @@ async def _close_out_print(printer_id: int, data: dict):
                         # Pass usage tracker results for AMS slot info in notifications
                         if usage_results:
                             archive_data["usage_results"] = usage_results
+                        # Which archive and print it was (archive redesign).
+                        archive_data["archive_id"] = archive_id
+                        archive_data["run_id"] = completed_run_id
                         # Add finish photo URL and image bytes if available
                         if finish_photo_filename:
                             photo_url, photo_bytes = await _finish_photo_for_notification(

backend/app/schemas/notification_template.py

@@ -52,6 +52,8 @@ EVENT_VARIABLES: dict[str, list[str]] = {
         "duration",
         "filament_grams",
         "filament_details",
+        "archive_id",
+        "run_id",
         "finish_photo_url",
         "timestamp",
         "app_name",
@@ -64,6 +66,8 @@ EVENT_VARIABLES: dict[str, list[str]] = {
         "filament_details",
         "progress",
         "reason",
+        "archive_id",
+        "run_id",
         "finish_photo_url",
         "timestamp",
         "app_name",
@@ -75,6 +79,8 @@ EVENT_VARIABLES: dict[str, list[str]] = {
         "filament_grams",
         "filament_details",
         "progress",
+        "archive_id",
+        "run_id",
         "finish_photo_url",
         "timestamp",
         "app_name",
@@ -198,6 +204,8 @@ SAMPLE_DATA: dict[str, dict[str, str]] = {
         "filament_grams": "15.2",
         "filament_details": "AMS-A T1 PLA: 12.4g | AMS-A T3 PETG: 2.8g",
         "finish_photo_url": "/api/v1/archives/123/photos/finish_20240115_154800_abc12345.jpg",
+        "archive_id": "123",
+        "run_id": "456",
         "timestamp": "2024-01-15 15:48",
         "app_name": "Bambuddy",
     },
@@ -210,6 +218,8 @@ SAMPLE_DATA: dict[str, dict[str, str]] = {
         "progress": "50",
         "reason": "Filament runout",
         "finish_photo_url": "/api/v1/archives/123/photos/finish_20240115_151500_def67890.jpg",
+        "archive_id": "123",
+        "run_id": "456",
         "timestamp": "2024-01-15 15:15",
         "app_name": "Bambuddy",
     },
@@ -221,6 +231,8 @@ SAMPLE_DATA: dict[str, dict[str, str]] = {
         "filament_details": "AMS-A T2 PLA: 4.6g",
         "progress": "30",
         "finish_photo_url": "/api/v1/archives/123/photos/finish_20240115_150000_ghi11223.jpg",
+        "archive_id": "123",
+        "run_id": "456",
         "timestamp": "2024-01-15 15:00",
         "app_name": "Bambuddy",
     },

backend/app/services/mqtt_relay.py

@@ -642,6 +642,74 @@ class MQTTRelayService:
             },
         )
 
+    # =========================================================================
+    # Run Events (archive redesign: each print of an archive is a run)
+    # =========================================================================
+
+    @staticmethod
+    def _run_payload(run, archive, printer_name: str | None) -> dict:
+        def iso(moment):
+            # Stored as naive UTC: say so.
+            if moment is None:
+                return None
+            return (moment if moment.tzinfo else moment.replace(tzinfo=timezone.utc)).isoformat()
+
+        return {
+            "run_id": run.id,
+            "archive_id": run.archive_id,
+            "print_name": run.print_name or (archive.print_name if archive is not None else None),
+            "printer_id": run.printer_id,
+            "printer_name": printer_name or run.printer_name,
+            "plate_id": run.plate_id,
+            "status": run.status,
+            "started_at": iso(run.started_at),
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+        }
+
+    async def on_run_started(self, run, archive, printer_serial: str, printer_name: str | None):
+        """Publish a print's run opened at its start: the print of an archive,
+        which the print and archive events above do not name yet."""
+        if not self.enabled or not self.connected:
+            return
+        self._publish(
+            f"{self.topic_prefix}/printers/{printer_serial}/run/started",
+            self._run_payload(run, archive, printer_name),
+        )
+
+    async def on_run_completed(self, run, archive, printer_serial: str, printer_name: str | None):
+        """Publish a print's run as its end wrote it: outcome and figures."""
+        if not self.enabled or not self.connected:
+            return
+        payload = self._run_payload(run, archive, printer_name)
+        completed_at = run.completed_at
+        if completed_at is not None and completed_at.tzinfo is None:
+            completed_at = completed_at.replace(tzinfo=timezone.utc)
+        payload.update(
+            {
+                "completed_at": completed_at.isoformat() if completed_at is not None else None,
+                "duration_seconds": run.duration_seconds,
+                # Not measured: an end found at a reconnect, its time the
+                # estimate or nothing (0).
+                "duration_unmeasured": bool(run.duration_unmeasured),
+                "failure_reason": run.failure_reason,
+                "filament_used_grams": run.filament_used_grams,
+                "cost": run.cost,
+                "energy_kwh": run.energy_kwh,
+            }
+        )
+        self._publish(f"{self.topic_prefix}/printers/{printer_serial}/run/completed", payload)
+
+    async def on_run_discarded(self, run, printer_serial: str, printer_name: str | None):
+        """Publish a run dropped without an end: its print ended unseen (a new
+        print started on the printer) or its archive was deleted. Its id is
+        gone after this."""
+        if not self.enabled or not self.connected:
+            return
+        self._publish(
+            f"{self.topic_prefix}/printers/{printer_serial}/run/discarded",
+            self._run_payload(run, None, printer_name),
+        )
+
     # =========================================================================
     # Filament/Spoolman Events
     # =========================================================================

backend/app/services/notification_service.py

@@ -1915,6 +1915,11 @@ class NotificationService:
                 variables["reason"] = archive_data["failure_reason"]
             if archive_data.get("finish_photo_url"):
                 variables["finish_photo_url"] = archive_data["finish_photo_url"]
+            # The archive (file) and run (this print) it was, for links and
+            # webhooks (archive redesign).
+            for key in ("archive_id", "run_id"):
+                if archive_data.get(key) is not None:
+                    variables[key] = str(archive_data[key])
 
             # Build per-slot breakdown string with AMS info when available
             if archive_data.get("usage_results"):

backend/app/services/print_log.py

@@ -377,16 +377,40 @@ async def open_run_on_printer(
     )
 
 
-async def discard_open_runs(db: AsyncSession, printer_id: int) -> None:
+async def discard_open_runs(db: AsyncSession, printer_id: int) -> list[SimpleNamespace]:
     """Drop the runs left open on the printer by prints that ended without
     their end reaching a run: cancelled at the next start as stale, or whose
     archive was deleted. As before runs were opened at the start, such a
-    print leaves no run. Does not commit."""
+    print leaves no run. Returns what they were, for announcing them gone.
+    Does not commit."""
+    discarded = [
+        run_snapshot(run)
+        for run in await db.scalars(
+            select(PrintLogEntry)
+            .where(PrintLogEntry.printer_id == printer_id, PrintLogEntry.status == RUN_OPEN)
+            .execution_options(**{INCLUDE_OPEN_RUNS: True})
+        )
+    ]
     await db.execute(
         delete(PrintLogEntry)
         .where(PrintLogEntry.printer_id == printer_id, PrintLogEntry.status == RUN_OPEN)
         .execution_options(**{INCLUDE_OPEN_RUNS: True})
     )
+    return discarded
+
+
+def run_snapshot(run: PrintLogEntry) -> SimpleNamespace:
+    """What a run was, kept past its row (``discard_open_runs``)."""
+    return SimpleNamespace(
+        id=run.id,
+        archive_id=run.archive_id,
+        printer_id=run.printer_id,
+        printer_name=run.printer_name,
+        plate_id=run.plate_id,
+        print_name=run.print_name,
+        status=run.status,
+        started_at=run.started_at,
+    )
 
 
 async def open_run(

backend/tests/integration/test_print_complete_end_to_end.py

@@ -602,3 +602,15 @@ async def test_the_completions_media_work_holds_the_archive(test_engine, db_sess
     }
     assert sum(1 for archive_id, _ in held if archive_id == queued_print["archive"]) >= 3
     assert not archive_work.is_held(queued_print["archive"])
+
+
+async def test_the_print_end_announces_the_run_it_wrote(test_engine, db_session, queued_print):
+    """On MQTT, and in the notification's variables (archive redesign P3)."""
+    publish = AsyncMock()
+    with patch("backend.app.main._publish_run_completed", publish):
+        notif = await _complete(test_engine, queued_print["printer"], "completed")
+
+    _archive, _item, logs, _charges = await _rows(db_session, queued_print["archive"], queued_print["item"])
+    publish.assert_awaited_once_with(logs[0].id, queued_print["printer"])
+    archive_data = notif.on_print_complete.await_args.kwargs["archive_data"]
+    assert (archive_data["archive_id"], archive_data["run_id"]) == (queued_print["archive"], logs[0].id)

backend/tests/integration/test_run_events.py

@@ -0,0 +1,267 @@
+"""Each print of an archive is announced as its run (archive redesign P3).
+
+The print and archive events go out before the print's run exists, so they
+cannot name it. ``run/started`` and ``run/completed`` do, once it does: the
+run, its archive, printer and plate, and at the end its outcome and figures.
+Notifications of a print's end carry the archive and run ids too.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app import main as main_module
+from backend.app.models.archive import PrintArchive
+from backend.app.models.print_log import INCLUDE_OPEN_RUNS, PrintLogEntry
+from backend.app.services.mqtt_relay import MQTTRelayService
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+T1 = datetime(2026, 10, 31, 9)
+
+
+@pytest.fixture
+def relay():
+    relay = MagicMock(on_run_started=AsyncMock(), on_run_completed=AsyncMock())
+    with patch.object(main_module, "mqtt_relay", relay):
+        yield relay
+
+
+@pytest.fixture
+def sessions(test_engine):
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    with patch.object(main_module, "async_session", maker):
+        yield maker
+
+
+async def _archive(db_session, printer_id: int) -> int:
+    archive = PrintArchive(
+        printer_id=printer_id,
+        filename="part.3mf",
+        print_name="part",
+        file_path="archives/test/part.3mf",
+        file_size=1,
+        status="printing",
+        started_at=T1,
+        subtask_id="job-1",
+    )
+    db_session.add(archive)
+    await db_session.commit()
+    return archive.id
+
+
+async def test_a_print_start_announces_its_run_once(db_session, printer_factory, relay, sessions):
+    printer = await printer_factory(name="A")
+    archive_id = await _archive(db_session, printer.id)
+
+    await main_module._open_run_for(archive_id, printer.id, "job-1")
+    # A restart resuming the print finds the run it opened.
+    await main_module._open_run_for(archive_id, printer.id, "job-1")
+
+    relay.on_run_started.assert_awaited_once()
+    run, archive, serial, name = relay.on_run_started.await_args.args
+    opened = await db_session.scalar(
+        select(PrintLogEntry.id)
+        .where(PrintLogEntry.archive_id == archive_id)
+        .execution_options(**{INCLUDE_OPEN_RUNS: True})
+    )
+    assert (run.id, archive.id, serial, name) == (opened, archive_id, printer.serial_number, "A")
+
+
+async def test_a_print_end_announces_its_run(db_session, printer_factory, relay, sessions):
+    printer = await printer_factory(name="A")
+    archive_id = await _archive(db_session, printer.id)
+    run = PrintLogEntry(archive_id=archive_id, printer_id=printer.id, status="failed", started_at=T1)
+    db_session.add(run)
+    await db_session.commit()
+
+    await main_module._publish_run_completed(run.id, printer.id)
+
+    relay.on_run_completed.assert_awaited_once()
+    published, archive, serial, _ = relay.on_run_completed.await_args.args
+    assert (published.id, archive.id, serial) == (run.id, archive_id, printer.serial_number)
+
+
+async def test_no_run_written_announces_nothing(relay, sessions):
+    await main_module._publish_run_completed(None, 1)
+
+    relay.on_run_completed.assert_not_awaited()
+
+
+def _relay() -> MQTTRelayService:
+    service = MQTTRelayService()
+    service.enabled = True
+    service.connected = True
+    service.topic_prefix = "bambuddy"
+    service._publish = MagicMock()
+    return service
+
+
+async def test_the_run_payloads():
+    run = PrintLogEntry(
+        id=7,
+        archive_id=3,
+        printer_id=2,
+        plate_id=2,
+        status="completed",
+        started_at=T1,
+        completed_at=T1.replace(hour=10),
+        duration_seconds=3600,
+        filament_used_grams=12.5,
+        cost=0.4,
+        print_name=None,
+    )
+    archive = PrintArchive(id=3, print_name="part", filename="part.3mf", file_path="", file_size=1)
+    service = _relay()
+
+    await service.on_run_started(run, archive, "SERIAL", "A")
+    await service.on_run_completed(run, archive, "SERIAL", "A")
+
+    (started_topic, started), (completed_topic, completed) = (c.args for c in service._publish.call_args_list)
+    assert started_topic == "bambuddy/printers/SERIAL/run/started"
+    assert {k: started[k] for k in ("run_id", "archive_id", "printer_id", "plate_id", "print_name")} == {
+        "run_id": 7,
+        "archive_id": 3,
+        "printer_id": 2,
+        "plate_id": 2,
+        "print_name": "part",
+    }
+    assert completed_topic == "bambuddy/printers/SERIAL/run/completed"
+    assert (completed["status"], completed["duration_seconds"], completed["filament_used_grams"]) == (
+        "completed",
+        3600,
+        12.5,
+    )
+
+
+async def test_a_print_end_notification_names_its_archive_and_run(db_session):
+    from backend.app.services.notification_service import NotificationService
+
+    service = NotificationService()
+    provider = MagicMock(name="provider")
+    with (
+        patch.object(service, "_get_providers_for_event", AsyncMock(return_value=[provider])),
+        patch.object(service, "_build_message_from_template", AsyncMock(return_value=("t", "m"))),
+        patch.object(service, "_send_to_providers", AsyncMock()) as send,
+    ):
+        await service.on_print_complete(
+            1, "A", "completed", {"filename": "part.3mf"}, db_session, archive_data={"archive_id": 3, "run_id": 7}
+        )
+
+    variables = send.await_args.kwargs["variables"]
+    assert (variables["archive_id"], variables["run_id"]) == ("3", "7")
+
+
+async def test_a_run_replaced_by_the_next_start_is_announced_gone(db_session, printer_factory, relay, sessions):
+    """Its print ended unseen; a new print opened the printer's next run."""
+    relay.on_run_discarded = AsyncMock()
+    printer = await printer_factory(name="A")
+    archive_id = await _archive(db_session, printer.id)
+    await main_module._open_run_for(archive_id, printer.id, "job-1")
+    first = await db_session.scalar(
+        select(PrintLogEntry.id)
+        .where(PrintLogEntry.archive_id == archive_id)
+        .execution_options(**{INCLUDE_OPEN_RUNS: True})
+    )
+    archive = await db_session.get(PrintArchive, archive_id)
+    archive.started_at, archive.subtask_id = T1.replace(hour=11), "job-2"
+    await db_session.commit()
+
+    await main_module._open_run_for(archive_id, printer.id, "job-2")
+
+    relay.on_run_discarded.assert_awaited_once()
+    assert relay.on_run_discarded.await_args.args[0].id == first
+    assert relay.on_run_started.await_count == 2
+
+
+async def test_discarding_announces_each_run(db_session, printer_factory, relay, sessions):
+    from backend.app.models.print_log import RUN_OPEN
+    from backend.app.services.print_log import discard_open_runs
+
+    relay.on_run_discarded = AsyncMock()
+    printer = await printer_factory(name="A")
+    archive_id = await _archive(db_session, printer.id)
+    run = PrintLogEntry(archive_id=archive_id, printer_id=printer.id, status=RUN_OPEN, started_at=T1)
+    db_session.add(run)
+    await db_session.commit()
+    run_id = run.id
+
+    discarded = await discard_open_runs(db_session, printer.id)
+    await db_session.commit()
+    await main_module._announce_discarded_runs(discarded, printer.id)
+
+    assert [r.id for r in discarded] == [run_id]
+    relay.on_run_discarded.assert_awaited_once()
+    assert relay.on_run_discarded.await_args.args[1] == printer.serial_number
+
+
+async def test_times_are_utc_and_an_unmeasured_duration_says_so():
+    run = PrintLogEntry(
+        id=7,
+        archive_id=3,
+        printer_id=2,
+        status="completed",
+        started_at=T1,
+        completed_at=T1.replace(hour=10),
+        duration_seconds=0,
+        duration_unmeasured=True,
+    )
+    service = _relay()
+
+    await service.on_run_completed(run, None, "SERIAL", "A")
+
+    payload = service._publish.call_args.args[1]
+    assert (payload["started_at"], payload["completed_at"]) == (
+        "2026-10-31T09:00:00+00:00",
+        "2026-10-31T10:00:00+00:00",
+    )
+    assert payload["duration_unmeasured"] is True
+
+
+async def test_the_kill_switch_notification_names_the_print(db_session, printer_factory, sessions):
+    from backend.app.models.print_log import RUN_OPEN
+
+    printer = await printer_factory(name="A")
+    archive_id = await _archive(db_session, printer.id)
+    run = PrintLogEntry(archive_id=archive_id, printer_id=printer.id, status=RUN_OPEN, started_at=T1)
+    db_session.add(run)
+    await db_session.commit()
+    run_id = run.id
+
+    with patch.object(main_module, "notification_service") as notify:
+        notify.on_print_complete = AsyncMock()
+        assert await main_module._send_kill_switch_provider_notification(printer.id, "A", {"filename": "part.3mf"})
+
+    assert notify.on_print_complete.await_args.kwargs["archive_data"] == {"archive_id": archive_id, "run_id": run_id}
+
+
+async def test_a_stop_while_the_printer_is_offline_announces_the_run_end(
+    async_client, db_session, printer_factory, relay
+):
+    from backend.app.models.print_log import RUN_OPEN
+    from backend.app.models.print_queue import PrintQueueItem
+
+    printer = await printer_factory(name="A")
+    archive_id = await _archive(db_session, printer.id)
+    item = PrintQueueItem(printer_id=printer.id, archive_id=archive_id, status="printing", position=1)
+    db_session.add(item)
+    run = PrintLogEntry(archive_id=archive_id, printer_id=printer.id, status=RUN_OPEN, started_at=T1)
+    db_session.add(run)
+    await db_session.commit()
+    run_id, item_id = run.id, item.id
+
+    with (
+        patch("backend.app.services.print_control.stop_print_by_user", return_value=False),
+        patch("backend.app.api.routes.print_queue.ws_manager") as ws,
+    ):
+        ws.send_archive_updated = AsyncMock()
+        response = await async_client.post(f"/api/v1/queue/{item_id}/stop")
+
+    assert response.status_code == 200, response.text
+    relay.on_run_completed.assert_awaited_once()
+    assert relay.on_run_completed.await_args.args[0].id == run_id

backend/tests/integration/test_run_lifecycle.py

@@ -1167,7 +1167,11 @@ async def test_a_stale_archive_cancelled_at_the_next_start_leaves_no_run_open(te
     )
     await db_session.commit()
 
-    with patch("backend.app.main._open_run_for", AsyncMock()):
+    announce = AsyncMock()
+    with (
+        patch("backend.app.main._open_run_for", AsyncMock()),
+        patch("backend.app.main._announce_discarded_runs", announce),
+    ):
         await _drive_print_start(
             test_engine, SimpleNamespace(id=farm.printer), download_ok=False, state_fields={"progress": 0.2}
         )
@@ -1176,6 +1180,9 @@ async def test_a_stale_archive_cancelled_at_the_next_start_leaves_no_run_open(te
     archive = await db_session.get(PrintArchive, farm.archive)
     assert archive.status == "cancelled"
     assert await _all_runs(db_session, farm.archive) == []
+    # Announced gone on MQTT (archive redesign P3): run/started named it.
+    [(discarded, printer_id)] = [c.args for c in announce.await_args_list]
+    assert ([r.archive_id for r in discarded], printer_id) == ([farm.archive], farm.printer)
 
 
 async def _charges(db_session) -> list[WalletTransaction]: