Archives: each print's energy from its own run (archive redesign P1b)

- A print's energy is worked out from its run's start reading, not the
  archive's, which another plate's start on another printer replaces
- An end without a written run gives no other run its energy
- A resume takes a new reading when the archive's is an earlier print's
- A restart resuming a print finds its run by the printer's job, so the
  run is kept after another plate of the archive started elsewhere

backend/app/main.py

@@ -1017,11 +1017,12 @@ async def _archive_of_open_run(printer_id: int, subtask_id: str | None = None) -
         return None
 
 
-async def _open_run_for(archive_id: int, printer_id: int) -> None:
+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
     start: until its readers move over, the archive still carries the print.
-    The archive's start must be committed before."""
+    The archive's start must be committed before. ``subtask_id`` is the
+    printer's job, which finds the run again at a resume."""
     try:
         from backend.app.models.archive import PrintArchive
         from backend.app.models.printer import Printer
@@ -1038,6 +1039,7 @@ async def _open_run_for(archive_id: int, printer_id: int) -> None:
                 printer_id=printer_id,
                 printer_name=printer_name,
                 plate_id=_print_plate_ids.get((printer_id, archive_id)),
+                subtask_id=subtask_id,
             )
             await db.commit()
     except Exception as e:
@@ -1222,6 +1224,21 @@ async def _kill_switch_notification_already_sent(task: asyncio.Task[bool] | None
         return False
 
 
+async def _run_needs_energy_start(archive, printer_id: int, db) -> bool:
+    """At a resume: whether the print's run still lacks its energy start
+    reading. The archive's may be an earlier print's. Commits a reading
+    found on the archive for this run onto it."""
+    try:
+        from backend.app.services.print_log import run_needs_energy_start
+
+        needs = await run_needs_energy_start(db, archive, printer_id=printer_id)
+        await db.commit()
+        return needs
+    except Exception as e:
+        logging.getLogger(__name__).warning("[RUN] Energy start of archive %s's run not checked: %s", archive.id, e)
+        return archive.energy_start_kwh is None
+
+
 async def _record_energy_start(archive, printer_id: int, db, *, context: str = "") -> bool:
     """Capture the smart plug lifetime counter on the archive at print start.
 
@@ -5286,7 +5303,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)
+                await _open_run_for(archive.id, printer_id, subtask_id)
 
                 # Start timelapse session if external camera is enabled (#1353).
                 # Queue / VP-dispatched prints land here in the expected-archive
@@ -5459,9 +5476,9 @@ 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)
+                await _open_run_for(existing_archive.id, printer_id, subtask_id)
                 _mark_owned_if_dispatched(existing_archive)
-                if existing_archive.energy_start_kwh is None:
+                if await _run_needs_energy_start(existing_archive, printer_id, db):
                     await _record_energy_start(existing_archive, printer_id, db, context="subtask-resume")
                 if not notification_sent:
                     archive_data = {
@@ -5560,9 +5577,9 @@ 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)
+                await _open_run_for(existing_archive.id, printer_id, subtask_id)
                 # Also set up energy tracking if not already tracked (#941: persisted column)
-                if existing_archive.energy_start_kwh is None:
+                if await _run_needs_energy_start(existing_archive, printer_id, db):
                     await _record_energy_start(existing_archive, printer_id, db, context="existing-printing")
                 # Send notification with archive data (existing archive)
                 if not notification_sent:
@@ -6221,7 +6238,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)
+                await _open_run_for(fallback_archive.id, printer_id, subtask_id)
 
                 # Record starting energy if smart plug available (#941: persisted column)
                 await _record_energy_start(fallback_archive, printer_id, db, context="fallback")
@@ -6377,7 +6394,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)
+                await _open_run_for(archive_id, printer_id, subtask_id)
 
                 logger.info("Created archive %s for %s", archive.id, downloaded_filename)
 
@@ -9737,15 +9754,21 @@ async def _close_out_print(printer_id: int, data: dict):
             logger.info("[ENERGY-BG] Starting energy calculation for archive %s", archive_id)
             async with async_session() as db:
                 from backend.app.models.archive import PrintArchive
+                from backend.app.models.print_log import PrintLogEntry
+                from backend.app.services.print_log import energy_start_of
 
                 archive = await db.get(PrintArchive, archive_id)
                 if archive is None:
                     logger.warning("[ENERGY-BG] Archive %s no longer exists", archive_id)
                     return
-                starting_kwh = archive.energy_start_kwh
-                if starting_kwh is None:
-                    logger.info("[ENERGY-BG] No start kWh recorded for archive %s", archive_id)
+                # The ended run's own start reading: the archive's is its
+                # latest start's, maybe another plate's on another printer.
+                run_row = await db.get(PrintLogEntry, completed_run_id) if completed_run_id is not None else None
+                start = energy_start_of(run_row, archive)
+                if start is None:
+                    logger.info("[ENERGY-BG] No start kWh recorded for archive %s's run", archive_id)
                     return
+                starting_kwh = start.kwh
 
                 candidates = await energy_plug_candidates(db, printer_id)
                 if not candidates:
@@ -9786,10 +9809,10 @@ async def _close_out_print(printer_id: int, data: dict):
                     energy_cost_value = round(
                         await print_energy_cost(
                             db,
-                            plug_id=archive.energy_start_plug_id,
-                            start_at=archive.energy_start_at,
+                            plug_id=start.plug_id,
+                            start_at=start.at,
                             start_kwh=starting_kwh,
-                            start_price=archive.energy_start_price,
+                            start_price=start.price,
                             end_kwh=energy["total"],
                             end_plug_id=plug.id,
                             price_now=await current_price(db),
@@ -9805,8 +9828,6 @@ async def _close_out_print(printer_id: int, data: dict):
                 # (#1378). Reprint energy lives in the matching PrintLogEntry below.
                 from sqlalchemy import func
 
-                from backend.app.models.print_log import PrintLogEntry
-
                 existing_runs = await db.scalar(
                     select(func.count(PrintLogEntry.id)).where(PrintLogEntry.archive_id == archive_id)
                 )
@@ -9816,19 +9837,9 @@ async def _close_out_print(printer_id: int, data: dict):
                     archive.energy_kwh = energy_used
                     archive.energy_cost = energy_cost_value
 
-                # 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).
-                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()
+                # The ended run's energy (write_log_entry above ran before this
+                # background task, so its energy fields are still NULL). With
+                # no run written, none: the archive's newest is another print's.
                 if run_row is not None:
                     run_row.energy_kwh = energy_used
                     run_row.energy_cost = energy_cost_value

backend/app/services/print_log.py

@@ -7,6 +7,7 @@ import logging
 import uuid
 from collections.abc import Set as AbstractSet
 from datetime import datetime, timezone
+from types import SimpleNamespace
 
 from sqlalchemy import delete, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -376,12 +377,14 @@ async def open_run(
     printer_id: int,
     printer_name: str | None = None,
     plate_id: int | None = None,
+    subtask_id: str | 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
+    repeated report) finds the run it opened: by the printer's job
+    (``subtask_id``) when it has one, else 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; a run without one gets its
     own billing run id, so each print of the archive is charged as its own.
@@ -391,6 +394,13 @@ async def open_run(
 
     if archive.started_at is None:
         return None
+    if subtask_id:
+        # The same job seen again: its run, though another plate of the
+        # archive may have started on another printer since and moved the
+        # archive's start.
+        running = await open_run_on_printer(db, printer_id, subtask_id=subtask_id)
+        if running is not None and running.archive_id == archive.id:
+            return running
     # 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)
@@ -441,10 +451,63 @@ async def mirror_energy_start(db: AsyncSession, archive, *, printer_id: int) ->
     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
+    for name in _ENERGY_START:
+        setattr(entry, name, getattr(archive, name))
+
+
+_ENERGY_START = ("energy_start_kwh", "energy_start_at", "energy_start_plug_id", "energy_start_price")
+
+
+def _reading_taken_for(archive, entry: PrintLogEntry) -> bool:
+    """Whether the archive's energy start reading was taken for this run: the
+    archive's latest start is this run's (another plate of it started since
+    takes the reading), and it was read at or after it (an earlier print's is
+    older). A reading without its time (taken before #1251) goes by the
+    start alone."""
+    if archive.energy_start_kwh is None:
+        return False
+    if entry.started_at is None or archive.started_at is None:
+        return False
+    if _utc(archive.started_at) != _utc(entry.started_at):
+        return False
+    return archive.energy_start_at is None or _utc(archive.energy_start_at) >= _utc(entry.started_at)
+
+
+async def run_needs_energy_start(db: AsyncSession, archive, *, printer_id: int) -> bool:
+    """Whether the print running on the archive still lacks its energy start
+    reading, at a resume. Its run's, or the archive's taken for it (copied
+    onto the run then), counts; the archive's of an earlier print does not.
+    Without a run, the archive's, as before. 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 archive.energy_start_kwh is None
+    if entry.energy_start_kwh is not None:
+        return False
+    if _reading_taken_for(archive, entry):
+        for name in _ENERGY_START:
+            setattr(entry, name, getattr(archive, name))
+        return False
+    return True
+
+
+def energy_start_of(entry: PrintLogEntry | None, archive) -> SimpleNamespace | None:
+    """The energy start reading of the run that ended: its own. Without one,
+    the archive's only when taken for this run (a run written only at its
+    end, or none written): another plate of the archive may have started on
+    another printer since and taken it."""
+    source = None
+    if entry is not None and entry.energy_start_kwh is not None:
+        source = entry
+    elif entry is None or _reading_taken_for(archive, entry):
+        source = archive
+    if source is None or source.energy_start_kwh is None:
+        return None
+    return SimpleNamespace(
+        kwh=source.energy_start_kwh,
+        at=source.energy_start_at,
+        plug_id=source.energy_start_plug_id,
+        price=source.energy_start_price,
+    )
 
 
 async def record_archive_wear(db: AsyncSession, archive, entry: PrintLogEntry) -> None:

backend/tests/integration/test_run_lifecycle.py

@@ -1370,3 +1370,158 @@ async def test_a_print_stopped_before_its_start_was_reported_frees_its_budget(te
 
     db_session.expire_all()
     assert (await db_session.get(BudgetReservation, hold_id)).status == "released"
+
+
+async def test_two_plates_on_two_printers_each_get_their_own_energy(test_engine, db_session, farm, printer_factory):
+    """Each start's reading stays with its run: plate 2's start on another
+    printer does not become plate 1's start."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    plugs = {}
+
+    async def candidates(db, printer_id):
+        plug = plugs.setdefault(printer_id, MagicMock(id=printer_id))
+        plug.name = f"Plug {printer_id}"
+        return [plug]
+
+    counters = iter([10.0, 20.0, 10.5, 20.3])
+
+    async def reading(found, read, db):
+        return found[0], {"total": next(counters)}
+
+    with (
+        patch("backend.app.main.energy_plug_candidates", side_effect=candidates),
+        patch("backend.app.main.select_energy_reading", side_effect=reading),
+        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)),
+    ):
+        # The start helper stubs the reading out; this test takes it.
+        real_reading = patch("backend.app.main._record_energy_start", main_module._record_energy_start)
+        other = (await printer_factory(name="Q")).id
+        for printer_id, plate in ((farm.printer, 1), (other, 2)):
+            register_expected_print(printer_id, FILE, farm.archive, plate_id=plate)
+            await _drive_print_start(
+                test_engine, SimpleNamespace(id=printer_id), download_ok=False, extra_patches=(real_reading,)
+            )
+            await asyncio.sleep(1.1)
+        await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+        await _complete(test_engine, other, "completed", name=SUBTASK)
+
+    runs = {r.printer_id: r.energy_kwh for r in await _runs(db_session, farm.archive)}
+    assert runs == {farm.printer: 0.5, other: pytest.approx(0.3)}
+
+
+async def test_a_resume_takes_a_reading_when_the_archive_has_an_earlier_prints(db_session, farm):
+    """The archive's reading is an earlier print's: the resumed run still
+    needs its own. One taken for this run is copied onto it instead."""
+    from datetime import datetime
+
+    from backend.app.main import _run_needs_energy_start
+
+    archive = await db_session.get(PrintArchive, farm.archive)
+    archive.started_at = datetime(2026, 10, 30, 10)
+    archive.energy_start_kwh = 5.0
+    archive.energy_start_at = datetime(2026, 10, 29, 10)
+    run = PrintLogEntry(
+        archive_id=farm.archive, printer_id=farm.printer, status="printing", started_at=archive.started_at
+    )
+    db_session.add(run)
+    await db_session.commit()
+    run_id = run.id
+
+    assert await _run_needs_energy_start(archive, farm.printer, db_session) is True
+
+    archive.energy_start_at = datetime(2026, 10, 30, 10, 0, 5)
+    await db_session.commit()
+    assert await _run_needs_energy_start(archive, farm.printer, db_session) is False
+    db_session.expire_all()
+    assert (await db_session.get(PrintLogEntry, run_id)).energy_start_kwh == 5.0
+
+
+async def test_an_end_whose_run_was_not_written_gives_no_other_run_its_energy(test_engine, db_session, farm):
+    """The archive's newest finished run is an earlier print's."""
+    from datetime import datetime
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from backend.app.services import print_log
+
+    earlier = PrintLogEntry(
+        archive_id=farm.archive, printer_id=farm.printer, status="completed", started_at=datetime(2026, 10, 29, 9)
+    )
+    db_session.add(earlier)
+    await db_session.commit()
+    earlier_id = earlier.id
+    await _start_queued(test_engine, db_session, farm, plate=1, run_id="run-1")
+    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)),
+        patch.object(print_log, "write_log_entry", AsyncMock(side_effect=ValueError("broken"))),
+    ):
+        await _complete(test_engine, farm.printer, "completed", name=SUBTASK)
+
+    db_session.expire_all()
+    assert (await db_session.get(PrintLogEntry, earlier_id)).energy_kwh is None
+
+
+async def test_a_run_without_its_own_reading_takes_no_later_plates():
+    """Its reading was not copied at its start; plate 2 of the archive
+    started on another printer since and took the archive's."""
+    from datetime import datetime
+
+    from backend.app.services.print_log import energy_start_of
+
+    t1, t2 = datetime(2026, 10, 30, 9), datetime(2026, 10, 30, 10)
+    run = PrintLogEntry(archive_id=1, printer_id=1, status="completed", started_at=t1)
+    archive = SimpleNamespace(
+        started_at=t2,
+        energy_start_kwh=20.0,
+        energy_start_at=datetime(2026, 10, 30, 10, 0, 5),
+        energy_start_plug_id=2,
+        energy_start_price=0.3,
+    )
+    assert energy_start_of(run, archive) is None
+    # The archive's latest start is the run's own: its reading is.
+    archive.started_at = t1
+    archive.energy_start_at = datetime(2026, 10, 30, 9, 0, 5)
+    assert energy_start_of(run, archive).kwh == 20.0
+
+
+async def test_a_restart_keeps_the_first_printers_run_while_another_plate_prints(
+    test_engine, db_session, farm, printer_factory
+):
+    """Resumed by its job after the other plate started: the same run, with
+    its own start, billing id and energy reading, not one stamped with the
+    other plate's start."""
+    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", energy_start_kwh=10.0)
+        .execution_options(include_open_runs=True)
+    )
+    await db_session.commit()
+    [before] = [r for r in await _all_runs(db_session, farm.archive) if r.printer_id == farm.printer]
+    kept = (before.id, before.started_at, before.billing_run_id)
+    main_module._active_prints.clear()  # the restart
+
+    await _drive_print_start(
+        test_engine,
+        SimpleNamespace(id=farm.printer),
+        download_ok=False,
+        state_fields={"progress": 40.0},
+        raw_data={"subtask_id": "job-p"},
+    )
+
+    [after] = [r for r in await _all_runs(db_session, farm.archive) if r.printer_id == farm.printer]
+    assert (after.id, after.started_at, after.billing_run_id) == kept
+    assert after.energy_start_kwh == 10.0