Archives: write each print's run when it starts (archive redesign P1a)

- The print log row of a print is opened at every print start
  (expected print, restart resume, fallback, auto-archive) and filled
  in at its end, matched on archive, printer and start, or as the
  archive's run open on that printer
- Open runs are left out of every ORM read, update and delete unless
  asked for (include_open_runs), so no statistic, log or total counts
  a print that hasn't ended
- One open run per printer: a new start or an end without its archive
  drops what an unseen end left open; whole-archive deletes take it
- New run columns for the per-print state still on the archive
  (shadow mode: written alongside, read from the archive)
- Energy goes to the run that ended, not the archive's newest

backend/app/api/routes/archives.py

@@ -2430,7 +2430,13 @@ async def delete_archive(
 
         from backend.app.models.print_log import PrintLogEntry
 
-        await db.execute(sa_delete(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id))
+        # Its open run too (a print started on the printer, with no queue item
+        # to refuse the delete): it has no archive left to end on.
+        await db.execute(
+            sa_delete(PrintLogEntry)
+            .where(PrintLogEntry.archive_id == archive_id)
+            .execution_options(include_open_runs=True)
+        )
         await db.commit()
 
         if not await service.delete_archive(archive_id):

backend/app/api/routes/users.py

@@ -664,7 +664,13 @@ async def delete_user(
 
     # The print log keeps who started each run as long as the user exists, and
     # outlives them either way (ON DELETE SET NULL, which SQLite doesn't apply).
-    await db.execute(sql_update(PrintLogEntry).where(PrintLogEntry.created_by_id == user_id).values(created_by_id=None))
+    # Every run of theirs, the one printing now too.
+    await db.execute(
+        sql_update(PrintLogEntry)
+        .where(PrintLogEntry.created_by_id == user_id)
+        .values(created_by_id=None)
+        .execution_options(include_open_runs=True)
+    )
 
     # Drop API keys owned by this user. The model declares ON DELETE CASCADE
     # so Postgres handles this automatically, but SQLite ships with FK

backend/app/core/database.py

@@ -5916,6 +5916,39 @@ async def _run_migrations(conn):
     await _safe_execute(conn, "ALTER TABLE print_log_entries ADD COLUMN plate_id INTEGER")
     await _backfill_print_log_plate_ids(conn)
 
+    # Migration: what describes one print, on the run (archive redesign P1).
+    # Written alongside the archive's columns, so no backfill. Nullable, no
+    # default; the types are the ones each dialect gives the model's columns.
+    double = "REAL" if is_sqlite() else "DOUBLE PRECISION"
+    stamp = "DATETIME" if is_sqlite() else "TIMESTAMP"
+    for column, kind in (
+        ("subtask_id", "VARCHAR(64)"),
+        ("billing_run_id", "VARCHAR(36)"),
+        ("cost_center_id", "INTEGER"),
+        ("wallet_charge_skipped", "BOOLEAN"),
+        ("energy_start_kwh", double),
+        ("energy_start_at", stamp),
+        ("energy_start_plug_id", "INTEGER"),
+        ("energy_start_price", double),
+        ("user_verdict_source", "VARCHAR(16)"),
+        ("user_verdict_at", stamp),
+        ("confirm_requested", "BOOLEAN"),
+        ("confirm_token", "VARCHAR(64)"),
+        ("confirm_token_used_at", stamp),
+        ("photos", "JSON"),
+        ("timelapse_path", "VARCHAR(500)"),
+        ("timelapse_baseline", "JSON"),
+        ("timelapse_baseline_unread", "BOOLEAN"),
+        ("bambuddy_forced_timelapse", "BOOLEAN"),
+        ("notes", "TEXT"),
+        ("quantity", "INTEGER"),
+    ):
+        await _safe_execute(conn, f"ALTER TABLE print_log_entries ADD COLUMN {column} {kind}")
+    await _safe_execute(
+        conn,
+        "CREATE UNIQUE INDEX IF NOT EXISTS ix_print_log_entries_confirm_token ON print_log_entries (confirm_token)",
+    )
+
     await _migrate_auth_datetimes_to_timestamp(conn)
     await _migrate_add_missing_set_null_fks(conn)
     await _migrate_wallet_transactions_keep_deleted_users(conn)

backend/app/main.py

@@ -986,6 +986,42 @@ def _bind_active_print(
             _active_prints[(printer_id, name)] = archive_id
 
 
+async def _discard_open_runs_of(printer_id: int) -> None:
+    """Drop the printer's open runs when its print ended with no archive to
+    end them on. Its own session; never in the way of the completion."""
+    try:
+        from backend.app.services.print_log import discard_open_runs
+
+        async with async_session() as db:
+            await discard_open_runs(db, printer_id)
+            await db.commit()
+    except Exception as e:
+        logging.getLogger(__name__).warning("[RUN] Could not drop the open runs of printer %s: %s", printer_id, e)
+
+
+async def _open_run_for(archive_id: int, printer_id: int) -> 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
+    start: until its readers move over, the archive still carries the print.
+    The archive's start must be committed before."""
+    try:
+        from backend.app.models.archive import PrintArchive
+        from backend.app.models.printer import Printer
+        from backend.app.services.print_log import open_run
+
+        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(
+                db, archive, printer_id=printer_id, printer_name=printer_name, plate_id=_print_plate_ids.get(archive_id)
+            )
+            await db.commit()
+    except Exception as e:
+        logging.getLogger(__name__).warning("[RUN] Could not open the run of archive %s: %s", archive_id, e)
+
+
 def _mark_owned_if_dispatched(archive) -> None:
     """Remember a running print's archive as Bambuddy's when it carries a
     dispatch marker Bambuddy writes itself: ``billing_run_id`` (minted per
@@ -1196,6 +1232,12 @@ async def _record_energy_start(archive, printer_id: int, db, *, context: str = "
         except Exception as e:
             _logger.warning("[ENERGY] Could not read the electricity price for archive %s: %s", archive.id, e)
             archive.energy_start_price = None
+        try:
+            from backend.app.services.print_log import mirror_energy_start
+
+            await mirror_energy_start(db, archive, printer_id=printer_id)
+        except Exception as e:
+            _logger.warning("[RUN] Energy start not copied onto the run of archive %s: %s", archive.id, e)
         await db.commit()
         _logger.info(
             "[ENERGY] Recorded starting energy%s for archive %s from plug '%s': %s kWh",
@@ -5195,6 +5237,7 @@ async def on_print_start(printer_id: int, data: dict):
 
                 # Track as active print
                 _bind_active_print(printer_id, archive.id, archive.filename, filename, subtask_name)
+                await _open_run_for(archive.id, printer_id)
 
                 # Start timelapse session if external camera is enabled (#1353).
                 # Queue / VP-dispatched prints land here in the expected-archive
@@ -5352,6 +5395,7 @@ async def on_print_start(printer_id: int, data: dict):
             if subtask_match:
                 logger.info("Resuming archive %s on subtask_id match (%s)", existing_archive.id, subtask_id)
                 _bind_active_print(printer_id, existing_archive.id, existing_archive.filename, filename, subtask_name)
+                await _open_run_for(existing_archive.id, printer_id)
                 _mark_owned_if_dispatched(existing_archive)
                 if existing_archive.energy_start_kwh is None:
                     await _record_energy_start(existing_archive, printer_id, db, context="subtask-resume")
@@ -5446,6 +5490,7 @@ async def on_print_start(printer_id: int, data: dict):
                 if subtask_id and existing_archive.subtask_id != subtask_id:
                     existing_archive.subtask_id = subtask_id
                     await db.commit()
+                await _open_run_for(existing_archive.id, printer_id)
                 # Also set up energy tracking if not already tracked (#941: persisted column)
                 if existing_archive.energy_start_kwh is None:
                     await _record_energy_start(existing_archive, printer_id, db, context="existing-printing")
@@ -6104,6 +6149,7 @@ async def on_print_start(printer_id: int, data: dict):
 
                 # Track as active print
                 _bind_active_print(printer_id, fallback_archive.id, fallback_archive.filename, filename, subtask_name)
+                await _open_run_for(fallback_archive.id, printer_id)
 
                 # Record starting energy if smart plug available (#941: persisted column)
                 await _record_energy_start(fallback_archive, printer_id, db, context="fallback")
@@ -6259,6 +6305,7 @@ async def on_print_start(printer_id: int, data: dict):
 
                 # Track this active print (use both original filename and downloaded filename)
                 _bind_active_print(printer_id, archive_id, downloaded_filename, filename, subtask_name)
+                await _open_run_for(archive_id, printer_id)
 
                 logger.info("Created archive %s for %s", archive.id, downloaded_filename)
 
@@ -9237,6 +9284,7 @@ async def _close_out_print(printer_id: int, data: dict):
                 logger.warning("[NOTIFY-BG] Failed to send notification without archive: %s", e, exc_info=True)
 
         spawn_background_task(_notify_no_archive(), name="notify-no-archive")
+        await _discard_open_runs_of(printer_id)
         return
 
     log_timing("Archive lookup")
@@ -9403,12 +9451,20 @@ async def _close_out_print(printer_id: int, data: dict):
     log_timing("Finance charge update")
 
     # Write independent print log entry (separate table, never touches archives)
+    # The run's id, for the energy work below: the archive's newest run is
+    # the one started last, which with plates of it on two printers need not
+    # be the one that just ended.
+    completed_run_id: int | None = None
     try:
         async with async_session() as db:
             from backend.app.models.archive import PrintArchive
-            from backend.app.services.print_log import record_archive_wear, write_log_entry
+            from backend.app.services.print_log import discard_open_runs, record_archive_wear, write_log_entry
 
             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)
+                await db.commit()
             if archive:
                 # Back-fill created_by_id on reprint (#730): reprint reuses the
                 # source archive row rather than creating a new one, so an
@@ -9499,6 +9555,7 @@ async def _close_out_print(printer_id: int, data: dict):
                 )
                 await record_archive_wear(db, archive, run_entry)
                 await db.commit()
+                completed_run_id = run_entry.id
                 logger.info("[PRINT_LOG] Log entry written for archive %s", archive_id)
     except Exception as e:
         logger.warning("[PRINT_LOG] Failed to write log entry for archive %s: %s", archive_id, e)
@@ -9600,13 +9657,16 @@ async def _close_out_print(printer_id: int, data: dict):
                 # Backfill the latest PrintLogEntry for this archive with energy
                 # (write_log_entry above ran before this background task completed,
                 # so energy fields are still NULL on that row).
-                latest_run = await db.execute(
-                    select(PrintLogEntry)
-                    .where(PrintLogEntry.archive_id == archive_id)
-                    .order_by(PrintLogEntry.id.desc())
-                    .limit(1)
-                )
-                run_row = latest_run.scalar_one_or_none()
+                if completed_run_id is not None:
+                    run_row = await db.get(PrintLogEntry, completed_run_id)
+                else:
+                    latest_run = await db.execute(
+                        select(PrintLogEntry)
+                        .where(PrintLogEntry.archive_id == archive_id)
+                        .order_by(PrintLogEntry.id.desc())
+                        .limit(1)
+                    )
+                    run_row = latest_run.scalar_one_or_none()
                 if run_row is not None:
                     run_row.energy_kwh = energy_used
                     run_row.energy_cost = energy_cost_value

backend/app/models/print_log.py

@@ -1,10 +1,19 @@
 from datetime import datetime
 
-from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, func
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy import JSON, Boolean, DateTime, Float, ForeignKey, Integer, String, Text, event, func
+from sqlalchemy.orm import Mapped, ORMExecuteState, Session, mapped_column, with_loader_criteria
 
 from backend.app.core.database import Base
 
+# The status of a run that has started and not ended. Such a run is written
+# when its print starts (archive redesign) and is not a print that happened
+# yet: statistics, the print log and everything else that reads runs leave it
+# out, without each having to say so (see _hide_open_runs below).
+RUN_OPEN = "printing"
+# Execution option for the few that need open runs too: the print's own start
+# and end, which find and fill in the run.
+INCLUDE_OPEN_RUNS = "include_open_runs"
+
 
 class PrintLogEntry(Base):
     """Independent print log entry. Written when print events occur.
@@ -64,3 +73,48 @@ class PrintLogEntry(Base):
     created_by_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
     created_by_username: Mapped[str | None] = mapped_column(String(100))
     created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+
+    # What describes one print and today still lives on the archive, which
+    # the archive redesign moves here. Written alongside the archive's own
+    # columns for now; read from the archive until each part moves over.
+    subtask_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    billing_run_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
+    cost_center_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    wallet_charge_skipped: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
+    energy_start_kwh: Mapped[float | None] = mapped_column(Float, nullable=True)
+    energy_start_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    energy_start_plug_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    energy_start_price: Mapped[float | None] = mapped_column(Float, nullable=True)
+    user_verdict_source: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    user_verdict_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    confirm_requested: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
+    confirm_token: Mapped[str | None] = mapped_column(String(64), nullable=True, unique=True, index=True)
+    confirm_token_used_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    photos: Mapped[list | None] = mapped_column(JSON, nullable=True)
+    timelapse_path: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    timelapse_baseline: Mapped[list | None] = mapped_column(JSON, nullable=True)
+    timelapse_baseline_unread: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
+    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)
+
+
+@event.listens_for(Session, "do_orm_execute")
+def _hide_open_runs(state: ORMExecuteState) -> None:
+    """Every ORM read, update and delete of runs leaves open runs out, unless
+    it asks for them with ``execution_options(include_open_runs=True)``.
+
+    Fail-closed on purpose: some 25 places read runs -- statistics, failure
+    analysis, metrics, project and batch totals, the print log, repricing,
+    purge -- and a run that has not ended counted in any of them is a print
+    that did not happen. A new reader is covered without knowing about it.
+    """
+    if not (state.is_select or state.is_update or state.is_delete):
+        return
+    if state.execution_options.get(INCLUDE_OPEN_RUNS, False):
+        return
+    state.statement = state.statement.options(_FINISHED_RUNS_ONLY)
+
+
+# Built once: a new option per statement doubled the listener's cost.
+_FINISHED_RUNS_ONLY = with_loader_criteria(PrintLogEntry, PrintLogEntry.status != RUN_OPEN, include_aliases=True)

backend/app/services/archive.py

@@ -1056,7 +1056,12 @@ async def _null_print_log_thumbnail_paths(db: AsyncSession, archive_id: int) ->
 
     from backend.app.models.print_log import PrintLogEntry
 
-    await db.execute(sa_update(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id).values(thumbnail_path=None))
+    await db.execute(
+        sa_update(PrintLogEntry)
+        .where(PrintLogEntry.archive_id == archive_id)
+        .values(thumbnail_path=None)
+        .execution_options(include_open_runs=True)
+    )
 
 
 async def _delete_related_queue_items(db: AsyncSession, archive_id: int) -> int:

backend/app/services/archive_purge.py

@@ -290,7 +290,12 @@ class ArchivePurgeService:
 
                     from backend.app.models.print_log import PrintLogEntry
 
-                    await delete_db.execute(sa_delete(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id))
+                    # Its open run too: it has no archive left to end on.
+                    await delete_db.execute(
+                        sa_delete(PrintLogEntry)
+                        .where(PrintLogEntry.archive_id == archive_id)
+                        .execution_options(include_open_runs=True)
+                    )
                     await delete_db.commit()
                     if await service.delete_archive(archive_id):
                         deleted += 1

backend/app/services/print_log.py

@@ -7,10 +7,11 @@ import logging
 from collections.abc import Set as AbstractSet
 from datetime import datetime, timezone
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import delete, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from backend.app.models.print_log import PrintLogEntry
+from backend.app.models.print_log import INCLUDE_OPEN_RUNS, RUN_OPEN, PrintLogEntry
+from backend.app.utils.local_time import utcnow_naive
 
 logger = logging.getLogger(__name__)
 
@@ -70,6 +71,13 @@ async def write_log_entry(
     not measured (#3310): ``completed_at`` is its start plus the slicer's
     estimate, or the reconnect. The time accuracy leaves it out.
     """
+    entry = await _entry_of_this_run(db, archive_id, printer_id, started_at)
+    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:
+        # 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
+
     if reconciled:
         duration: int | None = 0
     elif started_at and completed_at:
@@ -101,11 +109,23 @@ async def write_log_entry(
         "created_by_id": created_by_id,
         "created_by_username": created_by_username,
     }
-    entry = await _entry_of_this_run(db, archive_id, printer_id, started_at)
+    if filling_open_run:
+        # When the run was logged, as a run written at its end always was:
+        # the date ranges of statistics and the print log go by it.
+        fields["created_at"] = utcnow_naive()
     if entry is None:
+        if archive_id is not None and started_at is not None and status != RUN_OPEN:
+            # Shadow check while the run's start moves over (archive redesign):
+            # every print start that binds an archive opens its run, so a run
+            # ending without one is a start path that does not yet.
+            logger.info("[RUN] Archive %s: no run was opened when the print started; written at its end", archive_id)
         entry = PrintLogEntry(**fields)
         db.add(entry)
     elif entry.status != status:
+        if entry.status == RUN_OPEN and None not in (entry.plate_id, plate_id) and entry.plate_id != plate_id:
+            logger.warning(
+                "[RUN] Archive %s: the run opened for plate %s ends as plate %s", archive_id, entry.plate_id, plate_id
+            )
         # The run ended differently from how it was first logged: a print
         # stopped while its printer was offline that finished after all. The
         # real outcome replaces the stop rather than counting the run twice.
@@ -187,6 +207,8 @@ async def _entry_of_this_run(
     archive_id: int | None,
     printer_id: int | None,
     started_at: datetime | None,
+    *,
+    exact: bool = False,
 ) -> PrintLogEntry | None:
     """The entry already written for this run, if any.
 
@@ -194,17 +216,136 @@ async def _entry_of_this_run(
     archive printing on two printers share the archive row (and can share its
     start time), and a reprint starts at a new time. The queue item is not
     part of it: a repeated completion finds the item already closed and
-    reports none. Without an archive, printer or start time a run can't be
-    told apart, so nothing is matched.
+    reports none. Without an archive or printer a run can't be told apart,
+    so nothing is matched.
+
+    A run opened at its start is also found as the archive's run open on the
+    printer, unless ``exact``: the archive's start is its latest run's, and
+    another plate of it may have started on another printer since.
     """
-    if archive_id is None or printer_id is None or started_at is None:
+    if archive_id is None or printer_id is None:
         return None
-    query = select(PrintLogEntry).where(
-        PrintLogEntry.archive_id == archive_id,
-        PrintLogEntry.printer_id == printer_id,
-        PrintLogEntry.started_at == started_at,
+    if started_at is None:
+        return None if exact else await _open_run_on(db, archive_id, printer_id)
+    query = (
+        select(PrintLogEntry)
+        .where(
+            PrintLogEntry.archive_id == archive_id,
+            PrintLogEntry.printer_id == printer_id,
+            PrintLogEntry.started_at == started_at,
+        )
+        # The run opened when the print started is the one this fills in.
+        .execution_options(**{INCLUDE_OPEN_RUNS: True})
+    )
+    entry = (await db.scalars(query.order_by(PrintLogEntry.id.desc()).limit(1))).first()
+    # The archive's start is its latest run's. With another plate of it
+    # started on another printer since, this printer's run is found as the
+    # one open there.
+    if entry is not None or exact:
+        return entry
+    return await _open_run_on(db, archive_id, printer_id)
+
+
+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``)."""
+    return await db.scalar(
+        select(PrintLogEntry)
+        .where(
+            PrintLogEntry.archive_id == archive_id,
+            PrintLogEntry.printer_id == printer_id,
+            PrintLogEntry.status == RUN_OPEN,
+        )
+        .order_by(PrintLogEntry.id.desc())
+        .limit(1)
+        .execution_options(**{INCLUDE_OPEN_RUNS: True})
     )
-    return (await db.scalars(query.order_by(PrintLogEntry.id.desc()).limit(1))).first()
+
+
+async def discard_open_runs(db: AsyncSession, printer_id: int) -> None:
+    """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."""
+    await db.execute(
+        delete(PrintLogEntry)
+        .where(PrintLogEntry.printer_id == printer_id, PrintLogEntry.status == RUN_OPEN)
+        .execution_options(**{INCLUDE_OPEN_RUNS: True})
+    )
+
+
+async def open_run(
+    db: AsyncSession,
+    archive,
+    *,
+    printer_id: int,
+    printer_name: str | None = None,
+    plate_id: int | None = None,
+) -> PrintLogEntry | None:
+    """The run of the print that just started on ``archive``, written now with
+    the open status; its end fills it in (``write_log_entry``).
+
+    Idempotent: a print start seen again (a restart resuming the print, a
+    repeated report) finds the run it opened, keyed as its end will key it, by
+    archive, printer and start time. The queue item running it, if any, gives
+    the dispatch's own billing run and cost centre. Does not commit.
+    """
+    from backend.app.models.print_queue import PrintQueueItem
+
+    if archive.started_at is None:
+        return None
+    # Only the run of this very start: one still open on the printer from an
+    # earlier start of the archive is of a print that ended unseen.
+    entry = await _entry_of_this_run(db, archive.id, printer_id, archive.started_at, exact=True)
+    if entry is not None:
+        return entry
+    # A printer prints one thing at a time: a run still open on it is of a
+    # print that ended without its end reaching it.
+    await discard_open_runs(db, printer_id)
+    item = await db.scalar(
+        select(PrintQueueItem)
+        .where(
+            PrintQueueItem.archive_id == archive.id,
+            PrintQueueItem.printer_id == printer_id,
+            PrintQueueItem.status == "printing",
+        )
+        .order_by(PrintQueueItem.id.desc())
+        .limit(1)
+    )
+    entry = PrintLogEntry(
+        status=RUN_OPEN,
+        archive_id=archive.id,
+        queue_item_id=item.id if item is not None else None,
+        print_name=archive.print_name,
+        plate_id=plate_id if plate_id is not None else archive.plate_id,
+        printer_name=printer_name,
+        printer_id=printer_id,
+        started_at=archive.started_at,
+        filament_type=archive.filament_type,
+        filament_color=archive.filament_color,
+        thumbnail_path=archive.thumbnail_path,
+        created_by_id=archive.created_by_id,
+        subtask_id=archive.subtask_id,
+        billing_run_id=(item.billing_run_id if item is not None else None) or archive.billing_run_id,
+        cost_center_id=(item.cost_center_id if item is not None else None) or archive.cost_center_id,
+        confirm_requested=archive.confirm_requested,
+        quantity=archive.quantity,
+    )
+    db.add(entry)
+    await db.flush()
+    return entry
+
+
+async def mirror_energy_start(db: AsyncSession, archive, *, printer_id: int) -> None:
+    """Copy the archive's energy start reading onto the run it was taken for.
+    Does not commit."""
+    entry = await _entry_of_this_run(db, archive.id, printer_id, archive.started_at)
+    if entry is None or entry.status != RUN_OPEN:
+        return
+    entry.energy_start_kwh = archive.energy_start_kwh
+    entry.energy_start_at = archive.energy_start_at
+    entry.energy_start_plug_id = archive.energy_start_plug_id
+    entry.energy_start_price = archive.energy_start_price
 
 
 async def record_archive_wear(db: AsyncSession, archive, entry: PrintLogEntry) -> None:

backend/tests/integration/test_external_print_confirmation_1898.py

@@ -76,7 +76,9 @@ class _StubArchiveService:
         return archive
 
 
-async def _drive_print_start(test_engine, printer, *, download_ok: bool, extra_patches=(), state_fields=None) -> dict:
+async def _drive_print_start(
+    test_engine, printer, *, download_ok: bool, extra_patches=(), state_fields=None, raw_data=None
+) -> dict:
     """Run ``on_print_start`` against the test database.
 
     ``download_ok`` picks the branch: False leaves the 3MF unreachable and the
@@ -142,7 +144,10 @@ async def _drive_print_start(test_engine, printer, *, download_ok: bool, extra_p
 
         from backend.app.main import on_print_start
 
-        await on_print_start(printer.id, {"filename": DISPATCH, "subtask_name": SUBTASK})
+        data = {"filename": DISPATCH, "subtask_name": SUBTASK}
+        if raw_data is not None:
+            data["raw_data"] = raw_data
+        await on_print_start(printer.id, data)
     return mocks
 
 

backend/tests/integration/test_run_lifecycle.py

@@ -194,6 +194,8 @@ async def test_a_print_started_on_the_printer_is_an_archive_with_one_run(test_en
     printer_id = (await printer_factory(name="P")).id
 
     await _drive_print_start(test_engine, SimpleNamespace(id=printer_id), download_ok=False)
+    archives = (await db_session.scalars(select(PrintArchive))).all()
+    assert [r.status for r in await _all_runs(db_session, archives[0].id)] == ["printing"], "opened at its start"
     await _complete(test_engine, printer_id, "completed", name=SUBTASK)
 
     archives = (await db_session.scalars(select(PrintArchive))).all()
@@ -293,3 +295,312 @@ async def test_each_run_keeps_its_own_verdict(test_engine, db_session, farm):
     assert [(r.plate_id, r.user_verdict) for r in runs] == [(1, "reject"), (2, None)]
     archive = await db_session.get(PrintArchive, farm.archive)
     assert archive.user_verdict is None, "the latest run is not judged yet"
+
+
+async def _all_runs(db_session, archive_id: int) -> list[PrintLogEntry]:
+    from backend.app.models.print_log import INCLUDE_OPEN_RUNS
+
+    db_session.expire_all()
+    return list(
+        (
+            await db_session.scalars(
+                select(PrintLogEntry)
+                .where(PrintLogEntry.archive_id == archive_id)
+                .order_by(PrintLogEntry.id)
+                .execution_options(**{INCLUDE_OPEN_RUNS: True})
+            )
+        ).all()
+    )
+
+
+async def _start_queued(test_engine, db_session, farm, *, plate: int, run_id: str) -> int:
+    item = PrintQueueItem(
+        printer_id=farm.printer,
+        archive_id=farm.archive,
+        status="printing",
+        position=1,
+        plate_id=plate,
+        created_by_id=farm.user,
+        cost_center_id=farm.center,
+        billing_run_id=run_id,
+    )
+    db_session.add(item)
+    (await db_session.get(PrintArchive, farm.archive)).plate_id = plate
+    await db_session.commit()
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=plate, cost_center_id=farm.center)
+    await _drive_print_start(test_engine, SimpleNamespace(id=farm.printer), download_ok=False)
+    return item.id
+
+
+async def test_the_run_is_written_when_the_print_starts(test_engine, db_session, farm, async_client):
+    """Open, it is in no statistic, log or count; its end fills in the same run."""
+    item = await _start_queued(test_engine, db_session, farm, plate=2, run_id="run-1")
+
+    [run] = await _all_runs(db_session, farm.archive)
+    assert (run.status, run.plate_id, run.queue_item_id, run.billing_run_id, run.cost_center_id, run.printer_id) == (
+        "printing",
+        2,
+        item,
+        "run-1",
+        farm.center,
+        farm.printer,
+    )
+    archive = await db_session.get(PrintArchive, farm.archive)
+    assert run.started_at == archive.started_at
+    assert await _runs(db_session, farm.archive) == [], "an open run is not a print that happened"
+    log = (await async_client.get("/api/v1/print-log/")).json()
+    assert log["total"] == 0 and log["items"] == []
+    stats = (await async_client.get("/api/v1/archives/stats")).json()
+    assert stats["total_prints"] == 0
+
+    await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+
+    assert [(r.id, r.status, r.plate_id) for r in await _all_runs(db_session, farm.archive)] == [
+        (run.id, "completed", 2)
+    ]
+
+
+async def test_a_print_start_seen_again_opens_no_second_run(test_engine, db_session, farm):
+    """A restart resumes the running print: the run it opened stays the one."""
+    await _start_queued(test_engine, db_session, farm, plate=1, run_id="run-1")
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.subtask_id = "job-7"
+    await db_session.commit()
+    main_module._active_prints.clear()
+
+    await _drive_print_start(
+        test_engine,
+        SimpleNamespace(id=farm.printer),
+        download_ok=False,
+        state_fields={"subtask_id": "job-7", "progress": 50.0},
+    )
+
+    assert await db_session.scalar(select(func.count()).select_from(PrintArchive)) == 1, "resumed, not archived anew"
+    assert [r.status for r in await _all_runs(db_session, farm.archive)] == ["printing"]
+
+
+async def test_the_runs_energy_start_is_its_own(test_engine, db_session, farm):
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from backend.app.main import _record_energy_start
+
+    await _start_queued(test_engine, db_session, farm, plate=1, run_id="run-1")
+    plug = MagicMock(id=4)
+    plug.name = "Plug"
+    archive = await db_session.get(PrintArchive, farm.archive)
+    with (
+        patch("backend.app.main.energy_plug_candidates", AsyncMock(return_value=[plug])),
+        patch("backend.app.main.select_energy_reading", AsyncMock(return_value=(plug, {"total": 12.5}))),
+        patch("backend.app.services.energy_price.current_price", AsyncMock(return_value=0.3)),
+    ):
+        assert await _record_energy_start(archive, farm.printer, db_session)
+
+    [run] = await _all_runs(db_session, farm.archive)
+    assert (run.energy_start_kwh, run.energy_start_plug_id, run.energy_start_price) == (12.5, 4, 0.3)
+    archive = await db_session.get(PrintArchive, farm.archive)
+    assert run.energy_start_at == archive.energy_start_at
+
+
+async def test_an_open_run_is_left_out_of_reads_updates_and_deletes(db_session, farm):
+    from sqlalchemy import delete, update
+
+    from backend.app.models.print_log import INCLUDE_OPEN_RUNS
+
+    db_session.add_all(
+        [
+            PrintLogEntry(archive_id=farm.archive, status="completed"),
+            PrintLogEntry(archive_id=farm.archive, status="printing"),
+        ]
+    )
+    await db_session.commit()
+
+    assert await db_session.scalar(select(func.count(PrintLogEntry.id))) == 1
+    await db_session.execute(update(PrintLogEntry).values(notes="edited"))
+    await db_session.execute(delete(PrintLogEntry))
+    await db_session.commit()
+
+    left = (
+        await db_session.execute(
+            select(PrintLogEntry.status, PrintLogEntry.notes).execution_options(**{INCLUDE_OPEN_RUNS: True})
+        )
+    ).all()
+    assert left == [("printing", None)]
+
+
+@pytest.mark.parametrize("by", ["subtask", "name"])
+async def test_a_print_running_since_before_the_upgrade_gets_its_run(test_engine, db_session, farm, by):
+    """Bambuddy restarts into the new version mid-print: the archive is
+    printing and has no run; resuming the print opens it."""
+    from datetime import datetime, timedelta
+
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.status = "printing"
+    started = datetime.utcnow().replace(microsecond=0) - timedelta(minutes=20)
+    archive.started_at = started
+    archive.subtask_id = "job-7"
+    await db_session.commit()
+
+    await _drive_print_start(
+        test_engine,
+        SimpleNamespace(id=farm.printer),
+        download_ok=False,
+        state_fields={"progress": 40.0},
+        raw_data={"subtask_id": "job-7"} if by == "subtask" else None,
+    )
+
+    assert await db_session.scalar(select(func.count()).select_from(PrintArchive)) == 1, "resumed, not archived anew"
+    runs = await _all_runs(db_session, farm.archive)
+    assert [(r.status, r.subtask_id) for r in runs] == [("printing", "job-7")]
+    assert runs[0].started_at == started
+
+
+async def test_a_downloaded_print_opens_its_run(test_engine, db_session, printer_factory):
+    """Started on the printer, its 3MF fetched: the auto-archive path."""
+    printer_id = (await printer_factory(name="P")).id
+
+    await _drive_print_start(test_engine, SimpleNamespace(id=printer_id), download_ok=True)
+
+    archives = (await db_session.scalars(select(PrintArchive))).all()
+    assert len(archives) == 1
+    assert [r.status for r in await _all_runs(db_session, archives[0].id)] == ["printing"]
+
+
+async def test_two_plates_on_two_printers_at_once_leave_no_open_run(test_engine, db_session, farm, printer_factory):
+    """The second start moves the archive's start to its own; the first
+    printer's end still fills its own run, with its own start.
+
+    The second printer's end finds the archive no longer printing and logs
+    nothing, as before runs were opened at the start: the archive holds one
+    print's state. Its open run goes with that end. Moving the print's state
+    onto the run (P1b) is what gives it its own record."""
+    other = (await printer_factory(name="Q")).id
+    register_expected_print(farm.printer, FILE, farm.archive, plate_id=1)
+    await _drive_print_start(test_engine, SimpleNamespace(id=farm.printer), download_ok=False)
+    [first] = await _all_runs(db_session, farm.archive)
+    first_start = first.started_at
+    await asyncio.sleep(1.1)  # the second start is a later second
+    register_expected_print(other, FILE, farm.archive, plate_id=2)
+    await _drive_print_start(test_engine, SimpleNamespace(id=other), download_ok=False)
+
+    await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+    await _complete(test_engine, other, "completed", name=SUBTASK)
+
+    runs = await _all_runs(db_session, farm.archive)
+    assert [(r.id, r.printer_id, r.status, r.started_at) for r in runs] == [
+        (first.id, farm.printer, "completed", first_start)
+    ]
+    open_left = (
+        await db_session.scalars(
+            select(PrintLogEntry).where(PrintLogEntry.status == "printing").execution_options(include_open_runs=True)
+        )
+    ).all()
+    assert open_left == []
+
+
+async def test_a_run_is_dated_by_its_end(test_engine, db_session, farm):
+    """The print log and statistics filter runs by when they were logged,
+    which was always their end."""
+    from datetime import datetime, timedelta
+
+    await _start_queued(test_engine, db_session, farm, plate=1, run_id="run-1")
+    [run] = await _all_runs(db_session, farm.archive)
+    run.created_at = datetime(2026, 1, 1)  # opened long ago
+    await db_session.commit()
+
+    await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+
+    [run] = await _all_runs(db_session, farm.archive)
+    assert run.created_at > datetime.utcnow() - timedelta(minutes=5)
+
+
+async def test_a_run_left_open_by_an_unseen_end_is_dropped_at_the_next_start(test_engine, db_session, farm):
+    """A print whose end never reached its run (cancelled as stale at the next
+    start): as before runs opened at the start, it leaves no run."""
+    db_session.add(PrintLogEntry(archive_id=None, printer_id=farm.printer, status="printing"))
+    await db_session.commit()
+
+    await _start_queued(test_engine, db_session, farm, plate=1, run_id="run-1")
+
+    runs = (
+        await db_session.scalars(
+            select(PrintLogEntry)
+            .where(PrintLogEntry.printer_id == farm.printer)
+            .execution_options(include_open_runs=True)
+        )
+    ).all()
+    assert [(r.archive_id, r.status) for r in runs] == [(farm.archive, "printing")]
+
+
+async def test_a_print_whose_archive_was_deleted_leaves_no_run(test_engine, db_session, farm):
+    from sqlalchemy import update
+
+    await _start_queued(test_engine, db_session, farm, plate=1, run_id="run-1")
+    # Deleted while it printed (ON DELETE SET NULL takes the run's link).
+    await db_session.execute(update(PrintLogEntry).values(archive_id=None).execution_options(include_open_runs=True))
+    await db_session.delete(await db_session.get(PrintArchive, farm.archive))
+    await db_session.commit()
+
+    await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+
+    left = (await db_session.scalars(select(PrintLogEntry).execution_options(include_open_runs=True))).all()
+    assert left == []
+
+
+async def test_deleting_an_archive_with_its_stats_takes_its_open_run(db_session, farm, async_client):
+    db_session.add(PrintLogEntry(archive_id=farm.archive, printer_id=farm.printer, status="printing"))
+    await db_session.commit()
+
+    response = await async_client.delete(f"/api/v1/archives/{farm.archive}?purge_stats=true")
+    assert response.status_code in (200, 204), response.text
+
+    left = (await db_session.scalars(select(PrintLogEntry).execution_options(include_open_runs=True))).all()
+    assert left == []
+
+
+async def test_the_energy_goes_to_the_run_that_ended(test_engine, db_session, farm, printer_factory):
+    """Its run, not the archive's newest: a run opened earlier ends after a
+    later one of the archive was written."""
+    from datetime import datetime
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    await _start_queued(test_engine, db_session, farm, plate=1, run_id="run-1")
+    other = (await printer_factory(name="Q")).id
+    later = PrintLogEntry(
+        archive_id=farm.archive, printer_id=other, status="completed", started_at=datetime(2026, 10, 30, 9)
+    )
+    db_session.add(later)
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.energy_start_kwh = 10.0
+    await db_session.commit()
+    plug = MagicMock(id=1)
+    plug.name = "Plug"
+
+    with (
+        patch("backend.app.main.energy_plug_candidates", AsyncMock(return_value=[plug])),
+        patch("backend.app.main.select_energy_reading", AsyncMock(return_value=(plug, {"total": 10.5}))),
+        patch("backend.app.services.energy_price.print_energy_cost", AsyncMock(return_value=0.1)),
+        patch("backend.app.services.energy_price.current_price", AsyncMock(return_value=0.3)),
+    ):
+        await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+
+    runs = {r.printer_id: r.energy_kwh for r in await _all_runs(db_session, farm.archive)}
+    assert runs == {farm.printer: 0.5, other: None}
+
+
+async def test_a_reprint_after_an_unseen_end_opens_a_run_of_its_own(test_engine, db_session, farm):
+    """The archive's earlier run on this printer is still open (its end never
+    came); the new start is a new run, not that one carried on."""
+    from datetime import datetime
+
+    stale = PrintLogEntry(
+        archive_id=farm.archive, printer_id=farm.printer, status="printing", started_at=datetime(2026, 10, 1, 8)
+    )
+    db_session.add(stale)
+    await db_session.commit()
+
+    await _start_queued(test_engine, db_session, farm, plate=2, run_id="run-2")
+
+    [run] = await _all_runs(db_session, farm.archive)
+    archive = await db_session.get(PrintArchive, farm.archive)
+    assert run.started_at != datetime(2026, 10, 1, 8)
+    assert (run.started_at, run.plate_id) == (archive.started_at, 2)

backend/tests/unit/services/test_archive_cost_estimate_3261.py

@@ -65,6 +65,9 @@ async def _setup(db_session, printer_factory, archive_factory, *, spoolman=False
         status="printing",
         cost=14.18,
         with_run=with_run,
+        # An earlier run of the archive, ended; a run still printing is open
+        # and counts as none.
+        run_status="completed" if with_run else None,
         file_path="archive/print.3mf",
         filament_used_grams=grams,
         filament_type="PLA",