[modbus_controller] Add continuous polling option (#18080)

esphome/components/modbus/__init__.py

@@ -45,6 +45,7 @@ ModbusClient = modbus_ns.class_("ModbusClientHub", Modbus)
 ModbusDevice = modbus_ns.class_("ModbusDevice")
 ModbusClientDevice = modbus_ns.class_("ModbusClientDevice")
 ModbusServerDevice = modbus_ns.class_("ModbusServerDevice")
+CommandOptions = modbus_ns.struct("CommandOptions")
 MULTI_CONF = True
 
 CONF_ROLE = "role"
@@ -81,6 +82,19 @@ def _command_options(direction: str) -> list[_CommandOption]:
         raise ValueError(f"unknown command-options direction {direction!r}") from None
 
 
+# The write (mutating) function codes, matching modbus::helpers::is_function_code_write(). 0x17
+# (read/write multiple) is included: it mutates, so the hub treats it as a write despite its read half.
+_WRITE_FUNCTION_CODES = frozenset({0x05, 0x06, 0x0F, 0x10, 0x16, 0x17})
+
+
+def is_function_code_write(function_code: int) -> bool:
+    """True if the Modbus function code writes (mutates). The exception bit (0x80) is masked off first,
+    so an exception-flagged code still classifies by its base code - stricter than the runtime hub,
+    whose classify() treats an exception-flagged code as a read. Keep in sync with
+    modbus::helpers::is_function_code_write()."""
+    return function_code & 0x7F in _WRITE_FUNCTION_CODES
+
+
 def command_options_schema(
     *, direction: Literal["read", "write"], templatable: bool = False
 ) -> dict[cv.Optional, Any]:
@@ -98,6 +112,25 @@ def command_options_schema(
     }
 
 
+def command_options_expression(
+    config: ConfigType, *, direction: Literal["read", "write"]
+) -> cg.StructInitializer:
+    """Build the modbus::CommandOptions initializer for a config validated with
+    command_options_schema() of the same direction. For static (non-templatable) options only;
+    actions with lambda values use register_templatable_command_options() instead.
+    """
+    return cg.StructInitializer(
+        CommandOptions,
+        *(
+            # Construct the value as its declared cpp_type, so a future non-bool option (enum,
+            # uint16_t, ...) is emitted with the right type instead of whatever safe_exp() infers.
+            (option.field, option.cpp_type(config[option.conf_key]))
+            for option in _command_options(direction)
+            if option.conf_key in config
+        ),
+    )
+
+
 async def register_templatable_command_options(
     var: MockObj, config: ConfigType, args: TemplateArgsType, direction: str
 ) -> None:

esphome/components/modbus_client/__init__.py

@@ -157,10 +157,6 @@ _ACTION_BASE_SCHEMA = cv.Schema(
     }
 )
 
-# The write codes recognised by modbus::helpers::is_function_code_write() - keep in sync. 0x17
-# (read/write multiple) is included: it mutates, so the hub treats it as a write despite its read half.
-_WRITE_FUNCTION_CODES = frozenset({0x05, 0x06, 0x0F, 0x10, 0x16, 0x17})
-
 
 def _no_continuous_on_write(config: ConfigType) -> ConfigType:
     """Reject `continuous: true` on a static write PDU: continuous polling only applies to reads.
@@ -170,9 +166,7 @@ def _no_continuous_on_write(config: ConfigType) -> ConfigType:
     if (
         isinstance(pdu, list)
         and config.get(CONF_CONTINUOUS) is True
-        # Masking the exception bit (0x90 -> 0x10) makes this check stricter than the runtime hub,
-        # whose classify() treats an exception-flagged code as a read and leaves continuous in place.
-        and pdu[0] & 0x7F in _WRITE_FUNCTION_CODES
+        and modbus.is_function_code_write(pdu[0])
     ):
         raise cv.Invalid(
             f"'{CONF_CONTINUOUS}: true' does not apply to a write PDU (function code "

esphome/components/modbus_controller/__init__.py

@@ -11,7 +11,14 @@ from esphome.components.modbus.helpers import (
     EntityType,
 )
 import esphome.config_validation as cv
-from esphome.const import CONF_ADDRESS, CONF_ID, CONF_LAMBDA, CONF_NAME, CONF_OFFSET
+from esphome.const import (
+    CONF_ADDRESS,
+    CONF_CONTINUOUS,
+    CONF_ID,
+    CONF_LAMBDA,
+    CONF_NAME,
+    CONF_OFFSET,
+)
 from esphome.core import CORE
 from esphome.cpp_helpers import logging
 import esphome.final_validate as fv
@@ -125,6 +132,7 @@ CONFIG_SCHEMA = cv.All(
             ),
             cv.Optional(CONF_MAX_CMD_RETRIES, default=4): cv.positive_int,
             cv.Optional(CONF_OFFLINE_SKIP_UPDATES, default=0): cv.positive_int,
+            **modbus.command_options_schema(direction="read"),
             cv.Optional(
                 CONF_SERVER_REGISTERS,
             ): cv.invalid(
@@ -234,6 +242,35 @@ def migrate_custom_command(config: ConfigType) -> None:
     del config[CONF_CUSTOM_COMMAND]
 
 
+def _reject_continuous_write_custom_pdu(config: ConfigType) -> None:
+    """Final-validate: a custom_pdu whose function code writes (e.g. 0x17 read/write-multiple) cannot be
+    polled continuously - the hub ignores continuous for mutating codes and would warn on every update
+    while that range silently does not stream. Reject the combination instead. Runs after
+    migrate_custom_command, so it sees custom_pdu whether written directly or migrated from
+    custom_command."""
+    pdu = config.get(CONF_CUSTOM_PDU)
+    if pdu is None or not modbus.is_function_code_write(pdu[0]):
+        return
+    fconf = fv.full_config.get()
+    path = fconf.get_path_for_id(config[CONF_MODBUS_CONTROLLER_ID])[:-1]
+    controller = fconf.get_config_for_path(path)
+    if controller.get(CONF_CONTINUOUS) is True:
+        raise cv.Invalid(
+            f"a '{CONF_CUSTOM_PDU}' with a write function code (0x{pdu[0] & 0x7F:02X}) can't be polled "
+            f"continuously: the hub ignores 'continuous' for mutating codes. Remove 'continuous: true' "
+            f"from the '{controller[CONF_ID]}' modbus_controller, or use a read function code.",
+            [CONF_CUSTOM_PDU],
+        )
+
+
+def validate_custom_pdu_item(config: ConfigType) -> None:
+    """Final-validate for the read platforms that accept custom_pdu (sensor, binary_sensor,
+    text_sensor): migrate the deprecated custom_command, then reject a write-coded custom_pdu under a
+    continuously-polling controller."""
+    migrate_custom_command(config)
+    _reject_continuous_write_custom_pdu(config)
+
+
 def _final_validate(config: ConfigType) -> None:
     modbus.final_validate_modbus_device("modbus_controller", role="client")(config)
 
@@ -314,6 +351,11 @@ async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     cg.add(var.set_max_cmd_retries(config[CONF_MAX_CMD_RETRIES]))
     cg.add(var.set_offline_skip_updates(config[CONF_OFFLINE_SKIP_UPDATES]))
+    cg.add(
+        var.set_read_options(
+            modbus.command_options_expression(config, direction="read")
+        )
+    )
     await register_modbus_device(var, config)
     await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
 

esphome/components/modbus_controller/binary_sensor/__init__.py

@@ -8,9 +8,9 @@ from .. import (
     ModbusItemBaseSchema,
     SensorItem,
     add_modbus_base_properties,
-    migrate_custom_command,
     modbus_calc_properties,
     modbus_controller_ns,
+    validate_custom_pdu_item,
     validate_modbus_register,
 )
 from ..const import (
@@ -40,7 +40,7 @@ CONFIG_SCHEMA = cv.All(
     validate_modbus_register,
 )
 
-FINAL_VALIDATE_SCHEMA = migrate_custom_command
+FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
 
 
 async def to_code(config):

esphome/components/modbus_controller/modbus_controller.cpp

@@ -167,6 +167,7 @@ void ModbusController::queue_command(ModbusCommandItem command) {
   this->one_shot_command_items_.push_back(make_unique<ModbusCommandItem>(std::move(command)));
   // A refused frame gets no terminal callback (see the hub contract), so reclaim the item here.
   auto &item = this->one_shot_command_items_.back();
+  // We intentionally do not pass read_options_ here, because one-shot commands are usually writes, and are non-polling.
   if (!item->send()) {
     // The caller (e.g. a write entity) has usually already published optimistically - surface the loss.
     ESP_LOGW(TAG, "Command refused by hub: type=0x%X address=0x%X", static_cast<uint8_t>(item->register_type()),
@@ -203,7 +204,9 @@ void ModbusController::update() {
       ESP_LOGV(TAG, "Module offline - retrying");
       this->cmd_non_responses_ = 0;  // allow the probe through can_send()
       for (auto &cmd : this->polling_command_items_) {
-        if (!cmd.send()) {
+        // Probes carry the read-side options too, so a recovering device resumes streaming on the
+        // probe itself rather than waiting for the next update_interval.
+        if (!cmd.send(this->read_options_)) {
           ESP_LOGD(TAG, "Probe refused by hub for range 0x%X", cmd.register_address());
         }
       }
@@ -217,8 +220,9 @@ void ModbusController::update() {
   if (this->can_send()) {
     for (auto &cmd : this->polling_command_items_) {
       ESP_LOGVV(TAG, "Updating range 0x%X", cmd.register_address());
+      // read_options_ carries the controller's continuous flag (the offline probe above sends it too).
       // A refusal is already logged by the hub; note the affected range for controller-level diagnostics.
-      if (!cmd.send()) {
+      if (!cmd.send(this->read_options_)) {
         ESP_LOGD(TAG, "Poll refused by hub for range 0x%X", cmd.register_address());
       }
     }
@@ -496,16 +500,18 @@ ModbusCommandItem ModbusCommandItem::create_custom_command(
   return cmd;
 }
 
-bool ModbusCommandItem::send() {
+bool ModbusCommandItem::send(modbus::CommandOptions options) {
+  // Options pass straight through to the hub
   bool accepted;
   if (this->custom_pdu_ != nullptr) {
     // Custom polling command: send the sensor's ready-made PDU (function code + data, no address byte)
     // to this controller's own device address; the hub prepends the address and appends the CRC.
-    accepted = modbus::ModbusClientDevice::queue_pdu(std::span<const uint8_t>(*this->custom_pdu_));
+    accepted = modbus::ModbusClientDevice::queue_pdu(std::span<const uint8_t>(*this->custom_pdu_), options);
   } else if (this->function_code_ != FunctionCode::CUSTOM) {
     accepted = this->queue_pdu(modbus::helpers::create_client_pdu(
-        this->function_code_, this->start_address_, this->register_count_,
-        this->payload.empty() ? nullptr : this->payload.data(), this->payload.size()));
+                                   this->function_code_, this->start_address_, this->register_count_,
+                                   this->payload.empty() ? nullptr : this->payload.data(), this->payload.size()),
+                               options);
   } else {
     // Factory custom command: payload holds a complete raw frame (address + PDU). Send the PDU to the
     // frame's own address (which may differ from this controller's); the hub appends the CRC and routes
@@ -515,7 +521,7 @@ bool ModbusCommandItem::send() {
       ESP_LOGW(TAG, "Empty custom command frame, not sent");
       accepted = false;
     } else {
-      accepted = this->parent_->queue_pdu(frame[0], frame.subspan(1), this);
+      accepted = this->parent_->queue_pdu(frame[0], frame.subspan(1), this, options);
     }
   }
   // The on_command_sent trigger fires from on_sent() when the frame actually reaches the wire.

esphome/components/modbus_controller/modbus_controller.h

@@ -284,7 +284,9 @@ class ModbusCommandItem : public modbus::ModbusClientDevice {
   /// Queue this command's frame on the hub. Returns false when refused, in which case no callback ever comes.
   /// The item is the hub device, so it must stay alive until its terminal callback; a destroyed item's
   /// pending frame is silently retired.
-  bool send();
+  /// Options pass straight through to the hub; the polling path passes the controller's read-side
+  /// options so reads re-queue after each success, one-shot commands keep the default.
+  bool send(modbus::CommandOptions options = {});
 
   /// factory methods
   /** Create modbus read command
@@ -452,6 +454,10 @@ class ModbusController final : public PollingComponent {
   void set_max_cmd_retries(uint8_t max_cmd_retries) { this->max_cmd_retries_ = max_cmd_retries; }
   /// get how many times a command will be (re)sent if no response is received
   uint8_t get_max_cmd_retries() { return this->max_cmd_retries_; }
+  /// called by esphome generated code with the read-side command options applied to every poll
+  void set_read_options(modbus::CommandOptions options) { this->read_options_ = options; }
+  /// the read-side command options applied to every poll
+  const modbus::CommandOptions &read_options() const { return this->read_options_; }
 
  protected:
   /// parse sensormap_ and create range of sequential addresses
@@ -497,6 +503,8 @@ class ModbusController final : public PollingComponent {
   uint16_t offline_skip_updates_{0};
   /// How many times we will retry a command if we get no response
   uint8_t max_cmd_retries_{4};
+  /// read-side command options applied to every poll
+  modbus::CommandOptions read_options_{};
   /// Command sent callback
   CallbackManager<void(int, int)> command_sent_callback_{};
   /// Server online callback

esphome/components/modbus_controller/number/__init__.py

@@ -18,9 +18,9 @@ from .. import (
     ModbusItemBaseSchema,
     SensorItem,
     add_modbus_base_properties,
-    migrate_custom_command,
     modbus_calc_properties,
     modbus_controller_ns,
+    validate_custom_pdu_item,
 )
 from ..const import (
     CONF_BITMASK,
@@ -86,7 +86,7 @@ CONFIG_SCHEMA = cv.All(
     validate_modbus_number,
 )
 
-FINAL_VALIDATE_SCHEMA = migrate_custom_command
+FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
 
 
 async def to_code(config):

esphome/components/modbus_controller/sensor/__init__.py

@@ -8,9 +8,9 @@ from .. import (
     ModbusItemBaseSchema,
     SensorItem,
     add_modbus_base_properties,
-    migrate_custom_command,
     modbus_calc_properties,
     modbus_controller_ns,
+    validate_custom_pdu_item,
     validate_modbus_register,
 )
 from ..const import (
@@ -44,7 +44,7 @@ CONFIG_SCHEMA = cv.All(
     validate_modbus_register,
 )
 
-FINAL_VALIDATE_SCHEMA = migrate_custom_command
+FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
 
 
 async def to_code(config):

esphome/components/modbus_controller/switch/__init__.py

@@ -8,9 +8,9 @@ from .. import (
     ModbusItemBaseSchema,
     SensorItem,
     add_modbus_base_properties,
-    migrate_custom_command,
     modbus_calc_properties,
     modbus_controller_ns,
+    validate_custom_pdu_item,
     validate_modbus_register,
 )
 from ..const import (
@@ -45,7 +45,7 @@ CONFIG_SCHEMA = cv.All(
     validate_modbus_register,
 )
 
-FINAL_VALIDATE_SCHEMA = migrate_custom_command
+FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
 
 
 async def to_code(config):

esphome/components/modbus_controller/text_sensor/__init__.py

@@ -8,9 +8,9 @@ from .. import (
     ModbusItemBaseSchema,
     SensorItem,
     add_modbus_base_properties,
-    migrate_custom_command,
     modbus_calc_properties,
     modbus_controller_ns,
+    validate_custom_pdu_item,
     validate_modbus_register,
 )
 from ..const import (
@@ -55,7 +55,7 @@ CONFIG_SCHEMA = cv.All(
     validate_modbus_register,
 )
 
-FINAL_VALIDATE_SCHEMA = migrate_custom_command
+FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
 
 
 async def to_code(config):

tests/component_tests/modbus_controller/test_custom_pdu.py

@@ -1,19 +1,27 @@
-"""Schema-level config validation for custom_pdu and the deprecated custom_command alias.
+"""Config validation for custom_pdu and the deprecated custom_command alias.
 
 custom_command took a raw frame with a leading device address byte; custom_pdu takes the PDU only.
-The old key is still accepted at the schema level and auto-migrated later in final validate (which a
-bare-schema test can't reach), so these tests only cover what the schema itself enforces: the two keys
-are mutually exclusive, and custom_pdu takes byte-sized values.
+Most of these tests cover what the schema itself enforces (the two keys are mutually exclusive, and
+custom_pdu takes byte-sized values). The last two reach the final-validate step that a bare-schema
+test cannot: a write-coded custom_pdu polled continuously is rejected there.
 """
 
 import pytest
 from voluptuous import Invalid, MultipleInvalid
 
-from esphome.components.modbus_controller import ModbusItemBaseSchema
+from esphome.components.modbus_controller import (
+    ModbusItemBaseSchema,
+    validate_custom_pdu_item,
+)
 from esphome.components.modbus_controller.const import (
     CONF_CUSTOM_COMMAND,
     CONF_CUSTOM_PDU,
+    CONF_MODBUS_CONTROLLER_ID,
 )
+from esphome.config import Config
+from esphome.const import CONF_ADDRESS, CONF_CONTINUOUS, CONF_ID
+from esphome.core import ID
+import esphome.final_validate as fv
 
 
 def test_custom_command_accepted_at_schema_level() -> None:
@@ -45,3 +53,48 @@ def test_custom_pdu_rejects_non_byte_values() -> None:
     """PDU entries are bytes; a word-sized value is a sign the old raw format is being used."""
     with pytest.raises((Invalid, MultipleInvalid)):
         ModbusItemBaseSchema({CONF_CUSTOM_PDU: [0x0103, 0x002A]})
+
+
+def _controller_full_config(*, continuous: bool) -> Config:
+    """A minimal full-config graph with one modbus_controller declaring id 'ctl', enough for the
+    final-validate to resolve the controller (and its continuous flag) from an item's
+    modbus_controller_id."""
+    ctl_id = ID("ctl", is_declaration=True)
+    config = Config()
+    config["modbus_controller"] = [
+        {CONF_ID: ctl_id, CONF_ADDRESS: 1, CONF_CONTINUOUS: continuous}
+    ]
+    config.declare_ids.append((ctl_id, ["modbus_controller", 0, CONF_ID]))
+    return config
+
+
+@pytest.fixture
+def reset_full_config():
+    token = fv.full_config.set(Config())
+    yield
+    fv.full_config.reset(token)
+
+
+def test_continuous_write_custom_pdu_rejected(reset_full_config) -> None:
+    """A write-coded custom_pdu (0x17 = read/write-multiple) under a continuous controller is
+    rejected at final validate: the hub would strip continuous from the mutating code and warn on
+    every update."""
+    fv.full_config.set(_controller_full_config(continuous=True))
+    with pytest.raises(Invalid, match="can't be polled continuously"):
+        validate_custom_pdu_item(
+            {
+                CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
+                CONF_CUSTOM_PDU: [0x17, 0x00, 0x03, 0x00, 0x01],
+            }
+        )
+
+
+def test_continuous_read_custom_pdu_allowed(reset_full_config) -> None:
+    """A read-coded custom_pdu (0x03) under a continuous controller is fine - only writes stream."""
+    fv.full_config.set(_controller_full_config(continuous=True))
+    validate_custom_pdu_item(
+        {
+            CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
+            CONF_CUSTOM_PDU: [0x03, 0x00, 0x2A, 0x00, 0x01],
+        }
+    )

tests/components/modbus_controller/common.yaml

@@ -2,6 +2,7 @@ modbus_controller:
   - id: modbus_controller1
     address: 0x2
     modbus_id: modbus_bus
+    continuous: true
     on_online:
       then:
         logger.log: "Module Online"

tests/integration/fixtures/uart_mock_modbus_continuous.yaml

@@ -0,0 +1,115 @@
+esphome:
+  name: uart-mock-modbus-continuous
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+# When set, the mock server stops forwarding its replies to the controller, so the controller sees
+# timeouts - used by the recovery test to drive a live continuous poll offline and back.
+globals:
+  - id: silence_server
+    type: bool
+    initial_value: "false"
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - if:
+              condition:
+                lambda: "return !id(silence_server);"
+              then:
+                - uart_mock.inject_rx:
+                    id: virtual_uart_controller
+                    data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+    # Short timeout so the recovery test drives the poll offline quickly; when the server answers,
+    # replies arrive within turnaround_time, so this does not slow the streaming path.
+    send_wait_time: 100ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    id: modbus_controller_1
+    # A long update_interval means that without continuous polling only the boot poll would run in the
+    # test window. continuous: true re-queues the read after each success, so it streams as fast as the
+    # bus allows.
+    update_interval: 30s
+    continuous: true
+    # One retry so a silenced device trips offline fast (initial send + 1 retry, each 100ms).
+    max_cmd_retries: 1
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    registers:
+      # Each read returns the next counter value, so every poll publishes a distinct state the test can
+      # count (proving the read actually ran, not just that the state changed once).
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: |-
+          static uint16_t counter = 0;
+          return counter++;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "continuous_reg"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # Trigger the first poll deterministically. PollingComponent's first update() would otherwise land
+    # somewhere in the 30s update_interval; once this one read completes, continuous re-queuing takes over.
+    on_press:
+      - lambda: "id(modbus_controller_1)->update();"
+
+switch:
+  # Toggles whether the mock server forwards its replies. On = silence (controller sees timeouts);
+  # off = answer again. The recovery test uses it to drive a live continuous poll offline and back.
+  - platform: template
+    name: "Silence Server"
+    id: silence_server_switch
+    optimistic: true
+    turn_on_action:
+      - lambda: "id(silence_server) = true;"
+    turn_off_action:
+      - lambda: "id(silence_server) = false;"

tests/integration/test_uart_mock_modbus.py

@@ -736,6 +736,68 @@ async def test_uart_mock_modbus_custom_pdu(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
+@pytest.mark.asyncio
+async def test_uart_mock_modbus_continuous(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that `continuous: true` polls faster than the update_interval.
+
+    The controller's update_interval is 30s, so without continuous polling only the boot poll would
+    run during the short test window. With continuous the read is re-queued after each success, filling
+    idle bus time, so many reads arrive. The server returns an incrementing counter, so every read is a
+    distinct published state the tracker can count. (Bus warnings are not asserted here: continuous
+    polling deliberately saturates the bus, so the occasional timing hiccup is expected and off-topic;
+    the other tests cover clean operation at normal poll rates.)
+    """
+
+    tracker = SensorTracker(["continuous_reg"])
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        # setup_and_start_scenario presses the Start Scenario button, whose on_press triggers the
+        # controller's first update(). With continuous that one read re-queues and streams; without it
+        # the next poll would not run until the 30s update_interval elapses.
+        entities = await tracker.setup_and_start_scenario(client)
+        # Count reads over a window far shorter than the update_interval. Absent continuous polling we
+        # would see ~1 (the triggered poll); continuous re-queues, so the bus fills with reads.
+        await asyncio.sleep(3.0)
+        reads = len(tracker.sensor_states["continuous_reg"])
+        assert reads >= 5, (
+            "expected many continuous reads within the window (update_interval is 30s, so absent "
+            f"continuous polling we would see ~1), got {reads}"
+        )
+
+        # Recovery path: a live continuous poll that starts failing goes offline, and the next update()
+        # re-arms it once the device answers again. Silence the server so the poll's reads time out; with
+        # max_cmd_retries=1 and send_wait_time=100ms the device trips offline quickly and streaming stops.
+        silence = find_entity(entities, "silence_server", SwitchInfo)
+        assert silence is not None, "Silence Server switch not found"
+        start = find_entity(entities, "start_scenario", ButtonInfo)
+        assert start is not None, "Start Scenario button not found"
+
+        client.switch_command(silence.key, True)
+        await asyncio.sleep(1.0)  # let the poll fail and the device trip offline
+        plateau = len(tracker.sensor_states["continuous_reg"])
+        await asyncio.sleep(1.0)  # offline: no polls should land
+        assert len(tracker.sensor_states["continuous_reg"]) == plateau, (
+            "reads kept arriving after the server was silenced - the failed continuous poll did not stop"
+        )
+
+        # Answer again and trigger update(): the offline probe recovers the device and the continuous
+        # poll re-arms, so streaming resumes.
+        client.switch_command(silence.key, False)
+        client.button_command(start.key)
+        await asyncio.sleep(3.0)
+        resumed = len(tracker.sensor_states["continuous_reg"]) - plateau
+        assert resumed >= 5, (
+            f"continuous polling did not resume after the device recovered (got {resumed} new reads)"
+        )
+
+
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_offline(
     yaml_config: str,