M5-T04: add Modbus poll service and APScheduler polling job

app/config.py

@@ -41,6 +41,9 @@ class Settings(BaseSettings):
     auth_trust_forwarded_for: bool = False
     auth_totp_issuer: str = ""  # defaults to app_name when empty
 
+    # Modbus polling — global kill-switch; T08 will wire this into CONFIG_FIELDS/UI.
+    modbus_polling_enabled: bool = True
+
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

app/main.py

@@ -25,6 +25,7 @@ from app.config import get_settings
 from app.services.auth import AuthBootstrapError, initialize_auth_schema
 from app.services.config_page import seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap
 from app.services.public_ip import check_public_ipv4_and_notify
+from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SECONDS
 from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db
 
 logger = logging.getLogger(__name__)
@@ -48,6 +49,15 @@ def _run_scheduled_public_ip_check() -> None:
         session.close()
 
 
+def _run_scheduled_modbus_poll() -> None:
+    session_local = get_session_local()
+    session: Session = session_local()
+    try:
+        poll_all_enabled_devices(session, bootstrap_settings=get_settings())
+    finally:
+        session.close()
+
+
 def ensure_auth_db_ready() -> None:
     session_local = get_session_local()
     session: Session = session_local()
@@ -83,6 +93,14 @@ async def lifespan(_: FastAPI):
         max_instances=1,
         coalesce=True,
     )
+    scheduler.add_job(
+        _run_scheduled_modbus_poll,
+        trigger=IntervalTrigger(seconds=BASE_POLL_TICK_SECONDS),
+        id="modbus-poll",
+        replace_existing=True,
+        max_instances=1,
+        coalesce=True,
+    )
     scheduler.start()
     yield
     scheduler.shutdown(wait=False)

app/services/modbus_poll.py

@@ -0,0 +1,184 @@
+"""Modbus polling service — periodic read-and-store for enabled Modbus devices.
+
+Design notes
+------------
+- ``poll_device`` loads the device's YAML profile, calls the driver to bulk-read
+  all register blocks in a single TCP connection, decodes the raw registers into
+  an engineering-value dict via the profile, and persists one ``ModbusReading``
+  row.  It also stamps ``last_poll_at`` / ``last_poll_ok`` on the device.
+- All exceptions are caught and logged inside ``poll_device``; the function
+  never re-raises.  This prevents a single broken device from aborting the
+  sweep for all other devices.
+- ``poll_all_enabled_devices`` respects per-device ``poll_interval_s``: it skips
+  a device whose ``last_poll_at`` is recent enough.  This lets the APScheduler
+  job run on a short base tick (e.g. 5 s) while each device self-regulates its
+  own cadence.
+- The global ``modbus_polling_enabled`` flag from ``Settings`` (merged with any
+  DB overrides via ``build_runtime_settings``) gates the entire sweep as a
+  no-op.  T08 will expose this flag in CONFIG_FIELDS/UI; for now it is
+  controlled by the ``MODBUS_POLLING_ENABLED`` env var (or the default True).
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.config import Settings
+from app.integrations.modbus import driver, profiles
+from app.models.modbus import ModbusDevice, ModbusReading
+from app.services.config_page import build_runtime_settings
+
+logger = logging.getLogger(__name__)
+
+# Base polling tick in seconds (used by APScheduler; each device checks its own interval).
+BASE_POLL_TICK_SECONDS = 5
+
+
+def poll_device(session: Session, device: ModbusDevice) -> ModbusReading | None:
+    """Read one enabled Modbus device, decode, and persist a reading row.
+
+    Parameters
+    ----------
+    session:
+        Open SQLAlchemy session.  The caller is responsible for session
+        lifecycle (open / commit / close).
+    device:
+        ``ModbusDevice`` ORM instance to poll.
+
+    Returns
+    -------
+    ModbusReading | None
+        The newly-persisted reading on success, or ``None`` on any failure.
+        Failure is logged and the device's ``last_poll_ok`` is set to ``False``.
+    """
+    now = datetime.now(UTC)
+    try:
+        profile = profiles.load_profile(device.profile)
+        registers = driver.read_blocks(
+            device.host,
+            device.port,
+            device.unit_id,
+            [{"start": b.start, "count": b.count} for b in profile.blocks],
+        )
+        payload = profiles.decode(profile, registers)
+
+        reading = ModbusReading(
+            device_id=device.id,
+            recorded_at=now,
+            payload=payload,
+        )
+        session.add(reading)
+
+        device.last_poll_at = now
+        device.last_poll_ok = True
+        session.commit()
+
+        logger.debug(
+            "Polled device %r (id=%d): %d metrics recorded at %s",
+            device.friendly_name,
+            device.id,
+            len(payload),
+            now.isoformat(),
+        )
+        return reading
+
+    except Exception:  # noqa: BLE001
+        logger.exception(
+            "Failed to poll device %r (id=%d, host=%s:%d, unit_id=%d)",
+            device.friendly_name,
+            device.id,
+            device.host,
+            device.port,
+            device.unit_id,
+        )
+        # Rollback first — if session.add(reading) already ran in the success
+        # branch before commit() failed, that pending ModbusReading must be
+        # discarded.  Without rollback it would remain staged and could be
+        # flushed/committed by a subsequent operation (even for a different
+        # device), producing a spurious reading row for a poll that failed.
+        # Rollback also expires all ORM objects; device fields are re-set below.
+        try:
+            session.rollback()
+        except Exception:  # noqa: BLE001
+            logger.exception(
+                "session.rollback() failed for device id=%d; session may be unusable", device.id
+            )
+        try:
+            device.last_poll_at = now
+            device.last_poll_ok = False
+            session.commit()
+        except Exception:  # noqa: BLE001
+            logger.exception(
+                "Also failed to persist last_poll_ok=False for device id=%d", device.id
+            )
+        return None
+
+
+def poll_all_enabled_devices(
+    session: Session,
+    *,
+    bootstrap_settings: Settings,
+) -> None:
+    """Sweep all enabled Modbus devices and poll each one whose interval has elapsed.
+
+    Global kill-switch: if ``build_runtime_settings(...).modbus_polling_enabled``
+    is ``False``, this function returns immediately without touching any device.
+
+    Per-device interval: a device is skipped if its ``last_poll_at`` is not yet
+    ``poll_interval_s`` seconds in the past.  This allows the APScheduler job to
+    run on a short fixed tick while each device self-regulates its own cadence.
+
+    Exceptions from individual devices are swallowed inside ``poll_device``;
+    this function itself will not raise.
+
+    Parameters
+    ----------
+    session:
+        Open SQLAlchemy session.
+    bootstrap_settings:
+        The bootstrap ``Settings`` instance (from ``get_settings()``), used as
+        the base for ``build_runtime_settings`` to pick up any DB overrides.
+    """
+    try:
+        runtime_settings = build_runtime_settings(session, bootstrap_settings)
+    except Exception:  # noqa: BLE001
+        logger.exception("Failed to build runtime settings for Modbus poll; aborting sweep")
+        return
+
+    if not runtime_settings.modbus_polling_enabled:
+        logger.debug("Modbus polling is disabled (modbus_polling_enabled=False); skipping sweep")
+        return
+
+    try:
+        devices = session.execute(
+            select(ModbusDevice).where(ModbusDevice.enabled == True)  # noqa: E712
+        ).scalars().all()
+    except Exception:  # noqa: BLE001
+        logger.exception("Failed to query enabled Modbus devices; aborting sweep")
+        return
+
+    now = datetime.now(UTC)
+    for device in devices:
+        # Respect per-device poll interval: skip if last poll was too recent.
+        if device.last_poll_at is not None:
+            # SQLite may return a naive datetime despite DateTime(timezone=True) —
+            # treat naive values as UTC so the comparison is safe.
+            last_poll = device.last_poll_at
+            if last_poll.tzinfo is None:
+                last_poll = last_poll.replace(tzinfo=UTC)
+            elapsed = (now - last_poll).total_seconds()
+            if elapsed < device.poll_interval_s:
+                logger.debug(
+                    "Skipping device %r (id=%d): %.1fs elapsed < %ds interval",
+                    device.friendly_name,
+                    device.id,
+                    elapsed,
+                    device.poll_interval_s,
+                )
+                continue
+
+        poll_device(session, device)

docs/design/m5-iot-energy.md

@@ -329,7 +329,7 @@ Phase C（MQTT / Discovery，依赖 B 的设备数据与 provider 接口）
 - **Reviewer checklist**: 解码确为大端、高寄存器在前；地址与参考文档一致；**CLI 与 driver 都无任何写寄存器路径（仅 FC03/04）**；profile 纯协议知识、无部署项；`requirements.txt` 已同步锁定 `pymodbus`、`PyYAML`。
 
 ### M5-T04 — 采集 service + APScheduler 轮询
-- **Status**: `todo` · **Depends**: M5-T03
+- **Status**: `done` · **Depends**: M5-T03
 - **Context**: 周期扫所有 enabled 设备，调用 driver 读+解码，落 `modbus_reading.payload`。仿 public-ip 的同步 job 模式。
 - **Files**: `create app/services/modbus_poll.py`；`modify app/main.py`（lifespan 注册 job）；`create tests/test_modbus_poll.py`
 - **Steps**:

tests/test_modbus_poll.py

@@ -0,0 +1,465 @@
+"""Tests for M5-T04: Modbus polling service (app/services/modbus_poll.py).
+
+Coverage:
+  1. poll_all_enabled_devices inserts +N reading rows with correct payload
+     when the driver returns known register data.
+  2. A device whose driver raises leaves last_poll_ok=False; other devices
+     in the same sweep are still polled and their readings are stored.
+  3. modbus_polling_enabled=False causes the entire sweep to be skipped
+     (no readings inserted, no driver calls made).
+  4. Per-device interval: a device whose last_poll_at is too recent is
+     skipped; one whose interval has elapsed is polled.
+  5. poll_device returns None on driver failure and does not propagate.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import Session
+
+from app.config import Settings
+from app.models.modbus import ModbusDevice, ModbusReading
+from app.services.modbus_poll import poll_all_enabled_devices, poll_device
+
+
+# ---------------------------------------------------------------------------
+# Helpers and fixtures
+# ---------------------------------------------------------------------------
+
+
+def _make_app_alembic_config(database_url: str) -> Config:
+    cfg = Config("alembic_app.ini")
+    cfg.set_main_option("sqlalchemy.url", database_url)
+    return cfg
+
+
+@pytest.fixture()
+def poll_db(tmp_path: Path):
+    """Temporary SQLite DB upgraded to Alembic head, yields (engine, session_factory)."""
+    db_path = tmp_path / "poll_test.db"
+    db_url = f"sqlite:///{db_path}"
+    command.upgrade(_make_app_alembic_config(db_url), "head")
+    engine = create_engine(db_url, connect_args={"check_same_thread": False})
+    yield engine
+    engine.dispose()
+
+
+def _make_device(
+    session: Session,
+    *,
+    friendly_name: str = "Test Meter",
+    host: str = "10.0.0.1",
+    port: int = 502,
+    unit_id: int = 1,
+    profile: str = "sdm120",
+    poll_interval_s: int = 5,
+    enabled: bool = True,
+    last_poll_at: datetime | None = None,
+    last_poll_ok: bool | None = None,
+) -> ModbusDevice:
+    now = datetime.now(UTC)
+    device = ModbusDevice(
+        friendly_name=friendly_name,
+        host=host,
+        port=port,
+        unit_id=unit_id,
+        profile=profile,
+        poll_interval_s=poll_interval_s,
+        enabled=enabled,
+        last_poll_at=last_poll_at,
+        last_poll_ok=last_poll_ok,
+        created_at=now,
+        updated_at=now,
+    )
+    session.add(device)
+    session.commit()
+    return device
+
+
+# Known payload that matches what decode() should produce for a minimal SDM120 profile.
+KNOWN_PAYLOAD: dict[str, float] = {
+    "voltage": 230.20001220703125,
+    "current": 1.2999999523162842,
+    "active_power": 295.0,
+    "power_factor": 0.9800000190734863,
+    "frequency": 50.0,
+    "import_energy": 123.4000015258789,
+    "export_energy": 0.0,
+    "total_energy": 123.4000015258789,
+}
+
+
+def _make_bootstrap_settings(**overrides) -> Settings:
+    """Return a Settings instance with test defaults and optional overrides."""
+    return Settings(
+        _env_file=None,
+        app_database_url="sqlite:///:memory:",
+        auth_bootstrap_password="test",
+        **overrides,
+    )
+
+
+# ---------------------------------------------------------------------------
+# 1. Successful poll inserts +N readings with correct payload
+# ---------------------------------------------------------------------------
+
+
+def test_poll_all_inserts_readings_for_enabled_devices(poll_db):
+    """poll_all_enabled_devices inserts one reading per enabled device with correct payload."""
+    with Session(poll_db) as session:
+        d1 = _make_device(session, friendly_name="Meter A", host="10.0.0.1")
+        d2 = _make_device(session, friendly_name="Meter B", host="10.0.0.2")
+
+        # Verify baseline: no readings yet.
+        count_before = session.execute(select(ModbusReading)).scalars().all()
+        assert len(count_before) == 0
+
+        bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True)
+
+        with (
+            patch("app.services.modbus_poll.profiles.load_profile") as mock_load,
+            patch("app.services.modbus_poll.driver.read_blocks") as mock_read,
+            patch("app.services.modbus_poll.profiles.decode") as mock_decode,
+        ):
+            mock_profile = MagicMock()
+            mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)]
+            mock_load.return_value = mock_profile
+            mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334}
+            mock_decode.return_value = KNOWN_PAYLOAD
+
+            poll_all_enabled_devices(session, bootstrap_settings=bootstrap)
+
+        readings = session.execute(select(ModbusReading)).scalars().all()
+        assert len(readings) == 2, f"Expected 2 readings, got {len(readings)}"
+
+        # Both readings should have the known payload.
+        for reading in readings:
+            assert reading.payload == KNOWN_PAYLOAD, (
+                f"Payload mismatch for reading id={reading.id}: {reading.payload}"
+            )
+            assert reading.recorded_at is not None
+
+        # Devices should be marked last_poll_ok=True.
+        session.expire_all()
+        fetched_d1 = session.get(ModbusDevice, d1.id)
+        fetched_d2 = session.get(ModbusDevice, d2.id)
+        assert fetched_d1.last_poll_ok is True
+        assert fetched_d2.last_poll_ok is True
+        assert fetched_d1.last_poll_at is not None
+        assert fetched_d2.last_poll_at is not None
+
+
+# ---------------------------------------------------------------------------
+# 2. One device fails — others still succeed; failed device gets last_poll_ok=False
+# ---------------------------------------------------------------------------
+
+
+def test_poll_all_isolates_device_failure(poll_db):
+    """A driver error for one device must not prevent other devices from being polled."""
+    with Session(poll_db) as session:
+        d_ok = _make_device(session, friendly_name="OK Meter", host="10.0.0.1")
+        d_fail = _make_device(session, friendly_name="Broken Meter", host="10.0.0.2")
+
+        bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True)
+
+        call_count = {"n": 0}
+
+        def fake_read_blocks(host, port, unit_id, blocks, **kwargs):
+            call_count["n"] += 1
+            if host == "10.0.0.2":
+                raise ConnectionError("Cannot connect to broken meter")
+            return {0x0000: 0x4366, 0x0001: 0x3334}
+
+        with (
+            patch("app.services.modbus_poll.profiles.load_profile") as mock_load,
+            patch("app.services.modbus_poll.driver.read_blocks", side_effect=fake_read_blocks),
+            patch("app.services.modbus_poll.profiles.decode") as mock_decode,
+        ):
+            mock_profile = MagicMock()
+            mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)]
+            mock_load.return_value = mock_profile
+            mock_decode.return_value = KNOWN_PAYLOAD
+
+            # Must not raise even though one device fails.
+            poll_all_enabled_devices(session, bootstrap_settings=bootstrap)
+
+        # Only the OK meter should have a reading.
+        readings = session.execute(select(ModbusReading)).scalars().all()
+        assert len(readings) == 1, f"Expected 1 reading (OK meter only), got {len(readings)}"
+        assert readings[0].device_id == d_ok.id
+
+        # driver was called for both devices (fail attempt included).
+        assert call_count["n"] == 2
+
+        session.expire_all()
+        fetched_ok = session.get(ModbusDevice, d_ok.id)
+        fetched_fail = session.get(ModbusDevice, d_fail.id)
+        assert fetched_ok.last_poll_ok is True
+        assert fetched_fail.last_poll_ok is False
+
+
+# ---------------------------------------------------------------------------
+# 3. modbus_polling_enabled=False → entire sweep is skipped
+# ---------------------------------------------------------------------------
+
+
+def test_poll_all_skipped_when_polling_disabled(poll_db):
+    """When modbus_polling_enabled=False, no devices are polled and no readings are inserted."""
+    with Session(poll_db) as session:
+        _make_device(session, friendly_name="Should Be Skipped", host="10.0.0.1")
+
+        bootstrap = _make_bootstrap_settings(modbus_polling_enabled=False)
+
+        with patch("app.services.modbus_poll.driver.read_blocks") as mock_read:
+            poll_all_enabled_devices(session, bootstrap_settings=bootstrap)
+            mock_read.assert_not_called()
+
+        readings = session.execute(select(ModbusReading)).scalars().all()
+        assert len(readings) == 0, "No readings should be inserted when polling is disabled"
+
+
+def test_poll_all_skipped_via_env_var(poll_db, monkeypatch):
+    """MODBUS_POLLING_ENABLED=false env var disables polling via Settings override."""
+    monkeypatch.setenv("MODBUS_POLLING_ENABLED", "false")
+    # Build settings after env var is set.
+    bootstrap = Settings(_env_file=None, app_database_url="sqlite:///:memory:")
+
+    assert bootstrap.modbus_polling_enabled is False, (
+        "MODBUS_POLLING_ENABLED=false should yield modbus_polling_enabled=False"
+    )
+
+    with Session(poll_db) as session:
+        _make_device(session, friendly_name="Env-gated Meter", host="10.0.0.1")
+
+        with patch("app.services.modbus_poll.driver.read_blocks") as mock_read:
+            poll_all_enabled_devices(session, bootstrap_settings=bootstrap)
+            mock_read.assert_not_called()
+
+        readings = session.execute(select(ModbusReading)).scalars().all()
+        assert len(readings) == 0
+
+
+# ---------------------------------------------------------------------------
+# 4. Per-device interval: skip if recently polled, poll if interval elapsed
+# ---------------------------------------------------------------------------
+
+
+def test_poll_all_respects_per_device_interval(poll_db):
+    """Devices polled more recently than poll_interval_s are skipped; others are polled."""
+    now = datetime.now(UTC)
+
+    with Session(poll_db) as session:
+        # Device polled just 1 second ago with a 30-second interval → should be SKIPPED.
+        d_recent = _make_device(
+            session,
+            friendly_name="Recent Meter",
+            host="10.0.0.1",
+            poll_interval_s=30,
+            last_poll_at=now - timedelta(seconds=1),
+        )
+        # Device polled 60 seconds ago with a 30-second interval → should be POLLED.
+        d_due = _make_device(
+            session,
+            friendly_name="Due Meter",
+            host="10.0.0.2",
+            poll_interval_s=30,
+            last_poll_at=now - timedelta(seconds=60),
+        )
+        # Device never polled before → should be POLLED.
+        d_new = _make_device(
+            session,
+            friendly_name="New Meter",
+            host="10.0.0.3",
+            poll_interval_s=30,
+            last_poll_at=None,
+        )
+
+        bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True)
+
+        with (
+            patch("app.services.modbus_poll.profiles.load_profile") as mock_load,
+            patch("app.services.modbus_poll.driver.read_blocks") as mock_read,
+            patch("app.services.modbus_poll.profiles.decode") as mock_decode,
+        ):
+            mock_profile = MagicMock()
+            mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)]
+            mock_load.return_value = mock_profile
+            mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334}
+            mock_decode.return_value = {"voltage": 230.2}
+
+            poll_all_enabled_devices(session, bootstrap_settings=bootstrap)
+
+        readings = session.execute(select(ModbusReading)).scalars().all()
+        assert len(readings) == 2, (
+            f"Expected 2 readings (due + new), got {len(readings)}: "
+            f"{[r.device_id for r in readings]}"
+        )
+        polled_device_ids = {r.device_id for r in readings}
+        assert d_due.id in polled_device_ids, "Due meter should have been polled"
+        assert d_new.id in polled_device_ids, "New meter should have been polled"
+        assert d_recent.id not in polled_device_ids, "Recent meter should have been skipped"
+
+
+# ---------------------------------------------------------------------------
+# 5. poll_device returns None on failure, does not propagate exception
+# ---------------------------------------------------------------------------
+
+
+def test_poll_device_returns_none_on_driver_failure(poll_db):
+    """poll_device swallows driver exceptions and returns None."""
+    with Session(poll_db) as session:
+        device = _make_device(session, friendly_name="Failing Device", host="10.0.0.9")
+
+        with (
+            patch("app.services.modbus_poll.profiles.load_profile") as mock_load,
+            patch("app.services.modbus_poll.driver.read_blocks") as mock_read,
+        ):
+            mock_profile = MagicMock()
+            mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)]
+            mock_load.return_value = mock_profile
+            mock_read.side_effect = RuntimeError("Simulated driver failure")
+
+            # Must NOT raise.
+            result = poll_device(session, device)
+
+        assert result is None, "poll_device should return None on failure"
+
+        session.expire_all()
+        fetched = session.get(ModbusDevice, device.id)
+        assert fetched.last_poll_ok is False
+        assert fetched.last_poll_at is not None
+
+        # No reading row was created.
+        readings = session.execute(select(ModbusReading)).scalars().all()
+        assert len(readings) == 0
+
+
+def test_poll_device_returns_reading_on_success(poll_db):
+    """poll_device returns the inserted ModbusReading on success."""
+    with Session(poll_db) as session:
+        device = _make_device(session, friendly_name="Success Device", host="10.0.0.5")
+
+        with (
+            patch("app.services.modbus_poll.profiles.load_profile") as mock_load,
+            patch("app.services.modbus_poll.driver.read_blocks") as mock_read,
+            patch("app.services.modbus_poll.profiles.decode") as mock_decode,
+        ):
+            mock_profile = MagicMock()
+            mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)]
+            mock_load.return_value = mock_profile
+            mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334}
+            mock_decode.return_value = KNOWN_PAYLOAD
+
+            result = poll_device(session, device)
+
+        assert result is not None, "poll_device should return a ModbusReading on success"
+        assert isinstance(result, ModbusReading)
+        assert result.device_id == device.id
+        assert result.payload == KNOWN_PAYLOAD
+
+        session.expire_all()
+        fetched = session.get(ModbusDevice, device.id)
+        assert fetched.last_poll_ok is True
+
+
+# ---------------------------------------------------------------------------
+# 6. Disabled devices are not polled
+# ---------------------------------------------------------------------------
+
+
+def test_poll_all_skips_disabled_devices(poll_db):
+    """Devices with enabled=False must not be polled."""
+    with Session(poll_db) as session:
+        _make_device(session, friendly_name="Disabled Meter", host="10.0.0.1", enabled=False)
+
+        bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True)
+
+        with patch("app.services.modbus_poll.driver.read_blocks") as mock_read:
+            poll_all_enabled_devices(session, bootstrap_settings=bootstrap)
+            mock_read.assert_not_called()
+
+        readings = session.execute(select(ModbusReading)).scalars().all()
+        assert len(readings) == 0, "Disabled devices must not produce readings"
+
+
+# ---------------------------------------------------------------------------
+# 7. Rollback on commit failure: no reading survives a failed poll
+# ---------------------------------------------------------------------------
+
+
+def test_poll_device_commit_failure_leaves_no_reading(poll_db):
+    """When session.commit() raises after session.add(reading), the reading must NOT persist.
+
+    Scenario (cross-device): two devices, device A's success-branch commit fails,
+    device B's poll succeeds normally.  Expected outcome:
+      - 1 reading row total (device B only).
+      - device A: last_poll_ok=False, no reading row.
+      - device B: last_poll_ok=True, 1 reading row.
+
+    This guards against the "zombie reading" bug where a pending ModbusReading
+    added in the success branch (before commit failed) survived in the session
+    and was flushed by device B's subsequent commit.
+    """
+    with Session(poll_db) as session:
+        d_fail = _make_device(session, friendly_name="Commit-Fail Meter", host="10.0.0.1")
+        d_ok = _make_device(session, friendly_name="OK Meter", host="10.0.0.2")
+
+        bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True)
+
+        # Track how many times commit() is called so we can make only the
+        # first call (device A's success-branch commit) raise while letting
+        # the subsequent calls (device A's failure-state commit, device B's
+        # success commit) succeed normally.
+        original_commit = session.commit
+        commit_call_count = {"n": 0}
+
+        def patched_commit():
+            commit_call_count["n"] += 1
+            if commit_call_count["n"] == 1:
+                raise RuntimeError("Simulated commit failure (DB write error)")
+            return original_commit()
+
+        with (
+            patch("app.services.modbus_poll.profiles.load_profile") as mock_load,
+            patch("app.services.modbus_poll.driver.read_blocks") as mock_read,
+            patch("app.services.modbus_poll.profiles.decode") as mock_decode,
+            patch.object(session, "commit", side_effect=patched_commit),
+        ):
+            mock_profile = MagicMock()
+            mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)]
+            mock_load.return_value = mock_profile
+            mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334}
+            mock_decode.return_value = KNOWN_PAYLOAD
+
+            # Must not raise despite the first commit failing.
+            poll_all_enabled_devices(session, bootstrap_settings=bootstrap)
+
+        # Only device B (OK Meter) should have a reading — device A's reading
+        # must have been rolled back and must NOT have been committed via
+        # device B's subsequent commit (the "zombie reading" scenario).
+        readings = session.execute(select(ModbusReading)).scalars().all()
+        assert len(readings) == 1, (
+            f"Expected exactly 1 reading (OK Meter only), got {len(readings)}: "
+            f"device_ids={[r.device_id for r in readings]}"
+        )
+        assert readings[0].device_id == d_ok.id, (
+            "The surviving reading must belong to the OK Meter, not the commit-fail device"
+        )
+
+        session.expire_all()
+        fetched_fail = session.get(ModbusDevice, d_fail.id)
+        fetched_ok = session.get(ModbusDevice, d_ok.id)
+
+        assert fetched_fail.last_poll_ok is False, (
+            "Commit-fail device must be marked last_poll_ok=False"
+        )
+        assert fetched_ok.last_poll_ok is True, (
+            "OK Meter must be marked last_poll_ok=True"
+        )