Merge branch 'esp8266-native-build-surgery' into esp8266-native-toolchain-plumbing

2026-08-25 23:56:19 +00:00 · 2026-08-24 14:38:59 -05:00
parent 30fff6e91c 454e380057
commit 1c440fb30e
38 changed files with 857 additions and 120 deletions
@@ -56,7 +56,7 @@ jobs:

      # Initializes the CodeQL tools for scanning.
      - name: Initialize CodeQL
-        uses: github/codeql-action/init@ff2f1c621b7f889edc0d3c761ac2e6a3f8cdb0dd # v4.37.7
+        uses: github/codeql-action/init@db488ddef3bf6cb639b32c2e9a7c0a7ea8271d28 # v4.37.8
        with:
          languages: ${{ matrix.language }}
          build-mode: ${{ matrix.build-mode }}
@@ -84,6 +84,6 @@ jobs:
          exit 1

      - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@ff2f1c621b7f889edc0d3c761ac2e6a3f8cdb0dd # v4.37.7
+        uses: github/codeql-action/analyze@db488ddef3bf6cb639b32c2e9a7c0a7ea8271d28 # v4.37.8
        with:
          category: "/language:${{matrix.language}}"
@@ -30,8 +30,9 @@ void AudioHTTPMediaSource::dump_config() {
  ESP_LOGCONFIG(TAG,
                "Audio HTTP Media Source:\n"
                "  Buffer Size: %zu bytes\n"
+                "  Persistent Ring Buffer: %s\n"
                "  Decoder Task Stack in PSRAM: %s",
-                this->buffer_size_, YESNO(this->decoder_task_stack_in_psram_));
+                this->buffer_size_, YESNO(this->persistent_ring_buffer_), YESNO(this->decoder_task_stack_in_psram_));
 }

 void AudioHTTPMediaSource::setup() {
@@ -39,6 +40,7 @@ void AudioHTTPMediaSource::setup() {

  micro_decoder::DecoderConfig config;
  config.ring_buffer_size = this->buffer_size_;
+  config.persistent_ring_buffer = this->persistent_ring_buffer_;
  // Keep the transfer buffer smaller than the ring buffer so the reader can top up the ring
  // while the decoder is still draining it, instead of oscillating between empty and full.
  config.transfer_buffer_size = std::min(DEFAULT_TRANSFER_BUFFER_SIZE, this->buffer_size_ / 2);
@@ -33,6 +33,7 @@ class AudioHTTPMediaSource final : public Component,

  void set_buffer_size(size_t buffer_size) { this->buffer_size_ = buffer_size; }
  void set_task_stack_in_psram(bool task_stack_in_psram) { this->decoder_task_stack_in_psram_ = task_stack_in_psram; }
+  void set_persistent_ring_buffer(bool persistent) { this->persistent_ring_buffer_ = persistent; }

  // MediaSource interface implementation
  bool play_uri(const std::string &uri) override;
@@ -54,6 +55,7 @@ class AudioHTTPMediaSource final : public Component,
  // on_audio_write(). Must be atomic to avoid a data race.
  std::atomic<bool> pause_{false};
  bool decoder_task_stack_in_psram_{false};
+  bool persistent_ring_buffer_{false};
 };

 }  // namespace esphome::audio_http
@@ -7,6 +7,8 @@ from esphome.types import ConfigType
 CODEOWNERS = ["@kahrendt"]
 AUTO_LOAD = ["audio"]

+CONF_PERSISTENT_RING_BUFFER = "persistent_ring_buffer"
+
 audio_http_ns = cg.esphome_ns.namespace("audio_http")
 AudioHTTPMediaSource = audio_http_ns.class_(
    "AudioHTTPMediaSource", cg.Component, media_source.MediaSource
@@ -28,6 +30,7 @@ CONFIG_SCHEMA = cv.All(
                min=5000, max=1000000
            ),
            cv.Optional(CONF_TASK_STACK_IN_PSRAM): psram.validate_task_stack_in_psram,
+            cv.Optional(CONF_PERSISTENT_RING_BUFFER, default=False): cv.boolean,
        }
    )
    .extend(cv.COMPONENT_SCHEMA),
@@ -45,3 +48,4 @@ async def to_code(config: ConfigType) -> None:
        cg.add(var.set_task_stack_in_psram(True))
        psram.request_external_task_stack()
    cg.add(var.set_buffer_size(config[CONF_BUFFER_SIZE]))
+    cg.add(var.set_persistent_ring_buffer(config[CONF_PERSISTENT_RING_BUFFER]))
@@ -49,6 +49,13 @@ void MitsubishiCN105::initialize() { this->set_state_(State::CONNECTING); }

 bool MitsubishiCN105::update() {
  switch (this->state_) {
+    case State::DEFERRED_STATUS_REQUEST:
+      // Defer the next request to a later loop iteration; some units might not respond if a request is sent
+      // immediately after a response. See https://github.com/esphome/esphome/issues/18099. No minimum RX-to-TX delay
+      // is enforced.
+      this->set_state_(State::UPDATING_STATUS);
+      return false;
+
    case State::WAITING_FOR_SCHEDULED_STATUS_UPDATE:
      if (this->pending_updates_.any()) {
        this->status_update_wait_credit_ms_ =
@@ -101,12 +108,14 @@ bool MitsubishiCN105::should_transition(State from, State to) {
      return from == State::CONNECTING;

    case State::UPDATING_STATUS:
-      return from == State::CONNECTED || from == State::STATUS_UPDATED ||
-             from == State::WAITING_FOR_SCHEDULED_STATUS_UPDATE;
+      return from == State::DEFERRED_STATUS_REQUEST || from == State::WAITING_FOR_SCHEDULED_STATUS_UPDATE;

    case State::STATUS_UPDATED:
      return from == State::UPDATING_STATUS;

+    case State::DEFERRED_STATUS_REQUEST:
+      return from == State::CONNECTED || from == State::STATUS_UPDATED;
+
    case State::SCHEDULE_NEXT_STATUS_UPDATE:
      return from == State::STATUS_UPDATED || from == State::SETTINGS_APPLIED;

@@ -114,7 +123,7 @@ bool MitsubishiCN105::should_transition(State from, State to) {
      return from == State::SCHEDULE_NEXT_STATUS_UPDATE;

    case State::APPLYING_SETTINGS:
-      return from == State::WAITING_FOR_SCHEDULED_STATUS_UPDATE || from == State::STATUS_UPDATED;
+      return from == State::WAITING_FOR_SCHEDULED_STATUS_UPDATE;

    case State::SETTINGS_APPLIED:
      return from == State::APPLYING_SETTINGS;
@@ -122,9 +131,10 @@ bool MitsubishiCN105::should_transition(State from, State to) {
    case State::READ_TIMEOUT:
      return from == State::UPDATING_STATUS || from == State::APPLYING_SETTINGS || from == State::CONNECTING;

-    default:
+    case State::NOT_CONNECTED:
      return false;
  }
+  return false;
 }

 void MitsubishiCN105::did_transition_(State to) {
@@ -135,7 +145,7 @@ void MitsubishiCN105::did_transition_(State to) {

    case State::CONNECTED:
      this->current_status_msg_type_ = STATUS_MSG_SETTINGS;
-      this->set_state_(State::UPDATING_STATUS);
+      this->set_state_(State::DEFERRED_STATUS_REQUEST);
      break;

    case State::UPDATING_STATUS:
@@ -143,11 +153,14 @@ void MitsubishiCN105::did_transition_(State to) {
      break;

    case State::STATUS_UPDATED: {
-      if (this->pending_updates_.any() && this->is_status_initialized()) {
-        this->set_state_(State::APPLYING_SETTINGS);
-      } else if (this->current_status_msg_type_ == STATUS_MSG_SETTINGS && this->should_request_telemetry_()) {
+      // When present, pending settings are applied from WAITING_FOR_SCHEDULED_STATUS_UPDATE during the next update(),
+      // deferring transmission to a later loop iteration; some units might not respond if a request is sent
+      // immediately after a response, causing the request to time out.
+      const bool should_apply_pending_settings = this->pending_updates_.any() && this->is_status_initialized();
+      if (!should_apply_pending_settings && this->current_status_msg_type_ == STATUS_MSG_SETTINGS &&
+          this->should_request_telemetry_()) {
        this->current_status_msg_type_ = STATUS_MSG_TELEMETRY;
-        this->set_state_(State::UPDATING_STATUS);
+        this->set_state_(State::DEFERRED_STATUS_REQUEST);
      } else {
        this->set_state_(State::SCHEDULE_NEXT_STATUS_UPDATE);
      }
@@ -175,7 +188,9 @@ void MitsubishiCN105::did_transition_(State to) {
      this->set_state_(State::CONNECTING);
      break;

-    default:
+    case State::NOT_CONNECTED:
+    case State::DEFERRED_STATUS_REQUEST:
+    case State::WAITING_FOR_SCHEDULED_STATUS_UPDATE:
      break;
  }
 }
@@ -359,6 +374,8 @@ const LogString *MitsubishiCN105::state_to_string(State state) {
      return LOG_STR("UpdatingStatus");
    case State::STATUS_UPDATED:
      return LOG_STR("StatusUpdated");
+    case State::DEFERRED_STATUS_REQUEST:
+      return LOG_STR("DeferredStatusRequest");
    case State::SCHEDULE_NEXT_STATUS_UPDATE:
      return LOG_STR("ScheduleNextStatusUpdate");
    case State::WAITING_FOR_SCHEDULED_STATUS_UPDATE:
@@ -101,6 +101,7 @@ class MitsubishiCN105 {
    CONNECTED,
    UPDATING_STATUS,
    STATUS_UPDATED,
+    DEFERRED_STATUS_REQUEST,
    SCHEDULE_NEXT_STATUS_UPDATE,
    WAITING_FOR_SCHEDULED_STATUS_UPDATE,
    APPLYING_SETTINGS,
@@ -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:
@@ -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 "
@@ -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)

@@ -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):
@@ -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,9 +220,11 @@ 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());
+      }
    }
  }
  this->update_counter_++;
@@ -495,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
@@ -514,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.
@@ -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
@@ -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):
@@ -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):
@@ -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):
@@ -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):
@@ -9,7 +9,7 @@ from esphome.types import ConfigType
 CODEOWNERS = ["@esphome/core"]

 preferences_ns = cg.esphome_ns.namespace("preferences")
-IntervalSyncer = preferences_ns.class_("IntervalSyncer", cg.Component)
+IntervalSyncer = preferences_ns.class_("IntervalSyncer", cg.PollingComponent)

 CONF_FLASH_WRITE_INTERVAL = "flash_write_interval"
 CONF_RTC_STORAGE = "rtc_storage"
@@ -31,10 +31,7 @@ CONFIG_SCHEMA = cv.Schema(
 async def to_code(config: ConfigType) -> None:
    var = cg.new_Pvariable(config[CONF_ID])
    write_interval = config[CONF_FLASH_WRITE_INTERVAL]
-    if write_interval.total_milliseconds == 0:
-        cg.add_define("USE_PREFERENCES_SYNC_EVERY_LOOP")
-    else:
-        cg.add(var.set_write_interval(write_interval))
+    cg.add(var.set_update_interval(write_interval))
    if config.get(CONF_RTC_STORAGE):
        preferences.request_rtc_storage()
    await cg.register_component(var, config)
@@ -2,26 +2,19 @@

 #include "esphome/core/preferences.h"
 #include "esphome/core/component.h"
+// Include for ESPDEPRECATED, Remove before 2027.3.0
+#include "esphome/core/helpers.h"

 namespace esphome::preferences {

-class IntervalSyncer final : public Component {
+class IntervalSyncer final : public PollingComponent {
 public:
-#ifdef USE_PREFERENCES_SYNC_EVERY_LOOP
-  void loop() override { global_preferences->sync(); }
-#else
-  void set_write_interval(uint32_t write_interval) { this->write_interval_ = write_interval; }
-  void setup() override {
-    this->set_interval(this->write_interval_, []() { global_preferences->sync(); });
-  }
-#endif
+  // Remove before 2027.3.0
+  ESPDEPRECATED("Use set_update_interval() instead. Removed in 2027.3.0", "2026.9.0")
+  void set_write_interval(uint32_t write_interval) { this->set_update_interval(write_interval); }
+  void update() override { global_preferences->sync(); }
  void on_shutdown() override { global_preferences->sync(); }
  float get_setup_priority() const override { return setup_priority::BUS; }
-
-#ifndef USE_PREFERENCES_SYNC_EVERY_LOOP
- protected:
-  uint32_t write_interval_{60000};
-#endif
 };

 }  // namespace esphome::preferences
@@ -393,5 +393,16 @@ void IRAM_ATTR IDFUARTComponent::uart_rx_isr_callback(uart_port_t uart_num, uart
 }
 #endif  // USE_UART_WAKE_LOOP_ON_RX

+void IDFUARTComponent::on_shutdown() {
+  if (this->uart_num_ == UART_NUM_MAX || !uart_is_driver_installed(this->uart_num_))
+    return;
+  uart_wait_tx_done(this->uart_num_, pdMS_TO_TICKS(100));
+  // Keep the peripheral quiet across a soft reset so ROM output does not reach the attached device (#15472)
+  esp_err_t err = uart_driver_delete(this->uart_num_);
+  if (err != ESP_OK) {
+    ESP_LOGW(TAG, "uart_driver_delete failed: %s", esp_err_to_name(err));
+  }
+}
+
 }  // namespace esphome::uart
 #endif  // USE_ESP32
@@ -52,9 +52,11 @@ class IDFUARTComponent final : public UARTComponent, public Component {
  void load_settings(bool dump_config) override;
  using UARTComponent::load_settings;  // also bring in the no-arg overload for convenience

+  void on_shutdown() override;
+
 protected:
  void check_logger_conflict() override;
-  uart_port_t uart_num_;
+  uart_port_t uart_num_{UART_NUM_MAX};
  uart_config_t get_config_();

  bool has_peek_{false};
@@ -160,7 +160,6 @@
 #define USE_OUTPUT
 #define USE_OUTPUT_FLOAT_POWER_SCALING
 #define USE_POWER_SUPPLY
-#define USE_PREFERENCES_SYNC_EVERY_LOOP
 // Only defined by key-lookup preference backends; the slot-based platforms
 // (esp8266, rp2040) never set it in generated builds, and their preferences
 // managers do not provide load_from_key(), so the PreferencesKeyLookupContract
@@ -1,6 +1,6 @@
 pylint==4.0.7
 flake8==7.3.0  # also change in .pre-commit-config.yaml when updating
-ruff==0.16.3  # also change in .pre-commit-config.yaml when updating
+ruff==0.16.4  # also change in .pre-commit-config.yaml when updating
 pyupgrade==3.21.2  # also change in .pre-commit-config.yaml when updating
 prek==0.4.14  # also change in .github/workflows/ci.yml when updating

@@ -10,7 +10,7 @@ from pathlib import Path
 import subprocess
 import sys

-from helpers import get_all_dependencies, root_path as _root_path
+from helpers import get_all_dependencies, has_cpp_unit_tests, root_path as _root_path
 import yaml

 # Ensure the repo root is on sys.path so that ``tests.testing_helpers`` and
@@ -131,14 +131,11 @@ def filter_components_with_files(components: list[str], tests_dir: Path) -> list
    """
    filtered_components: list[str] = []
    for component in components:
-        test_dir = tests_dir / component
-        if test_dir.is_dir() and (
-            any(test_dir.glob("*.cpp")) or any(test_dir.glob("*.h"))
-        ):
+        if has_cpp_unit_tests(component, tests_dir):
            filtered_components.append(component)
        else:
            print(
-                f"WARNING: No files found for component '{component}' in {test_dir}, skipping.",
+                f"WARNING: No files found for component '{component}' in {tests_dir / component}, skipping.",
                file=sys.stderr,
            )
    return filtered_components
@@ -66,7 +66,6 @@ from helpers import (
    base_python_changed,
    changed_files,
    core_changed,
-    filter_component_and_test_cpp_files,
    filter_component_and_test_files,
    get_changed_components,
    get_component_from_path,
@@ -628,12 +627,17 @@ def determine_cpp_unit_tests(

    C++ unit tests will run when any of the following conditions are met:

-    1. Any C++ core source files changed (esphome/core/*), in which case
+    1. Any core C++ or Python files changed (esphome/core/*), in which case
       all cpp unit tests run.
    2. A test file for a component changed, which triggers tests for that
       component.
    3. The code for a component changed, which triggers tests for that
-       component and all components that depend on it.
+       component and all components that depend on it. Python files count
+       too: a component's Python decides which sources and defines go into
+       the host test build, so a Python-only change can break the link.
+
+    Components without C++ test sources are dropped from the list, so the
+    job is only scheduled when there is something to build.

    Args:
        branch: Branch to compare against. If None, uses default.
@@ -647,9 +651,7 @@ def determine_cpp_unit_tests(
    if core_changed(files):
        return (True, [])

-    # Filter to only C++ files
-    cpp_files = list(filter(filter_component_and_test_cpp_files, files))
-    return (False, get_cpp_changed_components(cpp_files))
+    return (False, get_cpp_changed_components(files))


 # Paths within tests/benchmarks/ that contain component benchmark files
@@ -1150,17 +1150,41 @@ def filter_component_and_test_files(file_path: str) -> bool:
    )


-def filter_component_and_test_cpp_files(file_path: str) -> bool:
-    """Check if a file is a C++ source file in component or test directories.
+def filter_cpp_unit_test_files(file_path: str) -> bool:
+    """Check if a file can affect a component's C++ unit test build.
+
+    Besides C++ sources, a component's Python code (defines, source file
+    filters, libraries) and the ``__init__.py`` manifest overrides under
+    ``tests/components/<component>/`` decide what the host test binary
+    compiles and links. Other Python files under ``tests/components/``
+    (pytest conftest.py, fixtures) do not.

    Args:
        file_path: Path to check

    Returns:
-        True if the file is a C++ source/header file in component or test directories
+        True if the file is a C++ or Python file in a component directory, or
+        a C++ file or ``__init__.py`` in a component test directory
    """
-    return file_path.endswith(CPP_FILE_EXTENSIONS) and file_path.startswith(
-        COMPONENT_AND_TESTS_PATHS
+    if file_path.startswith(ESPHOME_COMPONENTS_PATH):
+        return file_path.endswith(CPP_AND_PYTHON_FILE_EXTENSIONS)
+    if file_path.startswith(ESPHOME_TESTS_COMPONENTS_PATH):
+        return file_path.endswith(CPP_FILE_EXTENSIONS) or file_path.endswith(
+            "/__init__.py"
+        )
+    return False
+
+
+def has_cpp_unit_tests(component: str, tests_dir: Path) -> bool:
+    """Check if a component has C++ test or benchmark sources in ``tests_dir``.
+
+    Shared by CI job selection and the build itself
+    (``build_helpers.filter_components_with_files``) so both agree on
+    which components have something to build.
+    """
+    component_dir = tests_dir / component
+    return component_dir.is_dir() and (
+        any(component_dir.glob("*.cpp")) or any(component_dir.glob("*.h"))
    )


@@ -1486,41 +1510,41 @@ def base_python_changed(files: list[str]) -> bool:


 def get_cpp_changed_components(files: list[str]) -> list[str]:
-    """Get components that have changed C++ files or tests.
+    """Get components whose C++ unit tests are affected by changed files.

    This function analyzes a list of changed files and determines which components
    are affected. It handles two scenarios:

-    1. Test files changed (tests/components/<component>/*.cpp):
+    1. Test files changed (tests/components/<component>/*.cpp or __init__.py):
       - Adds the component to the affected list
       - Only that component needs to be tested

-    2. Component C++ files changed (esphome/components/<component>/*):
+    2. Component files changed (esphome/components/<component>/*.cpp or *.py):
       - Adds the component to the affected list
       - Also adds all components that depend on this component (recursively)
       - This ensures that changes propagate to dependent components

+    Python files count because a component's Python code decides which
+    sources and defines end up in the host test build. Components without
+    C++ test sources are dropped so CI does not schedule the job for nothing.
+
    Args:
-        files: List of file paths to analyze (should be C++ files)
+        files: List of changed file paths; irrelevant ones are ignored

    Returns:
        Sorted list of component names that need C++ unit tests run
    """
    components_graph = create_components_graph()
+    tests_dir = Path(root_path) / ESPHOME_TESTS_COMPONENTS_PATH
    affected: set[str] = set()
    for file in files:
-        if not file.endswith(CPP_FILE_EXTENSIONS):
+        if not filter_cpp_unit_test_files(file):
            continue
-        if file.startswith(ESPHOME_TESTS_COMPONENTS_PATH):
-            parts = file.split("/")
-            if len(parts) >= 4:
-                component_dir = Path(ESPHOME_TESTS_COMPONENTS_PATH) / parts[2]
-                if component_dir.is_dir():
-                    affected.add(parts[2])
-        elif file.startswith(ESPHOME_COMPONENTS_PATH):
-            parts = file.split("/")
-            if len(parts) >= 4:
-                component = parts[2]
-                affected.update(find_children_of_component(components_graph, component))
-                affected.add(component)
-    return sorted(affected)
+        parts = file.split("/")
+        if len(parts) < 4:
+            continue
+        component = parts[2]
+        affected.add(component)
+        if file.startswith(ESPHOME_COMPONENTS_PATH):
+            affected.update(find_children_of_component(components_graph, component))
+    return sorted(c for c in affected if has_cpp_unit_tests(c, tests_dir))
@@ -3,7 +3,6 @@ import argparse

 from helpers import (
    changed_files,
-    filter_component_and_test_cpp_files,
    filter_component_and_test_files,
    get_all_component_files,
    get_components_with_dependencies,
@@ -38,7 +37,7 @@ def main():
    parser.add_argument(
        "--cpp-changed",
        action="store_true",
-        help="List components with changed C++ files",
+        help="List components whose C++ unit tests are affected by changed files",
    )
    args = parser.parse_args()

@@ -78,9 +77,9 @@ def main():
        #   Returns: Components with code changes + their dependencies (not infrastructure)
        #   Reason: CI needs to test changed components and their dependents
        #
-        # - --cpp-changed: Used by CI to determine if any C++ files changed (script/determine-jobs.py)
-        #   Returns: Only components with changed C++ files
-        #   Reason: Only components with C++ changes need C++ testing
+        # - --cpp-changed: Mirrors the C++ unit test selection in script/determine-jobs.py
+        #   Returns: Components with changed C++ or Python files (plus dependents)
+        #   Reason: Python decides which sources and defines go into the host test build

        base_test_changed = any(
            "tests/test_build_components" in file for file in changed
@@ -115,9 +114,7 @@ def main():
        for c in get_components_with_dependencies(files, False):
            print(c)
    elif args.cpp_changed:
-        # Only look at changed cpp files
-        files = list(filter(filter_component_and_test_cpp_files, changed))
-        for c in get_cpp_changed_components(files):
+        for c in get_cpp_changed_components(changed):
            print(c)
    else:
        # Return all changed components (with dependencies) - default behavior
@@ -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],
+        }
+    )
@@ -4,4 +4,5 @@ media_source:
  - platform: audio_http
    id: audio_http_source
    buffer_size: 100000
+    persistent_ring_buffer: true
    task_stack_in_psram: true
@@ -15,11 +15,10 @@ struct MitsubishiCN105ClimateTestContext {

 TEST(MitsubishiCN105ClimateTests, CelsiusTemperatureMappingAndTraitsMatchExpectedValues) {
  MitsubishiCN105ClimateTestContext context;
-  const auto mapping = TemperatureMapping();

  for (int temperature = 16; temperature <= 31; ++temperature) {
-    EXPECT_EQ(mapping.to_mitsubishi(temperature), temperature);
-    EXPECT_EQ(mapping.from_mitsubishi(temperature), temperature);
+    EXPECT_EQ(context.component.get_temperature_mapping().to_mitsubishi(temperature), temperature);
+    EXPECT_EQ(context.component.get_temperature_mapping().from_mitsubishi(temperature), temperature);
  }

  const auto traits = context.sut.traits();
@@ -32,8 +31,6 @@ TEST(MitsubishiCN105ClimateTests, CelsiusTemperatureMappingAndTraitsMatchExpecte

 TEST(MitsubishiCN105ClimateTests, FahrenheitTemperatureMappingAndTraitsMatchExpectedValues) {
  MitsubishiCN105ClimateTestContext context;
-  auto mapping = TemperatureMapping();
-  mapping.set_use_fahrenheit(true);
  context.component.set_use_fahrenheit(true);

  const std::array cases{
@@ -46,8 +43,8 @@ TEST(MitsubishiCN105ClimateTests, FahrenheitTemperatureMappingAndTraitsMatchExpe
  };

  for (const auto &[fahrenheit, mitsubishi_celsius] : cases) {
-    EXPECT_FLOAT_EQ(mapping.to_mitsubishi(fahrenheit), mitsubishi_celsius);
-    EXPECT_FLOAT_EQ(mapping.from_mitsubishi(mitsubishi_celsius), fahrenheit);
+    EXPECT_FLOAT_EQ(context.component.get_temperature_mapping().to_mitsubishi(fahrenheit), mitsubishi_celsius);
+    EXPECT_FLOAT_EQ(context.component.get_temperature_mapping().from_mitsubishi(mitsubishi_celsius), fahrenheit);
  }
  const auto traits = context.sut.traits();
  EXPECT_EQ(traits.get_temperature_unit(), TemperatureUnit::FAHRENHEIT);
@@ -42,11 +42,17 @@ TEST(MitsubishiCN105Tests, ConnectAndUpdateStatus) {

  // All bytes from UART should be consumed
  EXPECT_TRUE(ctx.uart.rx.empty());
-  // After successful connect we request status, first settings (0x02)
+  // Defer the first settings request (0x02) until the next update.
+  EXPECT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::DEFERRED_STATUS_REQUEST);
+  EXPECT_TRUE(ctx.uart.tx.empty());
+
+  ctx.sut.set_current_time(201);
+  ASSERT_FALSE(ctx.sut.update());
+
  EXPECT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::UPDATING_STATUS);
  EXPECT_THAT(ctx.uart.tx, ::testing::ElementsAre(0xFC, 0x42, 0x01, 0x30, 0x10, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00,
                                                  0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x7B));
-  EXPECT_EQ(ctx.sut.operation_start_ms_, 200);
+  EXPECT_EQ(ctx.sut.operation_start_ms_, 201);

  // Clear TX bytes.
  ctx.uart.tx.clear();
@@ -75,15 +81,24 @@ TEST(MitsubishiCN105Tests, ConnectAndUpdateStatus) {
  EXPECT_EQ(ctx.sut.status().vane_mode, MitsubishiCN105::VaneMode::POSITION_4);
  EXPECT_EQ(ctx.sut.status().wide_vane_mode, MitsubishiCN105::WideVaneMode::SWING);

-  // Now fetch telemetry (0x03)
+  // Defer the telemetry request (0x03) until the next update.
+  EXPECT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::DEFERRED_STATUS_REQUEST);
+  EXPECT_TRUE(ctx.uart.tx.empty());
+
+  ctx.sut.set_current_time(301);
+  ASSERT_FALSE(ctx.sut.update());
+
  EXPECT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::UPDATING_STATUS);
  EXPECT_THAT(ctx.uart.tx, ::testing::ElementsAre(0xFC, 0x42, 0x01, 0x30, 0x10, 0x03, 0x00, 0x00, 0x00, 0x00, 0x00,
                                                  0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x7A));
-  EXPECT_EQ(ctx.sut.operation_start_ms_, 300);
+  EXPECT_EQ(ctx.sut.operation_start_ms_, 301);

  // Clear TX bytes.
  ctx.uart.tx.clear();

+  // Queue a setting while waiting for telemetry.
+  ctx.sut.set_power(true);
+
  // Telemetry response
  ctx.uart.push_rx({0xFC, 0x62, 0x01, 0x30, 0x10, 0x03, 0x00, 0x00, 0x0B, 0x00, 0x00,
                    0xAA, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xA5});
@@ -103,6 +118,13 @@ TEST(MitsubishiCN105Tests, ConnectAndUpdateStatus) {
  EXPECT_TRUE(ctx.uart.tx.empty());
  EXPECT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::WAITING_FOR_SCHEDULED_STATUS_UPDATE);
  EXPECT_EQ(ctx.sut.operation_start_ms_, 400);
+
+  // Apply the pending setting on the next update, outside RX processing.
+  ctx.sut.set_current_time(401);
+  ASSERT_FALSE(ctx.sut.update());
+  EXPECT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::APPLYING_SETTINGS);
+  EXPECT_FALSE(ctx.uart.tx.empty());
+  EXPECT_EQ(ctx.sut.operation_start_ms_, 401);
 }

 TEST(MitsubishiCN105Tests, NoResponseTriggersReconnect) {
@@ -469,6 +491,36 @@ TEST(MitsubishiCN105Tests, WriteInterruptsWaitingForNextStatusUpdate) {
  EXPECT_EQ(ctx.sut.status_update_wait_credit_ms_, 0);
 }

+TEST(MitsubishiCN105Tests, PendingSettingsTakePriorityOverDueTelemetry) {
+  MitsubishiCN105TestsContext ctx;
+
+  ctx.sut.status_.target_temperature = 24.0f;
+  ctx.sut.status_.room_temperature = 21.0f;
+  ASSERT_TRUE(ctx.sut.is_status_initialized());
+
+  ctx.sut.state_ = TestableMitsubishiCN105::State::STATUS_UPDATED;
+  ctx.sut.set_state(TestableMitsubishiCN105::State::SCHEDULE_NEXT_STATUS_UPDATE);
+  ctx.sut.set_current_time(1000);
+  ASSERT_FALSE(ctx.sut.update());
+  ASSERT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::UPDATING_STATUS);
+  ctx.uart.tx.clear();
+
+  ctx.sut.set_power(true);
+  ctx.uart.push_rx({0xFC, 0x62, 0x01, 0x30, 0x10, 0x02, 0x00, 0x00, 0x00, 0x08, 0x07,
+                    0x00, 0x04, 0x00, 0x00, 0x0C, 0x00, 0x00, 0x00, 0x00, 0x00, 0x3C});
+
+  ctx.sut.set_current_time(1001);
+  ASSERT_TRUE(ctx.sut.update());
+  EXPECT_TRUE(ctx.uart.tx.empty());
+  EXPECT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::WAITING_FOR_SCHEDULED_STATUS_UPDATE);
+
+  ctx.sut.set_current_time(1002);
+  ASSERT_FALSE(ctx.sut.update());
+  EXPECT_EQ(ctx.sut.state_, TestableMitsubishiCN105::State::APPLYING_SETTINGS);
+  EXPECT_THAT(ctx.uart.tx, ::testing::ElementsAre(0xFC, 0x41, 0x01, 0x30, 0x10, 0x01, 0x01, 0x00, 0x01, 0x00, 0x00,
+                                                  0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x7B));
+}
+
 TEST(MitsubishiCN105Tests, SetAndClearRemoteRoomTemp) {
  MitsubishiCN105TestsContext ctx;

@@ -2,6 +2,7 @@ modbus_controller:
  - id: modbus_controller1
    address: 0x2
    modbus_id: modbus_bus
+    continuous: true
    on_online:
      then:
        logger.log: "Module Online"
@@ -1,2 +1,9 @@
 preferences:
+  id: prefs_syncer
  flash_write_interval: 20s
+
+esphome:
+  on_boot:
+    then:
+      - component.suspend: prefs_syncer
+      - component.resume: prefs_syncer
@@ -0,0 +1,41 @@
+esphome:
+  name: test_suspend_resume_device
+
+host:
+
+logger:
+  level: DEBUG
+
+api:
+
+preferences:
+  id: prefs_syncer
+  flash_write_interval: 1s
+
+button:
+  - platform: template
+    name: "Save Preference"
+    on_press:
+      - lambda: |-
+          // save() only updates the in-memory map; only sync() persists it to disk.
+          ESPPreferenceObject pref = global_preferences->make_preference<uint32_t>(0xBEEF);
+          uint32_t value = 123;
+          if (pref.save(&value)) {
+            ESP_LOGI("test", "Preference saved in memory");
+          } else {
+            ESP_LOGE("test", "Preference save failed");
+          }
+
+  - platform: template
+    name: "Suspend Syncer"
+    on_press:
+      - component.suspend: prefs_syncer
+      - lambda: |-
+          ESP_LOGI("test", "Syncer suspended");
+
+  - platform: template
+    name: "Resume Syncer"
+    on_press:
+      - component.resume: prefs_syncer
+      - lambda: |-
+          ESP_LOGI("test", "Syncer resumed");
@@ -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;"
@@ -0,0 +1,137 @@
+"""Test that suspending/resuming the preferences IntervalSyncer actually stops/starts flash writes."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable
+from pathlib import Path
+import re
+from typing import Any
+
+from aioesphomeapi import ButtonInfo, EntityInfo
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+DEVICE_NAME = "test_suspend_resume_device"
+
+
+def find_entity_by_name(
+    entities: list[EntityInfo], entity_type: type, name: str
+) -> Any:
+    """Helper to find an entity by type and name."""
+    return next(
+        (e for e in entities if isinstance(e, entity_type) and e.name == name), None
+    )
+
+
+async def _wait_for(
+    awaitable: Awaitable[Any], message: str, timeout: float = 5.0
+) -> None:
+    """Await a future or coroutine, failing the test with a clear message on timeout."""
+    try:
+        await asyncio.wait_for(awaitable, timeout=timeout)
+    except TimeoutError:
+        pytest.fail(message)
+
+
+async def _poll_until_exists(path: Path) -> None:
+    """Poll for a file to appear, rather than guessing a sleep duration."""
+    while not path.exists():
+        await asyncio.sleep(0.05)
+
+
+@pytest.fixture(autouse=True)
+def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> Path:
+    """Keep host preferences per-test so this test never touches the real
+    ~/.esphome/prefs and never races other tests over ESPHOME_PREFDIR."""
+    prefdir = tmp_path / "prefs"
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(prefdir))
+    return prefdir / f"{DEVICE_NAME}.prefs"
+
+
+@pytest.mark.asyncio
+async def test_host_preferences_suspend_resume(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+    isolated_preferences: Path,
+) -> None:
+    """Test that a running syncer flushes, a suspended one doesn't, and resume restores flushing."""
+    pref_file = isolated_preferences
+
+    loop = asyncio.get_running_loop()
+    saved_in_memory = loop.create_future()
+    syncer_suspended = loop.create_future()
+    syncer_resumed = loop.create_future()
+
+    save_pattern = re.compile(r"Preference saved in memory")
+    suspend_pattern = re.compile(r"Syncer suspended")
+    resume_pattern = re.compile(r"Syncer resumed")
+
+    def check_output(line: str) -> None:
+        if save_pattern.search(line) and not saved_in_memory.done():
+            saved_in_memory.set_result(True)
+        if suspend_pattern.search(line) and not syncer_suspended.done():
+            syncer_suspended.set_result(True)
+        if resume_pattern.search(line) and not syncer_resumed.done():
+            syncer_resumed.set_result(True)
+
+    async with (
+        run_compiled(yaml_config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        entities, _ = await client.list_entities_services()
+
+        save_button = find_entity_by_name(entities, ButtonInfo, "Save Preference")
+        suspend_button = find_entity_by_name(entities, ButtonInfo, "Suspend Syncer")
+        resume_button = find_entity_by_name(entities, ButtonInfo, "Resume Syncer")
+        assert save_button is not None, "Save Preference button not found"
+        assert suspend_button is not None, "Suspend Syncer button not found"
+        assert resume_button is not None, "Resume Syncer button not found"
+
+        # --- Positive control: a running syncer flushes to disk. Without this,
+        # the suspend assertion below could pass for the wrong reason (e.g. wrong prefs path). ---
+        client.button_command(save_button.key)
+        await _wait_for(
+            saved_in_memory, "Preference was not saved to memory within timeout"
+        )
+        await _wait_for(
+            _poll_until_exists(pref_file),
+            "Running syncer never flushed to disk; positive control failed",
+            timeout=10.0,
+        )
+        saved_in_memory = loop.create_future()
+
+        # --- Suspend: a suspended syncer must not flush. ---
+        client.button_command(suspend_button.key)
+        await _wait_for(
+            syncer_suspended, "Syncer suspend command was not processed within timeout"
+        )
+        # Delete only after suspend is confirmed: the poller is now stopped, so
+        # nothing can recreate the file before the negative assertion below.
+        pref_file.unlink()
+
+        client.button_command(save_button.key)
+        await _wait_for(
+            saved_in_memory, "Preference was not saved to memory within timeout"
+        )
+
+        # Wait well past flash_write_interval (1s): a running syncer would
+        # have flushed to disk by now, a suspended one must not have. This is a
+        # negative assertion (proving absence), so a fixed sleep is unavoidable here.
+        await asyncio.sleep(1.5)
+        assert not pref_file.exists(), (
+            "Suspended syncer flushed to disk; component.suspend did not stop the poller"
+        )
+
+        # --- Resume: flushing must restart. ---
+        client.button_command(resume_button.key)
+        await _wait_for(
+            syncer_resumed, "Syncer resume command was not processed within timeout"
+        )
+        await _wait_for(
+            _poll_until_exists(pref_file),
+            "Resumed syncer never flushed to disk; component.resume did not restart the poller",
+            timeout=10.0,
+        )
@@ -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,
@@ -1214,6 +1214,56 @@ def test_count_changed_cpp_files_with_branch() -> None:
        mock_changed.assert_called_once_with("release")


+@pytest.mark.parametrize(
+    ("changed_files", "expected"),
+    [
+        # Core C++ change runs everything
+        (["esphome/core/helpers.cpp"], (True, [])),
+        # Core Python change runs everything too
+        (["esphome/core/config.py"], (True, [])),
+        # Component C++ change: component plus dependents with C++ tests
+        (["esphome/components/time/posix_tz.cpp"], (False, ["sntp", "time"])),
+        # Component Python change shapes the host build (defines, source
+        # filters), so it must trigger the same tests as a C++ change
+        (["esphome/components/time/__init__.py"], (False, ["sntp", "time"])),
+        # Nothing to build when no selected component has C++ tests
+        (["esphome/components/homeassistant/__init__.py"], (False, [])),
+        # Test manifest override changes only that component
+        (["tests/components/time/__init__.py"], (False, ["time"])),
+        # Test source change only that component
+        (["tests/components/time/posix_tz.cpp"], (False, ["time"])),
+        # pytest files and YAML build tests do not affect the test binary
+        (["tests/components/socket/conftest.py"], (False, [])),
+        (["tests/components/time/test.esp32-idf.yaml"], (False, [])),
+        (["README.md", "script/helpers.py"], (False, [])),
+        ([], (False, [])),
+    ],
+)
+def test_determine_cpp_unit_tests(
+    changed_files: list[str],
+    expected: tuple[bool, list[str]],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test which C++ unit tests a set of changed files selects."""
+    tests_dir = tmp_path / "tests" / "components"
+    for component in ("time", "sntp"):
+        (tests_dir / component).mkdir(parents=True)
+        (tests_dir / component / f"{component}.cpp").write_text("")
+    (tests_dir / "homeassistant").mkdir()
+    (tests_dir / "socket").mkdir()
+    monkeypatch.setattr(helpers, "root_path", str(tmp_path))
+    with (
+        patch.object(determine_jobs, "changed_files", return_value=changed_files),
+        patch.object(
+            helpers,
+            "create_components_graph",
+            return_value={"time": ["homeassistant", "sntp"]},
+        ),
+    ):
+        assert determine_jobs.determine_cpp_unit_tests() == expected
+
+
 def test_main_filters_components_without_tests(
    mock_determine_integration_tests: Mock,
    mock_should_run_clang_tidy: Mock,
@@ -2031,3 +2031,92 @@ def test_get_changed_files_from_command_gh_failure_keeps_stderr() -> None:
        pytest.raises(Exception, match="maximum number of changed files"),
    ):
        _get_changed_files_from_command(["gh", "pr", "diff", "123", "--name-only"])
+
+
+@pytest.mark.parametrize(
+    ("file_path", "expected"),
+    [
+        ("esphome/components/time/posix_tz.cpp", True),
+        ("esphome/components/time/posix_tz.h", True),
+        ("esphome/components/time/__init__.py", True),
+        ("esphome/components/sntp/time.py", True),
+        ("tests/components/time/posix_tz.cpp", True),
+        ("tests/components/time/__init__.py", True),
+        # Platform override: tests/components/<component>/<domain>/__init__.py
+        ("tests/components/template/sensor/__init__.py", True),
+        # pytest-only files do not shape the C++ test binary
+        ("tests/components/socket/conftest.py", False),
+        ("tests/components/socket/test_socket.py", False),
+        ("tests/components/time/test.esp32-idf.yaml", False),
+        ("esphome/core/time.cpp", False),
+        ("esphome/config.py", False),
+        ("script/helpers.py", False),
+        ("README.md", False),
+    ],
+)
+def test_filter_cpp_unit_test_files(file_path: str, expected: bool) -> None:
+    """Test which changed files can affect a component's C++ unit test build."""
+    assert helpers.filter_cpp_unit_test_files(file_path) is expected
+
+
+@pytest.fixture
+def cpp_unit_test_tree(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Fake repo root where time, sntp and api have C++ unit tests.
+
+    homeassistant depends on time but has no C++ tests, so it must be
+    dropped from the selection; socket has only pytest files.
+    """
+    tests_dir = tmp_path / "tests" / "components"
+    for component in ("time", "sntp", "api"):
+        (tests_dir / component).mkdir(parents=True)
+        (tests_dir / component / f"{component}.cpp").write_text("")
+    (tests_dir / "homeassistant").mkdir()
+    (tests_dir / "homeassistant" / "__init__.py").write_text("")
+    (tests_dir / "socket").mkdir()
+    (tests_dir / "socket" / "conftest.py").write_text("")
+    monkeypatch.setattr(helpers, "root_path", str(tmp_path))
+    monkeypatch.setattr(
+        helpers,
+        "create_components_graph",
+        lambda: {"time": ["homeassistant", "sntp"]},
+    )
+    return tmp_path
+
+
+@pytest.mark.parametrize(
+    ("files", "expected"),
+    [
+        # Component changes expand to dependents with C++ tests
+        (["esphome/components/time/posix_tz.cpp"], ["sntp", "time"]),
+        (["esphome/components/time/__init__.py"], ["sntp", "time"]),
+        # Dependent without C++ tests is dropped
+        (["esphome/components/homeassistant/__init__.py"], []),
+        # Test changes select only that component
+        (["tests/components/time/posix_tz.cpp"], ["time"]),
+        (["tests/components/time/__init__.py"], ["time"]),
+        (["tests/components/homeassistant/__init__.py"], []),
+        (["tests/components/socket/conftest.py"], []),
+        (["tests/components/time/test.esp32-idf.yaml"], []),
+        (
+            ["esphome/components/time/__init__.py", "tests/components/api/api.cpp"],
+            ["api", "sntp", "time"],
+        ),
+        ([], []),
+    ],
+)
+@pytest.mark.usefixtures("cpp_unit_test_tree")
+def test_get_cpp_changed_components(files: list[str], expected: list[str]) -> None:
+    """Test that C++ and Python component changes select the right unit tests."""
+    assert helpers.get_cpp_changed_components(files) == expected
+
+
+def test_get_cpp_changed_components_independent_of_cwd(
+    cpp_unit_test_tree: Path,
+    tmp_path_factory: pytest.TempPathFactory,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test directories resolve against root_path, not the current directory."""
+    monkeypatch.chdir(tmp_path_factory.mktemp("elsewhere"))
+    assert helpers.get_cpp_changed_components(
+        ["tests/components/time/__init__.py"]
+    ) == ["time"]