Archives: reconcile and match a print's end by its run (archive redesign P1b)

- The reconnect reconcile judges each run open on the printer by its own
  job, start and plate; another printer's plate of the archive is left
  alone. Its synthesised end names the archive, and closes it only while
  that run is still open there
- A stop that can't reach the printer flags its run ended_offline; only
  the print's real end replaces it, while it is the printer's latest run
- The stale cancel at a print start drops the stale run and unbinds only
  its own printer

backend/app/api/routes/print_queue.py

@@ -2954,6 +2954,7 @@ async def stop_queue_item(
                 failure_reason=archive.failure_reason,
                 thumbnail_path=archive.thumbnail_path,
                 created_by_id=archive.created_by_id,
+                ended_offline=True,
             )
             closed_archive_id = archive.id
     if not stop_sent:

backend/app/core/database.py

@@ -5942,6 +5942,7 @@ async def _run_migrations(conn):
         ("bambuddy_forced_timelapse", "BOOLEAN"),
         ("notes", "TEXT"),
         ("quantity", "INTEGER"),
+        ("ended_offline", "BOOLEAN"),
     ):
         await _safe_execute(conn, f"ALTER TABLE print_log_entries ADD COLUMN {column} {kind}")
     await _safe_execute(

backend/app/main.py

@@ -5535,9 +5535,15 @@ async def on_print_start(printer_id: int, data: dict):
                 # key loses no information and gives the Statistics breakdown a
                 # single bucket instead of two untranslatable prose strings.
                 existing_archive.failure_reason = "noStatusUpdate"
+                # And its run, as before runs were opened at the start: a new
+                # print starts on the printer, so whatever is open there ended
+                # 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)
                 await db.commit()
                 # 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]:
+                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]
                 # Fall through to create new archive (don't return)
             else:
@@ -7542,6 +7548,20 @@ async def on_print_running_observed(printer_id: int, data: dict):
     await _capture_timelapse_baseline_at_start(printer, printer_id, logger)
 
 
+class _RunOfArchive:
+    """An archive as one of its runs printed it: the run's job id, start and
+    plate, the rest the archive's. What the reconcile judges and closes."""
+
+    def __init__(self, archive, run):
+        self._archive = archive
+        self.subtask_id = run.subtask_id
+        self.started_at = run.started_at
+        self.plate_id = run.plate_id if run.plate_id is not None else archive.plate_id
+
+    def __getattr__(self, name):
+        return getattr(self._archive, name)
+
+
 def _is_active_archive_stale(archive, state) -> tuple[bool, str]:
     """Return ``(is_stale, reason)`` for an archive in ``status="printing"``
     against the printer's current MQTT state.
@@ -7789,16 +7809,46 @@ async def reconcile_stale_active_prints(printer_id: int) -> int:
         return 0
 
     from backend.app.models.archive import PrintArchive
+    from backend.app.models.print_log import INCLUDE_OPEN_RUNS, RUN_OPEN, PrintLogEntry
 
     reconciled = 0
     async with async_session() as db:
+        # The prints on this printer Bambuddy saw start: their runs open here.
+        # The archive row is shared by every plate of a file, its printer, job
+        # and start the latest start's, so each is judged as its run.
+        open_runs = (
+            await db.scalars(
+                select(PrintLogEntry)
+                .where(PrintLogEntry.printer_id == printer_id, PrintLogEntry.status == RUN_OPEN)
+                .execution_options(**{INCLUDE_OPEN_RUNS: True})
+            )
+        ).all()
+        by_run = {
+            archive.id: archive
+            for archive in (
+                await db.scalars(
+                    select(PrintArchive).where(
+                        PrintArchive.id.in_([run.archive_id for run in open_runs if run.archive_id is not None])
+                    )
+                )
+            ).all()
+        }
+        active: list = [_RunOfArchive(by_run[run.archive_id], run) for run in open_runs if run.archive_id in by_run]
+        # And as before runs were opened at the start: a printing archive of
+        # this printer with no run open anywhere (printing since before).
+        archives_with_open_runs = select(PrintLogEntry.archive_id).where(
+            PrintLogEntry.status == RUN_OPEN, PrintLogEntry.archive_id.is_not(None)
+        )
         result = await db.execute(
-            select(PrintArchive).where(
+            select(PrintArchive)
+            .where(
                 PrintArchive.printer_id == printer_id,
                 PrintArchive.status == "printing",
+                PrintArchive.id.not_in(archives_with_open_runs),
             )
+            .execution_options(**{INCLUDE_OPEN_RUNS: True})
         )
-        active = list(result.scalars().all())
+        active.extend(result.scalars().all())
 
     if not active:
         return 0
@@ -7843,6 +7893,10 @@ async def reconcile_stale_active_prints(printer_id: int) -> int:
             "raw_data": state.raw_data or {},
             "_reconciled": True,
         }
+        if isinstance(archive, _RunOfArchive):
+            # The archive it closes, which its names and printer may not find:
+            # another printer's plate of it can have started since.
+            payload["_reconciled_archive_id"] = archive.id
         if outcome != "aborted":
             # The printer's progress and layer are this job's, so a partial
             # run is charged what it printed rather than nothing.
@@ -8881,6 +8935,20 @@ async def _close_out_print(printer_id: int, data: dict):
     # the search matches names by substring.
     bound_archive_id = archive_id
 
+    if not archive_id and data.get("_reconciled") and data.get("_reconciled_archive_id"):
+        # The reconcile names the archive whose run on this printer it
+        # closes. Only while that run is open: the print's real end may have
+        # closed it meanwhile, or closed the print without managing to write
+        # it, and either way there is nothing left to end twice.
+        if await _archive_of_open_run(printer_id) != data["_reconciled_archive_id"]:
+            logger.info(
+                "[RECONCILE] Archive %s has no run open on printer %s any more; nothing to close",
+                data["_reconciled_archive_id"],
+                printer_id,
+            )
+            return
+        archive_id = data["_reconciled_archive_id"]
+
     ending_job = str(data.get("subtask_id") or "").strip()
     if not archive_id and ending_job not in ("", "0"):
         # Not bound (a restart since the start): this job's run, open on the
@@ -9630,6 +9698,8 @@ async def _close_out_print(printer_id: int, data: dict):
                     reconciled=bool(data.get("_reconciled")) and not data.get("_reconciled_completed_at"),
                     # Its time is the estimate or nothing, never measured (#3310).
                     duration_unmeasured=bool(data.get("_reconciled")),
+                    # The real end of a print stopped while its printer was offline.
+                    replaces_offline_stop=True,
                 )
                 await record_archive_wear(db, archive, run_entry)
                 await db.commit()

backend/app/models/print_log.py

@@ -97,6 +97,9 @@ class PrintLogEntry(Base):
     bambuddy_forced_timelapse: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
     notes: Mapped[str | None] = mapped_column(Text, nullable=True)
     quantity: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # Ended by a stop that could not reach the printer: logged without word
+    # from it. The printer's real end, if it still comes, replaces it.
+    ended_offline: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
 
 
 @event.listens_for(Session, "do_orm_execute")

backend/app/services/print_log.py

@@ -54,6 +54,8 @@ async def write_log_entry(
     created_by_username: str | None = None,
     reconciled: bool = False,
     duration_unmeasured: bool = False,
+    ended_offline: bool = False,
+    replaces_offline_stop: bool = False,
 ) -> PrintLogEntry:
     """Write a print log entry.
 
@@ -70,10 +72,17 @@ async def write_log_entry(
     ``duration_unmeasured`` marks a run closed at reconnect, whose times were
     not measured (#3310): ``completed_at`` is its start plus the slicer's
     estimate, or the reconnect. The time accuracy leaves it out.
+
+    ``ended_offline`` marks a run logged by a stop that could not reach the
+    printer. A print's end passes ``replaces_offline_stop``: it may then fill
+    such a run of its printer, the printer's latest, which its archive's start
+    (another plate's since, maybe) and its closed queue item no longer find.
     """
     entry = await _entry_of_this_run(db, archive_id, printer_id, started_at)
+    if entry is None and replaces_offline_stop and archive_id is not None and printer_id is not None:
+        entry = await _stopped_while_offline(db, archive_id, printer_id)
     filling_open_run = entry is not None and entry.status == RUN_OPEN
-    if entry is not None and filling_open_run and entry.started_at is not None:
+    if entry is not None and entry.started_at is not None:
         # The run's own start. The archive's is its latest run's: another
         # plate of it may have started on another printer since.
         started_at = entry.started_at
@@ -108,6 +117,7 @@ async def write_log_entry(
         "thumbnail_path": thumbnail_path,
         "created_by_id": created_by_id,
         "created_by_username": created_by_username,
+        "ended_offline": True if ended_offline else None,
     }
     if filling_open_run:
         # When the run was logged, as a run written at its end always was:
@@ -286,6 +296,23 @@ async def _entry_of_this_run(
     return await _open_run_on(db, archive_id, printer_id)
 
 
+async def _stopped_while_offline(db: AsyncSession, archive_id: int, printer_id: int) -> PrintLogEntry | None:
+    """The run a print's real end replaces after a stop logged while the
+    printer was offline: the printer's latest run, when it is of this
+    archive and was ended that way. A printer prints one thing at a time, so
+    any later run there, of any file, means that print is long over."""
+    latest = await db.scalar(
+        select(PrintLogEntry)
+        .where(PrintLogEntry.printer_id == printer_id)
+        .order_by(PrintLogEntry.started_at.desc().nulls_last(), PrintLogEntry.id.desc())
+        .limit(1)
+        .execution_options(**{INCLUDE_OPEN_RUNS: True})
+    )
+    if latest is not None and latest.archive_id == archive_id and latest.ended_offline:
+        return latest
+    return None
+
+
 async def _open_run_on(db: AsyncSession, archive_id: int, printer_id: int) -> PrintLogEntry | None:
     """The run of the archive open on the printer: there is one at most, a
     printer prints one thing at a time (``open_run``)."""

backend/tests/integration/test_run_lifecycle.py

@@ -891,3 +891,289 @@ async def test_a_later_plates_end_leaves_no_printing_archive_for_its_reconnect(
     ):
         assert await reconcile_stale_active_prints(other) == 0
     assert [r.printer_id for r in await _runs(db_session, farm.archive)] == [other]
+
+
+async def _reconcile(test_engine, printer_id: int, state) -> int:
+    """The reconnect's reconcile on ``printer_id`` against the test database,
+    its synthesised ends run one background task at a time."""
+    from unittest.mock import patch
+
+    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+    from backend.app.main import reconcile_stale_active_prints
+
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    payloads: list[tuple[int, dict]] = []
+
+    async def capture(pid, payload):
+        payloads.append((pid, payload))
+
+    with (
+        patch("backend.app.main.printer_manager.get_status", return_value=state),
+        patch("backend.app.main.async_session", maker),
+        patch("backend.app.main.on_print_complete", side_effect=capture),
+    ):
+        reconciled = await reconcile_stale_active_prints(printer_id)
+    # The ends it synthesised, run as on_print_complete runs them.
+    for pid, payload in payloads:
+        await _complete(
+            test_engine,
+            pid,
+            payload["status"],
+            name=payload["subtask_name"],
+            extra={k: v for k, v in payload.items() if k not in ("status", "subtask_name")},
+        )
+    return reconciled
+
+
+async def test_the_reconnect_closes_its_printers_run_of_a_shared_archive(
+    test_engine, db_session, farm, printer_factory
+):
+    """Plate 1's printer reconnects with its print over; the archive names
+    the other printer, whose plate still prints. Plate 1's run is closed,
+    plate 2's is left running."""
+    from unittest.mock import MagicMock
+
+    other = await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    main_module._active_prints.clear()  # the restart
+    finished = MagicMock(connected=True, state="FINISH", subtask_id="", progress=100, layer_num=0)
+
+    assert await _reconcile(test_engine, farm.printer, finished) == 1
+
+    runs = {r.printer_id: r.status for r in await _all_runs(db_session, farm.archive)}
+    assert runs[other] == "printing"
+    assert runs[farm.printer] != "printing"
+
+
+async def test_the_reconnect_leaves_a_run_still_printing_alone(test_engine, db_session, farm, printer_factory):
+    """The other printer reconnects mid-print, its own job on it: nothing to close."""
+    from unittest.mock import MagicMock
+
+    from sqlalchemy import update
+
+    other = await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    await db_session.execute(
+        update(PrintLogEntry)
+        .where(PrintLogEntry.printer_id == other)
+        .values(subtask_id="job-q")
+        .execution_options(include_open_runs=True)
+    )
+    await db_session.commit()
+    running = MagicMock(
+        connected=True, state="RUNNING", subtask_id="job-q", subtask_name=SUBTASK, progress=40, layer_num=3
+    )
+
+    assert await _reconcile(test_engine, other, running) == 0
+    assert {r.printer_id: r.status for r in await _all_runs(db_session, farm.archive)}[other] == "printing"
+
+
+async def test_a_real_end_after_an_offline_stop_is_one_run_on_a_shared_archive(
+    test_engine, db_session, farm, printer_factory, async_client
+):
+    """Plate 1 is stopped while its printer is offline, after plate 2 started
+    elsewhere (the archive has plate 2's start). Its printer then reports the
+    real end: it replaces the stop, it is not a second run."""
+    from unittest.mock import AsyncMock, patch
+
+    item = PrintQueueItem(printer_id=farm.printer, archive_id=farm.archive, status="printing", position=1, plate_id=1)
+    db_session.add(item)
+    await db_session.commit()
+    item_id = item.id
+    other = await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    own_start = {r.printer_id: r.started_at for r in await _all_runs(db_session, farm.archive)}[farm.printer]
+
+    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()
+        assert (await async_client.post(f"/api/v1/queue/{item_id}/stop")).status_code == 200
+    main_module._active_prints[(farm.printer, FILE)] = farm.archive
+
+    await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+
+    runs = [(r.printer_id, r.status) for r in await _all_runs(db_session, farm.archive)]
+    assert sorted(runs) == sorted([(farm.printer, "completed"), (other, "printing")])
+    ended = {r.printer_id: r for r in await _all_runs(db_session, farm.archive)}[farm.printer]
+    assert ended.started_at == own_start, "timed from its own start, not the other plate's"
+
+
+async def test_a_stale_archive_cancelled_at_the_next_start_keeps_other_printers_bound(farm):
+    """The stale-cancel at a print start drops the archive's bindings on its
+    own printer only: another printer printing a plate of it keeps its own."""
+    import ast
+    from pathlib import Path
+
+    source = Path(main_module.__file__).read_text()
+    tree = ast.parse(source)
+    unbinds = [
+        ast.unparse(node)
+        for node in ast.walk(tree)
+        if isinstance(node, ast.ListComp) and "existing_archive.id" in ast.unparse(node)
+    ]
+    assert unbinds and all("printer_id" in comp for comp in unbinds), unbinds
+
+
+async def test_the_reconnect_closes_a_finished_print_once(test_engine, db_session, farm, printer_factory):
+    """The later plate's printer reconnects with it over. The archive names
+    that printer and still says printing; its run is what is closed, once,
+    not the archive a second time as if it had no run."""
+    from unittest.mock import MagicMock
+
+    other = await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    main_module._active_prints.clear()
+    finished = MagicMock(connected=True, state="FINISH", subtask_id="", progress=100, layer_num=0)
+
+    assert await _reconcile(test_engine, other, finished) == 1
+
+
+async def test_the_reconnect_judges_each_print_by_its_own_job(test_engine, db_session, farm, printer_factory):
+    """Plate 1's printer reconnects running its job; the archive holds the
+    other plate's job id. Plate 1 is not taken for a print that changed."""
+    from unittest.mock import MagicMock
+
+    from sqlalchemy import update
+
+    await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    await db_session.execute(
+        update(PrintLogEntry)
+        .where(PrintLogEntry.printer_id == farm.printer)
+        .values(subtask_id="job-p")
+        .execution_options(include_open_runs=True)
+    )
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.subtask_id = "job-q"
+    await db_session.commit()
+    running = MagicMock(
+        connected=True, state="RUNNING", subtask_id="job-p", subtask_name=SUBTASK, progress=40, layer_num=3
+    )
+
+    assert await _reconcile(test_engine, farm.printer, running) == 0
+
+
+async def test_an_end_does_not_take_over_an_earlier_stop_of_the_file(test_engine, db_session, farm):
+    """Stopped from the queue while the printer was online (the same status and
+    reason an offline stop logs), then printed again: the reprint's end
+    writes its own run. Only a stop logged while the printer was offline is
+    the printer's real end to replace."""
+    from datetime import datetime
+
+    db_session.add(
+        PrintLogEntry(
+            archive_id=farm.archive,
+            printer_id=farm.printer,
+            status="cancelled",
+            failure_reason="userCancelled",
+            started_at=datetime(2026, 10, 29, 8),
+            completed_at=datetime(2026, 10, 29, 9),
+        )
+    )
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.status = "printing"
+    archive.started_at = datetime(2026, 10, 30, 8)  # printing since before runs were opened
+    await db_session.commit()
+    main_module._active_prints[(farm.printer, FILE)] = farm.archive
+
+    await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+
+    runs = sorted((r.started_at.day, r.status) for r in await _runs(db_session, farm.archive))
+    assert runs == [(29, "cancelled"), (30, "completed")]
+
+
+async def test_a_second_offline_stop_is_a_run_of_its_own(test_engine, db_session, farm, async_client):
+    """An earlier offline stop of the file on this printer is not this stop."""
+    from datetime import datetime
+    from unittest.mock import AsyncMock, patch
+
+    db_session.add(
+        PrintLogEntry(
+            archive_id=farm.archive,
+            printer_id=farm.printer,
+            status="cancelled",
+            failure_reason="userCancelled",
+            ended_offline=True,
+            started_at=datetime(2026, 10, 29, 8),
+        )
+    )
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.status = "printing"
+    archive.started_at = datetime(2026, 10, 30, 8)
+    item = PrintQueueItem(printer_id=farm.printer, archive_id=farm.archive, status="printing", position=1)
+    db_session.add(item)
+    await db_session.commit()
+    item_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()
+        assert (await async_client.post(f"/api/v1/queue/{item_id}/stop")).status_code == 200
+
+    assert len(await _runs(db_session, farm.archive)) == 2
+
+
+async def test_a_reconcile_after_the_real_end_closes_nothing(test_engine, db_session, farm, printer_factory):
+    """The reconcile read the printer's runs while plate 1 still printed; the
+    real end closed it before the synthesised one ran. That one ends nothing."""
+    from unittest.mock import MagicMock, patch
+
+    from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+    from backend.app.main import reconcile_stale_active_prints
+
+    await _two_printers_printing(test_engine, db_session, farm, printer_factory)
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    payloads: list[dict] = []
+
+    async def capture(pid, payload):
+        payloads.append(payload)
+
+    finished = MagicMock(connected=True, state="FINISH", subtask_id="", progress=100, layer_num=0)
+    with (
+        patch("backend.app.main.printer_manager.get_status", return_value=finished),
+        patch("backend.app.main.async_session", maker),
+        patch("backend.app.main.on_print_complete", side_effect=capture),
+    ):
+        await reconcile_stale_active_prints(farm.printer)
+    await _complete(test_engine, farm.printer, "completed", name=SUBTASK)  # the real end, first
+    [payload] = payloads
+
+    await _complete(
+        test_engine,
+        farm.printer,
+        payload["status"],
+        name=payload["subtask_name"],
+        extra={k: v for k, v in payload.items() if k not in ("status", "subtask_name")},
+    )
+
+    mine = [r.status for r in await _all_runs(db_session, farm.archive) if r.printer_id == farm.printer]
+    assert mine == ["completed"]
+
+
+async def test_a_stale_archive_cancelled_at_the_next_start_leaves_no_run_open(test_engine, db_session, farm):
+    """A print whose end was never seen, its archive cancelled as stale when
+    the next print starts: its run goes too, even when the new print's own
+    run is not opened, or the next reconnect would close it a second time."""
+    from datetime import datetime, timedelta
+    from unittest.mock import AsyncMock, patch
+
+    started = datetime.utcnow() - timedelta(hours=3)
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.status = "printing"
+    archive.started_at = started
+    archive.created_at = started
+    db_session.add(
+        PrintLogEntry(archive_id=farm.archive, printer_id=farm.printer, status="printing", started_at=started)
+    )
+    await db_session.commit()
+
+    with patch("backend.app.main._open_run_for", AsyncMock()):
+        await _drive_print_start(
+            test_engine, SimpleNamespace(id=farm.printer), download_ok=False, state_fields={"progress": 0.2}
+        )
+
+    db_session.expire_all()
+    archive = await db_session.get(PrintArchive, farm.archive)
+    assert archive.status == "cancelled"
+    assert await _all_runs(db_session, farm.archive) == []

backend/tests/unit/test_forget_printer.py

@@ -135,10 +135,12 @@ async def test_forget_printer_clears_main_and_keeps_other_printers():
         assert lost == []
         assert (GONE, 100) not in main._print_ams_mappings
         assert (GONE, 200) not in main._print_cost_center_ids
-        assert main._print_plate_ids == {(KEPT, 200): 3}
+        assert (GONE, 200) not in main._print_plate_ids
+        assert main._print_plate_ids.get((KEPT, 200)) == 3
         await asyncio.sleep(0)
         assert all(task.cancelled() for task in pending.values())
     finally:
+        main._print_plate_ids.pop((KEPT, 200), None)
         for entries in (gone, kept):
             for name, (key, value) in entries.items():
                 target = getattr(main, name)

backend/tests/unit/test_reconcile_stale_active_prints.py

@@ -217,6 +217,7 @@ class TestReconcileStaleActivePrints:
             mock_pm.get_status.return_value = _state("IDLE")
             with patch("backend.app.main.async_session") as mock_session:
                 session_ctx = AsyncMock()
+                session_ctx.scalars = AsyncMock(return_value=MagicMock(all=lambda: []))  # no open runs
                 session_ctx.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [])))
                 mock_session.return_value.__aenter__.return_value = session_ctx
                 count = await reconcile_stale_active_prints(printer_id=1)
@@ -231,6 +232,7 @@ class TestReconcileStaleActivePrints:
             mock_pm.get_status.return_value = _state("IDLE", subtask_id="", subtask_name="")
             with patch("backend.app.main.async_session") as mock_session:
                 session_ctx = AsyncMock()
+                session_ctx.scalars = AsyncMock(return_value=MagicMock(all=lambda: []))  # no open runs
                 session_ctx.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [stale])))
                 mock_session.return_value.__aenter__.return_value = session_ctx
                 with patch("backend.app.main.on_print_complete", new=AsyncMock()) as mock_complete:
@@ -254,6 +256,7 @@ class TestReconcileStaleActivePrints:
             mock_pm.get_status.return_value = _state("RUNNING", subtask_id="ABC123", subtask_name="ghost")
             with patch("backend.app.main.async_session") as mock_session:
                 session_ctx = AsyncMock()
+                session_ctx.scalars = AsyncMock(return_value=MagicMock(all=lambda: []))  # no open runs
                 session_ctx.execute = AsyncMock(
                     return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [healthy]))
                 )
@@ -278,6 +281,7 @@ class TestReconcileStaleActivePrints:
             mock_pm.get_status.return_value = _state("IDLE")
             with patch("backend.app.main.async_session") as mock_session:
                 session_ctx = AsyncMock()
+                session_ctx.scalars = AsyncMock(return_value=MagicMock(all=lambda: []))  # no open runs
                 session_ctx.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [a1, a2])))
                 mock_session.return_value.__aenter__.return_value = session_ctx
                 # First call raises, second call must still happen.
@@ -379,6 +383,7 @@ class TestReconcilePayload:
             mock_pm.get_status.return_value = state
             with patch("backend.app.main.async_session") as mock_session:
                 session_ctx = AsyncMock()
+                session_ctx.scalars = AsyncMock(return_value=MagicMock(all=lambda: []))  # no open runs
                 session_ctx.execute = AsyncMock(
                     return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [archive]))
                 )
@@ -435,6 +440,7 @@ async def test_the_queue_job_s_plate_reaches_the_estimate():
         mock_pm.get_status.return_value = _state("FINISH", subtask_id="ABC123", subtask_name="ghost")
         with patch("backend.app.main.async_session") as mock_session:
             session_ctx = AsyncMock()
+            session_ctx.scalars = AsyncMock(return_value=MagicMock(all=lambda: []))  # no open runs
             session_ctx.execute = AsyncMock(side_effect=[archives, plates])
             mock_session.return_value.__aenter__.return_value = session_ctx
             with (