From 40610e2798e78c1d8daa5030cbd670097ecc9a07 Mon Sep 17 00:00:00 2001
From: ichwars <drothe2012@gmail.com>
Date: Sun, 6 Sep 2026 21:52:47 +0200
Subject: [PATCH 1/2] fix(smart-plugs): select printer power deterministically

---
 backend/app/api/routes/smart_plugs.py         |  42 ++-
 backend/app/main.py                           |  34 +--
 backend/app/services/print_scheduler.py       |  21 +-
 backend/app/services/smart_plug_selection.py  | 145 +++++++++
 .../test_issue_140_smart_plug_selection.py    | 165 ++++++++++
 .../test_issue_140_smart_plug_selection.py    | 282 ++++++++++++++++++
 tools/check_source_size_budget.py             |   6 +-
 7 files changed, 634 insertions(+), 61 deletions(-)
 create mode 100644 backend/app/services/smart_plug_selection.py
 create mode 100644 backend/tests/integration/test_issue_140_smart_plug_selection.py
 create mode 100644 backend/tests/unit/test_issue_140_smart_plug_selection.py

diff --git a/backend/app/api/routes/smart_plugs.py b/backend/app/api/routes/smart_plugs.py
index 063b3ed29c..6d642472f5 100644
--- a/backend/app/api/routes/smart_plugs.py
+++ b/backend/app/api/routes/smart_plugs.py
@@ -41,6 +41,11 @@
 from backend.app.services.printer_manager import printer_manager
 from backend.app.services.rest_smart_plug import rest_smart_plug_service
 from backend.app.services.smart_plug_manager import smart_plug_manager
+from backend.app.services.smart_plug_selection import (
+    can_be_switched,
+    pick_power_plug,
+    plugs_for_printer,
+)
 from backend.app.services.tasmota import tasmota_service
 from backend.app.utils.local_time import utcnow_naive
 
@@ -183,25 +188,8 @@ async def get_smart_plug_by_printer(
     db: AsyncSession = Depends(get_db),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.SMART_PLUGS_READ),
 ):
-    """Get the main smart plug assigned to a printer.
-
-    When multiple plugs are assigned (e.g., a regular plug + script),
-    returns the main (non-script) plug for power control.
-    """
-    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-    plugs = result.scalars().all()
-
-    if not plugs:
-        return None
-
-    # If multiple plugs, prefer the non-script one (main power plug)
-    for plug in plugs:
-        is_script = plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script.")
-        if not is_script:
-            return plug
-
-    # All are scripts, return the first one
-    return plugs[0]
+    """Get the linked row that best represents the printer's power supply."""
+    return pick_power_plug(await plugs_for_printer(db, printer_id))
 
 
 @router.get("/by-printer/{printer_id}/scripts", response_model=list[SmartPlugResponse])
@@ -216,12 +204,20 @@ async def get_script_plugs_by_printer(
     show_on_printer_card enabled.
     Used to display action buttons alongside the main power plug.
     """
-    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-    plugs = result.scalars().all()
+    plugs = await plugs_for_printer(db, printer_id)
+    main_plug = pick_power_plug(plugs)
+    duplicate_id = main_plug.id if main_plug and can_be_switched(main_plug) else None
 
-    # Filter to HA entities with show_on_printer_card enabled
+    # A switchable main entity is already rendered in the Power row. Scripts
+    # remain here even in the script-only fallback so their one-click action is
+    # preserved.
     ha_entities = [
-        plug for plug in plugs if plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.show_on_printer_card
+        plug
+        for plug in plugs
+        if plug.plug_type == "homeassistant"
+        and plug.ha_entity_id
+        and plug.show_on_printer_card
+        and plug.id != duplicate_id
     ]
     return ha_entities
 
diff --git a/backend/app/main.py b/backend/app/main.py
index 6ff0bfb473..a46fb637ba 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -84,7 +84,6 @@
 from backend.app.core.database import async_session, engine, init_db
 from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
-from backend.app.models.smart_plug import SmartPlug
 from backend.app.services import (
     active_print_provenance as print_provenance,
     business_runtime,
@@ -123,6 +122,7 @@
     resolve_plate_id,
 )
 from backend.app.services.smart_plug_manager import smart_plug_manager
+from backend.app.services.smart_plug_selection import read_printer_energy
 from backend.app.services.spool_assignment_notifications import (
     notify_missing_spool_assignments_on_print_start,
 )
@@ -669,21 +669,19 @@ async def _record_energy_start(archive, printer_id: int, db, *, context: str = "
     """
     _logger = logging.getLogger(__name__)
     try:
-        plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-        plug = plug_result.scalar_one_or_none()
-        if not plug:
-            _logger.info("[ENERGY] No smart plug for printer %s (archive %s)", printer_id, archive.id)
-            return False
-        energy = await _get_plug_energy(plug, db)
-        if not energy or energy.get("total") is None:
-            _logger.warning("[ENERGY] No 'total' in energy response for archive %s", archive.id)
+        selected = await read_printer_energy(
+            db, printer_id, _get_plug_energy, log_prefix="ENERGY", context=f"archive {archive.id}"
+        )
+        if selected is None:
             return False
+        plug, energy = selected
         archive.energy_start_kwh = float(energy["total"])
         await db.commit()
         _logger.info(
-            "[ENERGY] Recorded starting energy%s for archive %s: %s kWh",
+            "[ENERGY] Recorded starting energy%s for archive %s from plug '%s': %s kWh",
             f" ({context})" if context else "",
             archive.id,
+            plug.name,
             energy["total"],
         )
         return True
@@ -4508,17 +4506,13 @@ async def _background_energy_calculation():
                     logger.info("[ENERGY-BG] No start kWh recorded for archive %s", archive_id)
                     return
 
-                plug_result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-                plug = plug_result.scalar_one_or_none()
-                if plug is None:
-                    logger.info("[ENERGY-BG] No smart plug for printer %s", printer_id)
-                    return
-
-                energy = await _get_plug_energy(plug, db)
-                logger.info("[ENERGY-BG] Energy response: %s", energy)
-                if not energy or energy.get("total") is None:
-                    logger.warning("[ENERGY-BG] No 'total' in energy response")
+                selected = await read_printer_energy(
+                    db, printer_id, _get_plug_energy, log_prefix="ENERGY-BG", context=f"archive {archive_id}"
+                )
+                if selected is None:
                     return
+                plug, energy = selected
+                logger.info("[ENERGY-BG] Energy response from plug '%s': %s", plug.name, energy)
 
                 energy_used = round(energy["total"] - starting_kwh, 4)
                 logger.info("[ENERGY-BG] Per-print energy: %s kWh", energy_used)
diff --git a/backend/app/services/print_scheduler.py b/backend/app/services/print_scheduler.py
index ff88395455..8d25350bc1 100644
--- a/backend/app/services/print_scheduler.py
+++ b/backend/app/services/print_scheduler.py
@@ -53,6 +53,10 @@
     supports_drying_while_printing,
 )
 from backend.app.services.smart_plug_manager import smart_plug_manager
+from backend.app.services.smart_plug_selection import (
+    pick_power_plug as select_power_plug,
+    plugs_for_printer,
+)
 from backend.app.utils.filename import derive_remote_filename
 from backend.app.utils.printer_models import is_gcode_compatible, normalize_printer_model
 from backend.app.utils.threemf_tools import extract_bed_temperature_from_3mf
@@ -600,11 +604,11 @@ def _claim_library_row(candidate: PrintQueueItem) -> None:
 
                     # If printer not connected, try to power on via smart plug
                     if not printer_connected:
-                        plugs = await self._get_smart_plugs(db, item.printer_id)
+                        plugs = await plugs_for_printer(db, item.printer_id)
                         auto_on_plugs = [p for p in plugs if p.auto_on and p.enabled]
                         if auto_on_plugs:
                             logger.info("Printer %s offline, attempting to power on via smart plug(s)", item.printer_id)
-                            primary_plug = self._pick_power_plug(auto_on_plugs)
+                            primary_plug = select_power_plug(auto_on_plugs) or auto_on_plugs[0]
                             powered_on = await self._power_on_and_wait(primary_plug, item.printer_id, db)
                             if powered_on:
                                 # Also turn on any remaining auto_on plugs (e.g., filter)
@@ -2439,19 +2443,6 @@ async def _stop_drying(self, printer_id: int):
                 printer_manager.send_drying_command(printer_id, ams_id, 0, 0, mode=0)
         self._drying_in_progress.pop(printer_id, None)
 
-    async def _get_smart_plugs(self, db: AsyncSession, printer_id: int) -> list[SmartPlug]:
-        """Get all smart plugs associated with a printer."""
-        result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-        return list(result.scalars().all())
-
-    @staticmethod
-    def _pick_power_plug(auto_on_plugs: list[SmartPlug]) -> SmartPlug:
-        """Pick the plug that actually powers the printer, falling back to first."""
-        for plug in auto_on_plugs:
-            if plug.controls_printer_power:
-                return plug
-        return auto_on_plugs[0]
-
     # Bundled defaults for preheat_filament_targets (#1468). Values are the
     # chamber-temperature recommendations BambuStudio ships for the matching
     # filament profile; users can override via Settings → Workflow → Preheat
diff --git a/backend/app/services/smart_plug_selection.py b/backend/app/services/smart_plug_selection.py
new file mode 100644
index 0000000000..ee0df3226f
--- /dev/null
+++ b/backend/app/services/smart_plug_selection.py
@@ -0,0 +1,145 @@
+"""Shared business rule for a printer's primary smart plug (#140).
+
+A printer may legitimately have several linked rows: its outlet, accessories,
+Home Assistant scripts, or a read-only MQTT meter.  Consumers must not depend
+on database row order when deciding which row represents printer power.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Awaitable, Callable, Iterable, Sequence
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.smart_plug import SmartPlug
+
+EnergyReader = Callable[[SmartPlug, AsyncSession | None], Awaitable[dict | None]]
+logger = logging.getLogger(__name__)
+
+
+def is_script_plug(plug: SmartPlug) -> bool:
+    """Return whether *plug* is a runnable HA script rather than a switch."""
+    entity_id = getattr(plug, "ha_entity_id", None)
+    return bool(getattr(plug, "plug_type", None) == "homeassistant" and entity_id and entity_id.startswith("script."))
+
+
+def can_be_switched(plug: SmartPlug) -> bool:
+    """Match the device kinds accepted by the manual control path."""
+    return not is_script_plug(plug) and getattr(plug, "plug_type", None) != "mqtt"
+
+
+def reports_power(plug: SmartPlug) -> bool:
+    """Whether configuration exposes a place to read current watts from."""
+    plug_type = getattr(plug, "plug_type", None)
+    if plug_type == "homeassistant":
+        return bool(getattr(plug, "ha_power_entity", None))
+    if plug_type == "mqtt":
+        return bool(getattr(plug, "mqtt_power_topic", None) or getattr(plug, "mqtt_topic", None))
+    if plug_type == "rest":
+        return bool(getattr(plug, "rest_power_path", None))
+    return True
+
+
+def power_plug_rank(plug: SmartPlug) -> tuple[bool, bool, bool, bool, bool, int]:
+    """Sort key for the row that represents the printer's power supply.
+
+    Capability comes first so the card never offers an unusable switch.  The
+    explicit printer-power relationship outranks display and metering details;
+    an accessory must not become the printer outlet merely because it reports
+    watts.  Nullable legacy flags rank behind explicit true values.  Lowest id
+    is the final stable tiebreaker.
+    """
+    plug_id = getattr(plug, "id", None)
+    return (
+        not can_be_switched(plug),
+        not bool(getattr(plug, "controls_printer_power", False)),
+        not bool(getattr(plug, "enabled", False)),
+        not bool(getattr(plug, "show_on_printer_card", False)),
+        not reports_power(plug),
+        plug_id if isinstance(plug_id, int) else 2**63 - 1,
+    )
+
+
+def _power_assignment(plug: SmartPlug) -> tuple[bool, bool, bool]:
+    """The identity-bearing part of the rank, excluding presentation details."""
+    return (
+        can_be_switched(plug),
+        bool(getattr(plug, "controls_printer_power", False)),
+        bool(getattr(plug, "enabled", False)),
+    )
+
+
+def rank_power_plugs(plugs: Iterable[SmartPlug]) -> list[SmartPlug]:
+    """Return linked plugs in deterministic primary-power order."""
+    return sorted(plugs, key=power_plug_rank)
+
+
+def pick_power_plug(plugs: Iterable[SmartPlug]) -> SmartPlug | None:
+    """Return the best printer-power row, retaining single-row fallbacks."""
+    return min(plugs, key=power_plug_rank, default=None)
+
+
+async def plugs_for_printer(
+    db: AsyncSession,
+    printer_id: int | None,
+) -> list[SmartPlug]:
+    """Load one printer's plugs and apply the shared business ordering."""
+    if printer_id is None:
+        return []
+    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id).order_by(SmartPlug.id))
+    return rank_power_plugs(result.scalars().all())
+
+
+async def select_energy_reading(
+    candidates: Sequence[SmartPlug],
+    read_energy: EnergyReader,
+    db: AsyncSession | None,
+) -> tuple[SmartPlug, dict] | None:
+    """Return the first ranked plug with a real lifetime counter.
+
+    A power-only response is not consumption.  Fallback is limited to rows
+    with the same printer-power assignment, so an accessory meter cannot be
+    billed merely because the actual outlet has no lifetime counter.
+    """
+    ranked = rank_power_plugs(candidates)
+    if not ranked:
+        return None
+
+    # Do not cross from the chosen printer-power relationship into an
+    # accessory merely because the accessory has a meter. Multiple rows with
+    # the same switchability/power/enabled role remain valid fallbacks for a
+    # temporarily unavailable counter.
+    assignment = _power_assignment(ranked[0])
+    for plug in ranked:
+        if _power_assignment(plug) != assignment:
+            break
+        energy = await read_energy(plug, db)
+        if energy and energy.get("total") is not None:
+            return plug, energy
+    return None
+
+
+async def read_printer_energy(
+    db: AsyncSession,
+    printer_id: int,
+    read_energy: EnergyReader,
+    *,
+    log_prefix: str,
+    context: str,
+) -> tuple[SmartPlug, dict] | None:
+    """Read the deterministically assigned printer meter with visible failures."""
+    candidates = await plugs_for_printer(db, printer_id)
+    if not candidates:
+        logger.info("[%s] No smart plug for printer %s (%s)", log_prefix, printer_id, context)
+        return None
+    selected = await select_energy_reading(candidates, read_energy, db)
+    if selected is None:
+        logger.warning(
+            "[%s] No assigned plug reports a lifetime counter for %s (linked: %s)",
+            log_prefix,
+            context,
+            ", ".join(plug.name for plug in candidates),
+        )
+    return selected
diff --git a/backend/tests/integration/test_issue_140_smart_plug_selection.py b/backend/tests/integration/test_issue_140_smart_plug_selection.py
new file mode 100644
index 0000000000..51bca6516f
--- /dev/null
+++ b/backend/tests/integration/test_issue_140_smart_plug_selection.py
@@ -0,0 +1,165 @@
+"""API-level coverage for the printer card's primary smart plug (#140)."""
+
+import pytest
+from httpx import AsyncClient
+
+pytestmark = pytest.mark.integration
+
+MAIN = "/api/v1/smart-plugs/by-printer/{}"
+ENTITIES = "/api/v1/smart-plugs/by-printer/{}/scripts"
+
+
+async def _ha(factory, printer, entity_id: str, **overrides):
+    return await factory(
+        plug_type="homeassistant",
+        printer_id=printer.id,
+        ha_entity_id=entity_id,
+        **overrides,
+    )
+
+
+class TestPrinterCardPowerRow:
+    async def test_first_created_accessory_does_not_displace_outlet(
+        self,
+        async_client: AsyncClient,
+        printer_factory,
+        smart_plug_factory,
+    ):
+        printer = await printer_factory()
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "switch.exhaust_fan",
+            name="Exhaust Fan",
+            controls_printer_power=False,
+        )
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "switch.printer_outlet",
+            name="Printer Outlet",
+            ha_power_entity="sensor.printer_power",
+        )
+
+        response = await async_client.get(MAIN.format(printer.id))
+
+        assert response.status_code == 200
+        assert response.json()["name"] == "Printer Outlet"
+
+    async def test_enabled_outlet_beats_disabled_outlet(
+        self,
+        async_client: AsyncClient,
+        printer_factory,
+        smart_plug_factory,
+    ):
+        printer = await printer_factory()
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "switch.retired",
+            name="Retired Outlet",
+            enabled=False,
+        )
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "switch.live",
+            name="Live Outlet",
+        )
+
+        response = await async_client.get(MAIN.format(printer.id))
+
+        assert response.json()["name"] == "Live Outlet"
+
+    async def test_switch_beats_ha_script_and_monitor_only_mqtt(
+        self,
+        async_client: AsyncClient,
+        printer_factory,
+        smart_plug_factory,
+    ):
+        printer = await printer_factory()
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "script.start_accessories",
+            name="Start Accessories",
+        )
+        await smart_plug_factory(
+            name="MQTT Meter",
+            plug_type="mqtt",
+            printer_id=printer.id,
+            mqtt_power_topic="tele/printer/SENSOR",
+        )
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "switch.printer",
+            name="Printer Outlet",
+        )
+
+        response = await async_client.get(MAIN.format(printer.id))
+
+        assert response.json()["name"] == "Printer Outlet"
+
+    async def test_no_plugs_returns_null(
+        self,
+        async_client: AsyncClient,
+        printer_factory,
+    ):
+        printer = await printer_factory()
+
+        response = await async_client.get(MAIN.format(printer.id))
+
+        assert response.status_code == 200
+        assert response.json() is None
+
+
+class TestAssociatedEntityRow:
+    async def test_main_switch_is_not_rendered_twice_but_accessories_remain(
+        self,
+        async_client: AsyncClient,
+        printer_factory,
+        smart_plug_factory,
+    ):
+        printer = await printer_factory()
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "switch.fan",
+            name="Fan",
+            controls_printer_power=False,
+        )
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "script.notify",
+            name="Notify Script",
+            controls_printer_power=False,
+        )
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "switch.outlet",
+            name="Printer Outlet",
+        )
+
+        response = await async_client.get(ENTITIES.format(printer.id))
+
+        assert response.status_code == 200
+        assert sorted(plug["name"] for plug in response.json()) == ["Fan", "Notify Script"]
+
+    async def test_script_only_fallback_keeps_one_click_scripts_visible(
+        self,
+        async_client: AsyncClient,
+        printer_factory,
+        smart_plug_factory,
+    ):
+        printer = await printer_factory()
+        await _ha(smart_plug_factory, printer, "script.a", name="Script A")
+        await _ha(smart_plug_factory, printer, "script.b", name="Script B")
+
+        main = await async_client.get(MAIN.format(printer.id))
+        entities = await async_client.get(ENTITIES.format(printer.id))
+
+        assert main.json()["name"] == "Script A"
+        assert sorted(plug["name"] for plug in entities.json()) == ["Script A", "Script B"]
diff --git a/backend/tests/unit/test_issue_140_smart_plug_selection.py b/backend/tests/unit/test_issue_140_smart_plug_selection.py
new file mode 100644
index 0000000000..02b338c6d7
--- /dev/null
+++ b/backend/tests/unit/test_issue_140_smart_plug_selection.py
@@ -0,0 +1,282 @@
+"""Regression coverage for deterministic printer power-plug selection (#140)."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from backend.app.services.smart_plug_selection import (
+    can_be_switched,
+    pick_power_plug,
+    plugs_for_printer,
+    rank_power_plugs,
+    reports_power,
+    select_energy_reading,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def _plug(plug_id: int = 1, **overrides) -> SimpleNamespace:
+    defaults = {
+        "id": plug_id,
+        "name": f"Plug {plug_id}",
+        "plug_type": "tasmota",
+        "ha_entity_id": None,
+        "ha_power_entity": None,
+        "mqtt_topic": None,
+        "mqtt_power_topic": None,
+        "rest_power_path": None,
+        "controls_printer_power": True,
+        "enabled": True,
+        "show_on_printer_card": True,
+    }
+    return SimpleNamespace(**(defaults | overrides))
+
+
+class TestSharedPowerPlugRanking:
+    def test_earlier_accessory_does_not_displace_printer_outlet(self):
+        fan = _plug(
+            1,
+            name="Exhaust Fan",
+            plug_type="homeassistant",
+            ha_entity_id="switch.exhaust_fan",
+            controls_printer_power=False,
+        )
+        outlet = _plug(
+            2,
+            name="Printer Outlet",
+            plug_type="homeassistant",
+            ha_entity_id="switch.printer_outlet",
+            ha_power_entity="sensor.printer_power",
+        )
+
+        assert pick_power_plug([fan, outlet]) is outlet
+        assert pick_power_plug([outlet, fan]) is outlet
+
+    def test_switch_beats_ha_script_and_monitor_only_mqtt(self):
+        script = _plug(
+            1,
+            plug_type="homeassistant",
+            ha_entity_id="script.start_accessories",
+        )
+        monitor = _plug(2, plug_type="mqtt", mqtt_power_topic="tele/printer/SENSOR")
+        switch = _plug(3, plug_type="homeassistant", ha_entity_id="switch.printer")
+
+        assert rank_power_plugs([script, monitor, switch]) == [switch, monitor, script]
+
+    def test_printer_power_flag_outranks_accessory_measurement(self):
+        metered_fan = _plug(
+            1,
+            controls_printer_power=False,
+            ha_power_entity="sensor.fan_power",
+        )
+        unmetered_outlet = _plug(
+            2,
+            plug_type="homeassistant",
+            ha_entity_id="switch.printer",
+        )
+
+        assert pick_power_plug([metered_fan, unmetered_outlet]) is unmetered_outlet
+
+    def test_enabled_plug_beats_disabled_plug(self):
+        disabled = _plug(1, enabled=False)
+        enabled = _plug(2)
+
+        assert pick_power_plug([disabled, enabled]) is enabled
+
+    def test_measurement_breaks_an_otherwise_equal_tie(self):
+        unmetered = _plug(
+            1,
+            plug_type="homeassistant",
+            ha_entity_id="switch.a",
+        )
+        metered = _plug(
+            2,
+            plug_type="homeassistant",
+            ha_entity_id="switch.b",
+            ha_power_entity="sensor.b_power",
+        )
+
+        assert pick_power_plug([unmetered, metered]) is metered
+
+    def test_equal_plugs_resolve_by_lowest_id(self):
+        first = _plug(1)
+        second = _plug(2)
+
+        assert pick_power_plug([second, first]) is first
+
+    def test_nullable_legacy_flags_rank_last_without_raising(self):
+        legacy = _plug(
+            1,
+            controls_printer_power=None,
+            enabled=None,
+            show_on_printer_card=None,
+        )
+        ordinary = _plug(2)
+
+        assert pick_power_plug([legacy, ordinary]) is ordinary
+        assert pick_power_plug([legacy]) is legacy
+
+    def test_empty_and_script_only_fallbacks_remain_supported(self):
+        script = _plug(
+            1,
+            plug_type="homeassistant",
+            ha_entity_id="script.start_accessories",
+        )
+
+        assert pick_power_plug([]) is None
+        assert pick_power_plug([script]) is script
+
+
+class TestCapabilities:
+    @pytest.mark.parametrize(
+        ("plug", "expected"),
+        [
+            (_plug(plug_type="tasmota"), True),
+            (_plug(plug_type="rest"), True),
+            (_plug(plug_type="homeassistant", ha_entity_id="switch.outlet"), True),
+            (_plug(plug_type="homeassistant", ha_entity_id="light.chamber"), True),
+            (_plug(plug_type="homeassistant", ha_entity_id="script.start"), False),
+            (_plug(plug_type="mqtt"), False),
+        ],
+    )
+    def test_switchability_matches_control_support(self, plug, expected):
+        assert can_be_switched(plug) is expected
+
+    @pytest.mark.parametrize(
+        ("plug", "expected"),
+        [
+            (_plug(plug_type="tasmota"), True),
+            (_plug(plug_type="homeassistant", ha_entity_id="switch.a"), False),
+            (_plug(plug_type="homeassistant", ha_power_entity="sensor.power"), True),
+            (_plug(plug_type="mqtt"), False),
+            (_plug(plug_type="mqtt", mqtt_topic="zigbee2mqtt/plug"), True),
+            (_plug(plug_type="rest"), False),
+            (_plug(plug_type="rest", rest_power_path="apower"), True),
+        ],
+    )
+    def test_configured_power_measurement_is_part_of_the_rank(self, plug, expected):
+        assert reports_power(plug) is expected
+
+
+class TestPrinterPlugQuery:
+    @pytest.mark.asyncio
+    async def test_returns_only_requested_printer_in_business_order(
+        self,
+        db_session,
+        printer_factory,
+        smart_plug_factory,
+    ):
+        printer = await printer_factory(name="P1S")
+        other = await printer_factory(name="X1C")
+        accessory = await smart_plug_factory(
+            name="Fan",
+            plug_type="homeassistant",
+            printer_id=printer.id,
+            controls_printer_power=False,
+        )
+        outlet = await smart_plug_factory(
+            name="P1S Outlet",
+            plug_type="homeassistant",
+            printer_id=printer.id,
+            controls_printer_power=True,
+        )
+        await smart_plug_factory(name="X1C Outlet", printer_id=other.id)
+        await smart_plug_factory(name="Bench Plug", printer_id=None)
+
+        candidates = await plugs_for_printer(db_session, printer.id)
+
+        assert [plug.id for plug in candidates] == [outlet.id, accessory.id]
+
+    @pytest.mark.asyncio
+    async def test_none_printer_id_does_not_borrow_unlinked_plugs(
+        self,
+        db_session,
+        smart_plug_factory,
+    ):
+        await smart_plug_factory(name="Bench Plug", printer_id=None)
+
+        assert await plugs_for_printer(db_session, None) == []
+
+
+class TestEnergySelection:
+    @pytest.mark.asyncio
+    async def test_does_not_bill_a_metered_accessory_when_outlet_has_no_counter(self):
+        outlet = _plug(1, name="Printer Outlet")
+        accessory = _plug(2, name="Fan", controls_printer_power=False)
+        read = AsyncMock(side_effect=lambda plug, _db: {"power": 120.0} if plug is outlet else {"total": 8.5})
+
+        selected = await select_energy_reading([outlet, accessory], read, db=None)
+
+        assert selected is None
+        read.assert_awaited_once_with(outlet, None)
+
+    @pytest.mark.asyncio
+    async def test_can_fall_back_within_same_printer_power_assignment(self):
+        unavailable = _plug(1, name="Offline meter")
+        available = _plug(2, name="Online meter")
+
+        async def read(plug, _db):
+            return None if plug is unavailable else {"total": 8.5}
+
+        selected = await select_energy_reading([unavailable, available], read, db=None)
+
+        assert selected == (available, {"total": 8.5})
+
+    @pytest.mark.asyncio
+    async def test_multiple_meters_choose_deterministically_and_poll_once(self):
+        first = _plug(1, name="First meter")
+        second = _plug(2, name="Second meter")
+        read = AsyncMock(return_value={"total": 0.0})
+
+        selected = await select_energy_reading(
+            rank_power_plugs([second, first]),
+            read,
+            db=None,
+        )
+
+        assert selected == (first, {"total": 0.0})
+        read.assert_awaited_once_with(first, None)
+
+    @pytest.mark.asyncio
+    async def test_missing_counters_return_none_instead_of_inventing_usage(self):
+        read = AsyncMock(return_value={"power": 4.0, "total": None})
+
+        assert await select_energy_reading([_plug()], read, db=None) is None
+
+    @pytest.mark.asyncio
+    async def test_record_energy_start_handles_multiple_linked_plugs(
+        self,
+        db_session,
+        printer_factory,
+        smart_plug_factory,
+        archive_factory,
+    ):
+        printer = await printer_factory()
+        await smart_plug_factory(
+            name="Fan",
+            plug_type="homeassistant",
+            printer_id=printer.id,
+            controls_printer_power=False,
+        )
+        await smart_plug_factory(
+            name="Printer Outlet",
+            plug_type="homeassistant",
+            printer_id=printer.id,
+            controls_printer_power=True,
+        )
+        archive = await archive_factory(printer.id)
+
+        from backend.app.main import _record_energy_start
+
+        async def read(plug, _db):
+            if plug.name == "Printer Outlet":
+                return {"total": 41.5}
+            return {"power": 2.0}
+
+        with patch("backend.app.main._get_plug_energy", side_effect=read):
+            recorded = await _record_energy_start(archive, printer.id, db_session)
+
+        assert recorded is True
+        assert archive.energy_start_kwh == 41.5
diff --git a/tools/check_source_size_budget.py b/tools/check_source_size_budget.py
index f59b92dad7..330bb9a241 100644
--- a/tools/check_source_size_budget.py
+++ b/tools/check_source_size_budget.py
@@ -85,10 +85,10 @@
     "backend/app/api/routes/mfa.py": 2262,
     "backend/app/api/routes/spoolman_inventory.py": 2060,
     "backend/app/core/database.py": 4224,
-    "backend/app/main.py": 6831,
+    "backend/app/main.py": 6825,
     "backend/app/services/bambu_mqtt.py": 5928,
     "backend/app/services/notification_service.py": 2184,
-    "backend/app/services/print_scheduler.py": 3800,
+    "backend/app/services/print_scheduler.py": 3791,
     "backend/tests/integration/test_mfa_api.py": 5132,
     "backend/tests/integration/test_printers_api.py": 4100,
     "backend/tests/unit/services/test_bambu_mqtt.py": 6746,
@@ -129,7 +129,7 @@
     "backend/app/core/database.py::seed_default_groups": 347,
     "backend/app/main.py::lifespan": 401,
     "backend/app/main.py::on_ams_change": 677,
-    "backend/app/main.py::on_print_complete": 1211,
+    "backend/app/main.py::on_print_complete": 1207,
     "backend/app/main.py::on_printer_status_change": 314,
     "backend/app/main.py::on_print_start": 1041,
     "backend/app/services/bambu_mqtt.py::_handle_ams_data": 552,

From cdf117ee0c42cce487b7bb8fb12bd9c41ef16c57 Mon Sep 17 00:00:00 2001
From: ichwars <drothe2012@gmail.com>
Date: Mon, 7 Sep 2026 06:55:04 +0200
Subject: [PATCH 2/2] fix: keep printer power source stable

---
 backend/app/core/database.py                  |  5 +--
 backend/app/main.py                           | 28 ++++++-------
 backend/app/models/archive.py                 |  1 +
 backend/app/services/smart_plug_selection.py  | 14 ++++---
 .../test_issue_140_smart_plug_selection.py    | 25 +++++++++++
 .../test_issue_140_smart_plug_selection.py    | 41 ++++++++++++++++++-
 tools/check_source_size_budget.py             |  8 ++--
 7 files changed, 93 insertions(+), 29 deletions(-)

diff --git a/backend/app/core/database.py b/backend/app/core/database.py
index 6d81f33c66..4b7191ce30 100644
--- a/backend/app/core/database.py
+++ b/backend/app/core/database.py
@@ -2356,10 +2356,9 @@ async def run_migrations(conn):
     else:
         await _safe_execute(conn, "ALTER TABLE users ALTER COLUMN password_hash DROP NOT NULL")
 
-    # Migration: Add energy_start_kwh to print_archives (#941)
-    # Persists the smart plug lifetime counter captured at print start, so per-print
-    # energy tracking survives a backend restart mid-print.
+    # Persist the starting counter and its exact plug across restarts (#941, #140).
     await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN energy_start_kwh REAL")
+    await _safe_execute(conn, "ALTER TABLE print_archives ADD COLUMN energy_start_plug_id INTEGER")
     # Migration: Add subtask_id to print_archives (#972)
     # MQTT-provided task identifier used to resume the same archive row across a
     # backend restart mid-print. Without it, a long print (e.g. 13h) triggers
diff --git a/backend/app/main.py b/backend/app/main.py
index a46fb637ba..4ff3b13720 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -675,7 +675,7 @@ async def _record_energy_start(archive, printer_id: int, db, *, context: str = "
         if selected is None:
             return False
         plug, energy = selected
-        archive.energy_start_kwh = float(energy["total"])
+        archive.energy_start_kwh, archive.energy_start_plug_id = float(energy["total"]), plug.id
         await db.commit()
         _logger.info(
             "[ENERGY] Recorded starting energy%s for archive %s from plug '%s': %s kWh",
@@ -4483,15 +4483,10 @@ async def _notify_no_archive():
 
     log_timing("Print log entry")
 
-    # Run slow operations as background tasks to avoid blocking the event loop
-    # These operations can take 5-10+ seconds and would freeze the UI if awaited
+    # Slow operations run in the background so completion does not freeze the UI.
 
     async def _background_energy_calculation():
-        """Calculate and save energy usage in background.
-
-        Reads the starting kWh from the archive row (#941: persisted so a mid-print
-        backend restart no longer loses per-print energy data).
-        """
+        """Calculate energy from the persisted start reading without blocking."""
         try:
             logger.info("[ENERGY-BG] Starting energy calculation for archive %s", archive_id)
             async with async_session() as db:
@@ -4501,13 +4496,18 @@ async def _background_energy_calculation():
                 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)
+                starting_kwh, starting_plug_id = archive.energy_start_kwh, archive.energy_start_plug_id
+                if starting_kwh is None or not isinstance(starting_plug_id, int):
+                    logger.info("[ENERGY-BG] No start energy source recorded for archive %s", archive_id)
                     return
 
                 selected = await read_printer_energy(
-                    db, printer_id, _get_plug_energy, log_prefix="ENERGY-BG", context=f"archive {archive_id}"
+                    db,
+                    printer_id,
+                    _get_plug_energy,
+                    log_prefix="ENERGY-BG",
+                    context=f"archive {archive_id}",
+                    plug_id=starting_plug_id,
                 )
                 if selected is None:
                     return
@@ -4531,9 +4531,7 @@ async def _background_energy_calculation():
                 cost_per_kwh = float(energy_cost_per_kwh) if energy_cost_per_kwh else 0.15
                 energy_cost_value = round(energy_used * cost_per_kwh, 3)
 
-                # First-run-only overwrite of archive.energy_kwh / energy_cost so a
-                # reprint doesn't visually clobber the source archive's energy data
-                # (#1378). Reprint energy lives in the matching PrintLogEntry below.
+                # Do not let reprints clobber source-archive energy (#1378).
                 from sqlalchemy import func
 
                 from backend.app.models.print_log import PrintLogEntry
diff --git a/backend/app/models/archive.py b/backend/app/models/archive.py
index 746c3dc40b..a09f5521f1 100644
--- a/backend/app/models/archive.py
+++ b/backend/app/models/archive.py
@@ -85,6 +85,7 @@ class PrintArchive(Base):
     # Plug lifetime counter captured at print start; delta at print end becomes energy_kwh.
     # Persisted so per-print tracking survives backend restarts mid-print (#941).
     energy_start_kwh: Mapped[float | None] = mapped_column(Float)
+    energy_start_plug_id: Mapped[int | None] = mapped_column(Integer)
 
     # Timestamps
     created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
diff --git a/backend/app/services/smart_plug_selection.py b/backend/app/services/smart_plug_selection.py
index ee0df3226f..47af8aa4a0 100644
--- a/backend/app/services/smart_plug_selection.py
+++ b/backend/app/services/smart_plug_selection.py
@@ -45,16 +45,15 @@ def reports_power(plug: SmartPlug) -> bool:
 def power_plug_rank(plug: SmartPlug) -> tuple[bool, bool, bool, bool, bool, int]:
     """Sort key for the row that represents the printer's power supply.
 
-    Capability comes first so the card never offers an unusable switch.  The
-    explicit printer-power relationship outranks display and metering details;
-    an accessory must not become the printer outlet merely because it reports
-    watts.  Nullable legacy flags rank behind explicit true values.  Lowest id
-    is the final stable tiebreaker.
+    The explicit printer-power relationship comes first, including HA scripts
+    that start a printer. Switchability, display, and metering details only
+    break ties between equally assigned rows. Nullable legacy flags rank behind
+    explicit true values. Lowest id is the final stable tiebreaker.
     """
     plug_id = getattr(plug, "id", None)
     return (
-        not can_be_switched(plug),
         not bool(getattr(plug, "controls_printer_power", False)),
+        not can_be_switched(plug),
         not bool(getattr(plug, "enabled", False)),
         not bool(getattr(plug, "show_on_printer_card", False)),
         not reports_power(plug),
@@ -128,9 +127,12 @@ async def read_printer_energy(
     *,
     log_prefix: str,
     context: str,
+    plug_id: int | None = None,
 ) -> tuple[SmartPlug, dict] | None:
     """Read the deterministically assigned printer meter with visible failures."""
     candidates = await plugs_for_printer(db, printer_id)
+    if plug_id is not None:
+        candidates = [candidate for candidate in candidates if candidate.id == plug_id]
     if not candidates:
         logger.info("[%s] No smart plug for printer %s (%s)", log_prefix, printer_id, context)
         return None
diff --git a/backend/tests/integration/test_issue_140_smart_plug_selection.py b/backend/tests/integration/test_issue_140_smart_plug_selection.py
index 51bca6516f..75c7854d04 100644
--- a/backend/tests/integration/test_issue_140_smart_plug_selection.py
+++ b/backend/tests/integration/test_issue_140_smart_plug_selection.py
@@ -101,6 +101,31 @@ async def test_switch_beats_ha_script_and_monitor_only_mqtt(
 
         assert response.json()["name"] == "Printer Outlet"
 
+    async def test_explicit_power_script_beats_switchable_accessory(
+        self,
+        async_client: AsyncClient,
+        printer_factory,
+        smart_plug_factory,
+    ):
+        printer = await printer_factory()
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "switch.exhaust_fan",
+            name="Exhaust Fan",
+            controls_printer_power=False,
+        )
+        await _ha(
+            smart_plug_factory,
+            printer,
+            "script.start_printer",
+            name="Start Printer",
+        )
+
+        response = await async_client.get(MAIN.format(printer.id))
+
+        assert response.json()["name"] == "Start Printer"
+
     async def test_no_plugs_returns_null(
         self,
         async_client: AsyncClient,
diff --git a/backend/tests/unit/test_issue_140_smart_plug_selection.py b/backend/tests/unit/test_issue_140_smart_plug_selection.py
index 02b338c6d7..c6feaf2c1e 100644
--- a/backend/tests/unit/test_issue_140_smart_plug_selection.py
+++ b/backend/tests/unit/test_issue_140_smart_plug_selection.py
@@ -10,6 +10,7 @@
     pick_power_plug,
     plugs_for_printer,
     rank_power_plugs,
+    read_printer_energy,
     reports_power,
     select_energy_reading,
 )
@@ -79,6 +80,21 @@ def test_printer_power_flag_outranks_accessory_measurement(self):
 
         assert pick_power_plug([metered_fan, unmetered_outlet]) is unmetered_outlet
 
+    def test_explicit_power_script_outranks_switchable_accessory(self):
+        power_script = _plug(
+            1,
+            plug_type="homeassistant",
+            ha_entity_id="script.start_printer",
+        )
+        accessory = _plug(
+            2,
+            plug_type="homeassistant",
+            ha_entity_id="switch.exhaust_fan",
+            controls_printer_power=False,
+        )
+
+        assert pick_power_plug([accessory, power_script]) is power_script
+
     def test_enabled_plug_beats_disabled_plug(self):
         disabled = _plug(1, enabled=False)
         enabled = _plug(2)
@@ -245,6 +261,28 @@ async def test_missing_counters_return_none_instead_of_inventing_usage(self):
 
         assert await select_energy_reading([_plug()], read, db=None) is None
 
+    @pytest.mark.asyncio
+    async def test_completion_reads_only_the_meter_persisted_at_start(self):
+        recovered = _plug(1, name="Recovered meter")
+        starting_meter = _plug(2, name="Starting meter")
+        read = AsyncMock(return_value={"total": 12.0})
+
+        with patch(
+            "backend.app.services.smart_plug_selection.plugs_for_printer",
+            new=AsyncMock(return_value=[recovered, starting_meter]),
+        ):
+            selected = await read_printer_energy(
+                None,
+                42,
+                read,
+                log_prefix="TEST",
+                context="archive 7",
+                plug_id=starting_meter.id,
+            )
+
+        assert selected == (starting_meter, {"total": 12.0})
+        read.assert_awaited_once_with(starting_meter, None)
+
     @pytest.mark.asyncio
     async def test_record_energy_start_handles_multiple_linked_plugs(
         self,
@@ -260,7 +298,7 @@ async def test_record_energy_start_handles_multiple_linked_plugs(
             printer_id=printer.id,
             controls_printer_power=False,
         )
-        await smart_plug_factory(
+        outlet = await smart_plug_factory(
             name="Printer Outlet",
             plug_type="homeassistant",
             printer_id=printer.id,
@@ -280,3 +318,4 @@ async def read(plug, _db):
 
         assert recorded is True
         assert archive.energy_start_kwh == 41.5
+        assert archive.energy_start_plug_id == outlet.id
diff --git a/tools/check_source_size_budget.py b/tools/check_source_size_budget.py
index 330bb9a241..c4328cc23c 100644
--- a/tools/check_source_size_budget.py
+++ b/tools/check_source_size_budget.py
@@ -84,8 +84,8 @@
     "backend/app/api/routes/projects.py": 2136,
     "backend/app/api/routes/mfa.py": 2262,
     "backend/app/api/routes/spoolman_inventory.py": 2060,
-    "backend/app/core/database.py": 4224,
-    "backend/app/main.py": 6825,
+    "backend/app/core/database.py": 4223,
+    "backend/app/main.py": 6823,
     "backend/app/services/bambu_mqtt.py": 5928,
     "backend/app/services/notification_service.py": 2184,
     "backend/app/services/print_scheduler.py": 3791,
@@ -125,11 +125,11 @@
     "backend/app/api/routes/spoolman.py::link_spool": 297,
     "backend/app/api/routes/support.py::_collect_support_info": 452,
     "backend/app/api/routes/updates.py::_perform_update": 303,
-    "backend/app/core/database.py::run_migrations": 2804,
+    "backend/app/core/database.py::run_migrations": 2803,
     "backend/app/core/database.py::seed_default_groups": 347,
     "backend/app/main.py::lifespan": 401,
     "backend/app/main.py::on_ams_change": 677,
-    "backend/app/main.py::on_print_complete": 1207,
+    "backend/app/main.py::on_print_complete": 1205,
     "backend/app/main.py::on_printer_status_change": 314,
     "backend/app/main.py::on_print_start": 1041,
     "backend/app/services/bambu_mqtt.py::_handle_ams_data": 552,