Merge pull request #19334 from esphome/bump-2026.9.0b5

2026.9.0b5

Doxyfile

@@ -48,7 +48,7 @@ PROJECT_NAME           = ESPHome
 # could be handy for archiving the generated documentation or if some version
 # control system is used.
 
-PROJECT_NUMBER         = 2026.9.0b4
+PROJECT_NUMBER         = 2026.9.0b5
 
 # Using the PROJECT_BRIEF tag one can provide an optional one line description
 # for a project that appears at the top of each page and should give viewer a

esphome/components/binary_sensor/__init__.py

@@ -452,7 +452,9 @@ _BINARY_SENSOR_SCHEMA = (
             cv.Optional(
                 CONF_DEVICE_CLASS, visibility=cv.Visibility.ADVANCED
             ): validate_device_class,
-            cv.Optional(CONF_FILTERS): validate_filters,
+            cv.Optional(
+                CONF_FILTERS, visibility=cv.Visibility.ADVANCED
+            ): validate_filters,
             cv.Optional(CONF_ON_PRESS): automation.validate_automation({}),
             cv.Optional(CONF_ON_RELEASE): automation.validate_automation({}),
             cv.Optional(CONF_ON_CLICK): cv.All(

esphome/components/bk72xx_ble/__init__.py

@@ -23,7 +23,7 @@ public ble_api.h.
 import logging
 
 import esphome.codegen as cg
-from esphome.components import libretiny
+from esphome.components import libretiny, wifi
 from esphome.components.libretiny.const import (
     FAMILY_BK7231N,
     FAMILY_BK7231Q,
@@ -84,6 +84,15 @@ def _final_validate(config: ConfigType) -> None:
     # which run on a BLE 4.2 board. The hard error is raised at codegen.
     if msg := _unsupported_family_message(libretiny.get_libretiny_family()):
         _LOGGER.warning("%s (this configuration cannot compile)", msg)
+    # Any wifi power_save_mode other than NONE also arms the Beken SDK's MCU
+    # sleep. With the BLE controller running, that sleep never wakes up once the
+    # station is stopped (adapter restart after failed roams, wifi.disable): the
+    # device is dead until a power cycle (esphome#18592). Keep power save off
+    # until LibreTiny ships the SDK-side fix (libretiny-eu/libretiny#414).
+    wifi.force_power_save_off(
+        "with BLE running, the Beken SDK's MCU sleep halts the device once the "
+        "station is stopped (https://github.com/esphome/esphome/issues/18592)"
+    )
 
 
 FINAL_VALIDATE_SCHEMA = _final_validate

esphome/components/esp8266_pwm/esp8266_pwm.h

@@ -29,7 +29,7 @@ class ESP8266PWM final : public output::FloatOutput, public Component {
   void write_state(float state) override;
 
   InternalGPIOPin *pin_;
-  float frequency_{1000.0};
+  float frequency_{1000.0};  // Keep in sync with DEFAULT_FREQUENCY in output.py
   /// Cache last output level for dynamic frequency updating
   float last_output_{0.0};
 };

esphome/components/esp8266_pwm/output.py

@@ -22,6 +22,10 @@ ESP8266PWM = esp8266_pwm_ns.class_("ESP8266PWM", output.FloatOutput, cg.Componen
 SetFrequencyAction = esp8266_pwm_ns.class_("SetFrequencyAction", automation.Action)
 validate_frequency = cv.All(cv.frequency, cv.float_range(min=1.0e-6))
 
+# Schema default that also matches the C++ initializer in esp8266_pwm.h; codegen
+# skips the setter when the config equals it.
+DEFAULT_FREQUENCY = 1000.0
+
 CONFIG_SCHEMA = cv.All(
     output.FLOAT_OUTPUT_SCHEMA.extend(
         {
@@ -29,7 +33,7 @@ CONFIG_SCHEMA = cv.All(
             cv.Required(CONF_PIN): cv.All(
                 pins.internal_gpio_output_pin_schema, valid_pwm_pin
             ),
-            cv.Optional(CONF_FREQUENCY, default="1kHz"): validate_frequency,
+            cv.Optional(CONF_FREQUENCY, default=DEFAULT_FREQUENCY): validate_frequency,
         }
     ).extend(cv.COMPONENT_SCHEMA),
     cv.require_framework_version(
@@ -48,7 +52,9 @@ async def to_code(config: ConfigType) -> None:
     pin = await cg.gpio_pin_expression(config[CONF_PIN])
     cg.add(var.set_pin(pin))
 
-    cg.add(var.set_frequency(config[CONF_FREQUENCY]))
+    # Skip the setter when the config matches the C++ initializer (DEFAULT_FREQUENCY).
+    if (frequency := config[CONF_FREQUENCY]) != DEFAULT_FREQUENCY:
+        cg.add(var.set_frequency(frequency))
 
 
 @automation.register_action(

esphome/components/ethernet/__init__.py

@@ -420,7 +420,9 @@ def _validate(config: ConfigType) -> ConfigType:
 BASE_SCHEMA = cv.Schema(
     {
         cv.GenerateID(): cv.declare_id(EthernetComponent),
-        cv.Optional(CONF_MANUAL_IP): MANUAL_IP_SCHEMA,
+        cv.Optional(
+            CONF_MANUAL_IP, visibility=cv.Visibility.ADVANCED
+        ): MANUAL_IP_SCHEMA,
         cv.Optional(CONF_DOMAIN, default=".local"): cv.domain_name,
         cv.Optional(CONF_USE_ADDRESS): cv.string_strict,
         cv.Optional(CONF_MAC_ADDRESS): cv.mac_address,

esphome/components/file/image.py

@@ -42,7 +42,7 @@ from esphome.const import (
     CONF_TYPE,
     CONF_URL,
 )
-from esphome.core import CORE, HexInt
+from esphome.core import HexInt
 from esphome.cpp_generator import MockObj, MockObjClass
 from esphome.external_files import RemoteFile
 from esphome.types import ConfigType
@@ -76,16 +76,18 @@ def compute_local_image_path(value: str | ConfigType) -> Path:
     return external_files.compute_local_file_path(DOMAIN, url)
 
 
-def local_path(value: str | ConfigType) -> str:
-    value = value[CONF_PATH] if isinstance(value, dict) else value
-    return str(CORE.relative_config_path(value))
+def local_path(value: Path | ConfigType) -> Path:
+    # cv.file_ has already resolved the path against the config dir.
+    return value[CONF_PATH] if isinstance(value, dict) else value
 
 
-def download_file(url: str, path: Path) -> str:
+def download_file(url: str, path: Path) -> Path:
     # The shared NETWORK_TIMEOUT applies; a per-caller timeout would be
     # silently ignored on a per-run memo hit anyway (memos key by path).
     external_files.download_content(url, path)
-    return str(path)
+    # Keep the Path: config-hash normalizes Path values under the data dir,
+    # which a str would dump verbatim and break the CLI/add-on comparison.
+    return path
 
 
 def _gh_svg_url_path(mdi_id: str, source: str) -> tuple[str, Path]:
@@ -93,13 +95,13 @@ def _gh_svg_url_path(mdi_id: str, source: str) -> tuple[str, Path]:
     return MDI_SOURCES[source] + mdi_id + ".svg", base_dir / f"{mdi_id}.svg"
 
 
-def download_gh_svg(value: str | ConfigType, source: str) -> str:
+def download_gh_svg(value: str | ConfigType, source: str) -> Path:
     mdi_id = value[CONF_ICON] if isinstance(value, dict) else value
     url, path = _gh_svg_url_path(mdi_id, source)
     return download_file(url, path)
 
 
-def download_image(value: str | ConfigType) -> str:
+def download_image(value: str | ConfigType) -> Path:
     value = value[CONF_URL] if isinstance(value, dict) else value
     return download_file(value, compute_local_image_path(value))
 
@@ -147,7 +149,7 @@ def _extract_entry_ref(entry: ConfigType) -> RemoteFile | None:
 PREFETCH_FILES = external_files.single_stage_prefetch(_extract_entry_ref)
 
 
-def validate_file_shorthand(value: Any) -> str:
+def validate_file_shorthand(value: Any) -> Path:
     value = cv.string_strict(value)
     if (remote := _parse_remote_shorthand(value)) is not None:
         return download_file(remote.url, remote.path)
@@ -165,7 +167,7 @@ LOCAL_SCHEMA = cv.All(
 
 
 def mdi_schema(source: str) -> cv.All:
-    def validate_mdi(value: ConfigType) -> str:
+    def validate_mdi(value: ConfigType) -> Path:
         return download_gh_svg(value, source)
 
     return cv.All(

esphome/components/gpio/switch/__init__.py

@@ -15,9 +15,13 @@ CONFIG_SCHEMA = (
     .extend(
         {
             cv.Required(CONF_PIN): pins.gpio_output_pin_schema,
-            cv.Optional(CONF_INTERLOCK): cv.ensure_list(cv.use_id(switch.Switch)),
             cv.Optional(
-                CONF_INTERLOCK_WAIT_TIME, default="0ms"
+                CONF_INTERLOCK, visibility=cv.Visibility.ADVANCED
+            ): cv.ensure_list(cv.use_id(switch.Switch)),
+            cv.Optional(
+                CONF_INTERLOCK_WAIT_TIME,
+                default="0ms",
+                visibility=cv.Visibility.ADVANCED,
             ): cv.positive_time_period_milliseconds,
         }
     )

esphome/components/light/__init__.py

@@ -340,6 +340,10 @@ RESTORE_MODES = {
     "RESTORE_AND_ON": LightRestoreMode.LIGHT_RESTORE_AND_ON,
 }
 
+# Schema default that also matches the C++ initializer in light_state.h; codegen
+# skips the setter when the config equals it.
+DEFAULT_FLASH_TRANSITION_LENGTH = "0s"
+
 LIGHT_SCHEMA = (
     cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMMAND_COMPONENT_SCHEMA)
@@ -387,7 +391,7 @@ BRIGHTNESS_ONLY_LIGHT_SCHEMA = LIGHT_SCHEMA.extend(
             CONF_DEFAULT_TRANSITION_LENGTH, default="1s"
         ): cv.positive_time_period_milliseconds,
         cv.Optional(
-            CONF_FLASH_TRANSITION_LENGTH, default="0s"
+            CONF_FLASH_TRANSITION_LENGTH, default=DEFAULT_FLASH_TRANSITION_LENGTH
         ): cv.positive_time_period_milliseconds,
         cv.Optional(CONF_EFFECTS): validate_effects(MONOCHROMATIC_EFFECTS),
     }
@@ -502,9 +506,12 @@ async def setup_light_core_(light_var, config, output_var):
         default_transition_length := config.get(CONF_DEFAULT_TRANSITION_LENGTH)
     ) is not None:
         cg.add(light_var.set_default_transition_length(default_transition_length))
+    # Skip the setter when the config matches the C++ initializer.
     if (
         flash_transition_length := config.get(CONF_FLASH_TRANSITION_LENGTH)
-    ) is not None:
+    ) is not None and flash_transition_length != cv.time_period(
+        DEFAULT_FLASH_TRANSITION_LENGTH
+    ):
         cg.add(light_var.set_flash_transition_length(flash_transition_length))
     if (gamma_correct := config.get(CONF_GAMMA_CORRECT)) is not None:
         cg.add(light_var.set_gamma_correct(gamma_correct))
@@ -514,7 +521,8 @@ async def setup_light_core_(light_var, config, output_var):
     effects = await cg.build_registry_list(
         EFFECTS_REGISTRY, config.get(CONF_EFFECTS, [])
     )
-    cg.add(light_var.add_effects(effects))
+    if effects:
+        cg.add(light_var.add_effects(effects))
 
     for conf in config.get(CONF_ON_TURN_ON, []):
         trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], light_var)

esphome/components/light/light_state.h

@@ -356,7 +356,7 @@ class LightState : public EntityBase, public Component {
   /// Default transition length for all transitions in ms.
   uint32_t default_transition_length_{};
   /// Transition length to use for flash transitions.
-  uint32_t flash_transition_length_{};
+  uint32_t flash_transition_length_{};  // Keep in sync with DEFAULT_FLASH_TRANSITION_LENGTH in __init__.py
   /// Gamma correction factor for the light.
   float gamma_correct_{};
 #ifdef USE_LIGHT_GAMMA_LUT

esphome/components/logger/__init__.py

@@ -362,12 +362,13 @@ async def to_code(config: ConfigType) -> None:
     # pre_setup() switches on uart_ to decide which hardware to initialize
     # (e.g. UART0 vs USB_SERIAL_JTAG). Without this, uart_ is still the
     # default UART_SELECTION_UART0 and the wrong hardware gets initialized.
-    if CONF_HARDWARE_UART in config:
-        cg.add(
-            log.set_uart_selection(
-                HARDWARE_UART_TO_UART_SELECTION[config[CONF_HARDWARE_UART]]
-            )
-        )
+    # uart_ is UART0 in C++ except on LibreTiny where it is DEFAULT; skip the
+    # setter when the config matches it.
+    cpp_default_uart = DEFAULT if CORE.is_libretiny else UART0
+    if (
+        hardware_uart := config.get(CONF_HARDWARE_UART)
+    ) is not None and hardware_uart != cpp_default_uart:
+        cg.add(log.set_uart_selection(HARDWARE_UART_TO_UART_SELECTION[hardware_uart]))
     # pre_setup() sets global_logger and must run before any other code
     # that may call ESP_LOG* (e.g. setup_preferences contains ESP_LOGVV).
     cg.add(log.pre_setup())

esphome/components/logger/logger.h

@@ -352,10 +352,10 @@ class Logger final : public Component {
   // Group smaller types together at the end
   uint8_t current_level_{ESPHOME_LOG_LEVEL_VERY_VERBOSE};
 #if defined(USE_ESP32) || defined(USE_ESP8266) || defined(USE_RP2) || defined(USE_ZEPHYR)
-  UARTSelection uart_{UART_SELECTION_UART0};
+  UARTSelection uart_{UART_SELECTION_UART0};  // Must match cpp_default_uart in __init__.py
 #endif
 #ifdef USE_LIBRETINY
-  UARTSelection uart_{UART_SELECTION_DEFAULT};
+  UARTSelection uart_{UART_SELECTION_DEFAULT};  // Must match cpp_default_uart in __init__.py
 #endif
 #if defined(USE_ESP32) || defined(USE_HOST) || defined(USE_LIBRETINY) || defined(USE_ZEPHYR)
   bool main_task_recursion_guard_{false};

esphome/components/modbus/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Callable
 import logging
 from typing import Any, Literal, NamedTuple
 
@@ -48,6 +49,8 @@ ModbusServerDevice = modbus_ns.class_("ModbusServerDevice")
 CommandOptions = modbus_ns.struct("CommandOptions")
 MULTI_CONF = True
 
+CONF_ALLOW_BROADCAST_READ = "allow_broadcast_read"
+CONF_EXPECT_BROADCAST_WRITE_RESPONSE = "expect_broadcast_write_response"
 CONF_ROLE = "role"
 CONF_MODBUS_ID = "modbus_id"
 CONF_SEND_WAIT_TIME = "send_wait_time"
@@ -56,6 +59,28 @@ CONF_TURNAROUND_TIME = "turnaround_time"
 MODBUS_ROLES = ["client", "server"]
 
 
+# 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})
+
+# Codes the hub refuses at address 0; keep in sync with modbus::helpers::is_function_code_broadcastable().
+_NON_BROADCASTABLE_FUNCTION_CODES = frozenset(
+    {0x01, 0x02, 0x03, 0x04, 0x14, 0x15, 0x17, 0x18}
+)
+
+
+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 (the runtime hub never queues one:
+    queue_pdu() refuses them). Keep in sync with modbus::helpers::is_function_code_write()."""
+    return function_code & 0x7F in _WRITE_FUNCTION_CODES
+
+
+def is_function_code_broadcastable(function_code: int) -> bool:
+    """True if the hub accepts the function code at address 0 without allow_broadcast_read."""
+    return function_code & 0x7F not in _NON_BROADCASTABLE_FUNCTION_CODES
+
+
 class _CommandOption(NamedTuple):
     """One per-command option forwarded to the hub (modbus::CommandOptions)."""
 
@@ -64,14 +89,47 @@ class _CommandOption(NamedTuple):
     validator: Any  # the static (non-templatable) validator for the key
     cpp_type: Any  # the C++ type the value is generated as
     default: Any
+    # Function codes the hub honours the option on; it is stripped from any other.
+    applies_to: Callable[[int], bool]
+    requires_broadcast_address: bool = False
 
 
-# Per-direction command options. Single-sourcing the schema and the setter generation here keeps
-# them from drifting; the C++ side must add the matching field per the rules documented on
-# CommandOptions (modbus.h).
+def _not_write(function_code: int) -> bool:
+    return not is_function_code_write(function_code)
+
+
+def _not_broadcastable(function_code: int) -> bool:
+    return not is_function_code_broadcastable(function_code)
+
+
+# Per-direction command options, single-sourced so the schema, setters and applicability rule cannot
+# drift; the C++ side adds the matching field per the rules on CommandOptions (modbus.h).
 _COMMAND_OPTIONS: dict[str, list[_CommandOption]] = {
-    "read": [_CommandOption(CONF_CONTINUOUS, "continuous", cv.boolean, bool, False)],
-    "write": [],
+    "read": [
+        _CommandOption(
+            CONF_CONTINUOUS, "continuous", cv.boolean, bool, False, _not_write
+        ),
+        _CommandOption(
+            CONF_ALLOW_BROADCAST_READ,
+            "allow_broadcast_read",
+            cv.boolean,
+            bool,
+            False,
+            _not_broadcastable,
+            requires_broadcast_address=True,
+        ),
+    ],
+    "write": [
+        _CommandOption(
+            CONF_EXPECT_BROADCAST_WRITE_RESPONSE,
+            "expect_broadcast_write_response",
+            cv.boolean,
+            bool,
+            False,
+            is_function_code_broadcastable,
+            requires_broadcast_address=True,
+        ),
+    ],
 }
 
 
@@ -82,32 +140,75 @@ 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 broadcast_only_option_keys() -> list[str]:
+    return [
+        option.conf_key
+        for options in _COMMAND_OPTIONS.values()
+        for option in options
+        if option.requires_broadcast_address
+    ]
 
 
-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 (the runtime hub never queues one:
-    queue_pdu() refuses them). Keep in sync with modbus::helpers::is_function_code_write()."""
-    return function_code & 0x7F in _WRITE_FUNCTION_CODES
+def reject_broadcast_options_for_unicast(
+    address_key: str,
+) -> Callable[[ConfigType], ConfigType]:
+    """Reject a broadcast-only option set true on a literal address other than 0."""
+
+    def validator(config: ConfigType) -> ConfigType:
+        address = config.get(address_key)
+        if not isinstance(address, int) or address == BROADCAST_ADDRESS:
+            return config
+        for key in broadcast_only_option_keys():
+            if config.get(key) is True:
+                raise cv.Invalid(
+                    f"'{key}' only applies to the broadcast address; set '{address_key}: 0' or "
+                    f"remove the option.",
+                    path=[key],
+                )
+        return config
+
+    return validator
+
+
+def reject_inapplicable_command_options(
+    pdu_key: str,
+) -> Callable[[ConfigType], ConfigType]:
+    """Reject an option set true that the hub would strip from a literal PDU's function code."""
+
+    def validator(config: ConfigType) -> ConfigType:
+        pdu = config[pdu_key]
+        if not isinstance(pdu, list):
+            return config
+        for direction in _COMMAND_OPTIONS:
+            for option in _command_options(direction):
+                if config.get(option.conf_key) is True and not option.applies_to(
+                    pdu[0]
+                ):
+                    raise cv.Invalid(
+                        f"'{option.conf_key}: true' does not apply to function code "
+                        f"0x{pdu[0]:02X}",
+                        path=[option.conf_key],
+                    )
+        return config
+
+    return validator
 
 
 def command_options_schema(
-    *, direction: Literal["read", "write"], templatable: bool = False
+    *,
+    direction: Literal["read", "write"],
+    templatable: bool = False,
+    function_code: int | None = None,
 ) -> dict[cv.Optional, Any]:
-    """Schema fragment for the per-command options a component forwards to the hub
-    (modbus::CommandOptions). Extend this into any schema that queues commands. Keys are
-    direction-specific so a schema never offers an option the hub would strip (e.g.
-    continuous on a write); the write side has no options yet. For actions (templatable=True the
-    keys also accept lambdas), register the values with register_templatable_command_options().
+    """Schema fragment for the per-command options of one direction; `function_code` (a typed
+    action's fixed code) leaves out the options that do not apply to it.
     """
     return {
         cv.Optional(option.conf_key, default=option.default): (
             cv.templatable(option.validator) if templatable else option.validator
         )
         for option in _command_options(direction)
+        if function_code is None or option.applies_to(function_code)
     }
 
 
@@ -130,6 +231,25 @@ def command_options_expression(
     )
 
 
+def add_command_options(
+    var: MockObj,
+    setter: str,
+    config: ConfigType,
+    *,
+    direction: Literal["read", "write"],
+) -> None:
+    """Emit `var.<setter>(<options>)` for a config validated with command_options_schema() of the
+    same direction, skipped when every option is at its C++ default."""
+    if all(
+        config.get(option.conf_key, option.default) == option.default
+        for option in _command_options(direction)
+    ):
+        return
+    cg.add(
+        getattr(var, setter)(command_options_expression(config, direction=direction))
+    )
+
+
 async def register_templatable_command_options(
     var: MockObj, config: ConfigType, args: TemplateArgsType, direction: str
 ) -> None:

esphome/components/modbus/modbus.cpp

@@ -832,7 +832,7 @@ void ModbusClientHub::send_next_frame_() {
   }
 
   cmd->sent();
-  if (cmd->frame.address() == BROADCAST_ADDRESS) {
+  if (cmd->fire_and_forget()) {
     // A broadcast (address 0) is never answered (Modbus 4.1), so it is fire-and-forget: on_sent above
     // reports the transmission, and the entry then retires with no terminal callback instead of
     // occupying the waiting slot until the send-wait timeout expires. The turnaround delay already
@@ -1074,11 +1074,6 @@ bool ModbusClientHub::queue_pdu(uint8_t address, std::span<const uint8_t> pdu, M
     return false;
   }
 
-  if (address == BROADCAST_ADDRESS && !helpers::is_function_code_broadcastable(pdu[0])) {
-    ESP_LOGW(TAG, "Broadcast refused for function 0x%X: a broadcast (address 0) is never answered", pdu[0]);
-    return false;
-  }
-
   // Normalize the caller's options in place (the param is a by-value copy) so everything stored or
   // merged below carries effective options, never the raw request.
   // continuous is ignored for every mutating code (re-writing a value forever is never intended).
@@ -1086,6 +1081,24 @@ bool ModbusClientHub::queue_pdu(uint8_t address, std::span<const uint8_t> pdu, M
     ESP_LOGW(TAG, "continuous is ignored for a mutating function (0x%X, address %" PRIu8 ")", pdu[0], address);
     options.continuous = false;
   }
+  if (address != BROADCAST_ADDRESS) {
+    options.allow_broadcast_read = false;
+    options.expect_broadcast_write_response = false;
+  } else {
+    const bool broadcastable = helpers::is_function_code_broadcastable(pdu[0]);
+    if (options.allow_broadcast_read && broadcastable) {
+      ESP_LOGV(TAG, "allow_broadcast_read is ignored for function 0x%X: it is broadcastable", pdu[0]);
+      options.allow_broadcast_read = false;
+    }
+    if (options.expect_broadcast_write_response && !broadcastable) {
+      ESP_LOGV(TAG, "expect_broadcast_write_response is ignored for function 0x%X: it is not broadcastable", pdu[0]);
+      options.expect_broadcast_write_response = false;
+    }
+    if (!broadcastable && !options.allow_broadcast_read) {
+      ESP_LOGW(TAG, "Broadcast refused for function 0x%X: a broadcast (address 0) is never answered", pdu[0]);
+      return false;
+    }
+  }
 
   // A duplicate of a live entry with the same owner is not queued twice; it resolves against that
   // entry: anonymous -> dropped; continuous incoming -> convert the entry to a poll; one-shot onto a
@@ -1126,6 +1139,7 @@ bool ModbusClientHub::queue_pdu(uint8_t address, std::span<const uint8_t> pdu, M
       ESP_LOGV(TAG, "Frame already active for %" PRIu8 ", request absorbed (pending %" PRIu8 ")", address,
                item.pending);
     }
+    item.options.expect_broadcast_write_response |= options.expect_broadcast_write_response;
     return true;
   }
 

esphome/components/modbus/modbus.h

@@ -111,11 +111,15 @@ enum class FrameState : uint8_t {
 // Per-command send options. Append-only; pass via designated initializers ({.continuous = true}).
 // A new field reaches the queue with no plumbing but arrives inert until it defines three rules:
 // normalization in queue_pdu(), a merge rule for duplicate absorption, and teardown in
-// retire()/silent_retire().
+// retire()/silent_retire(). Bit-packed: stored per entry, controller and writer entity, passed by value.
 struct CommandOptions {
   // A continuous poll lives in the queue until cancelled or failed; ignored for mutating codes.
-  bool continuous{false};
+  bool continuous : 1 {false};
+  // Wait for the reply to a read sent to address 0, for a device that answers the broadcast address.
+  bool allow_broadcast_read : 1 {false};
+  bool expect_broadcast_write_response : 1 {false};
 };
+static_assert(sizeof(CommandOptions) == 1, "CommandOptions must stay one byte");
 
 struct ModbusDeviceCommand {
   ModbusClientDevice *device;
@@ -158,6 +162,10 @@ struct ModbusDeviceCommand {
     this->pending = 0;
     this->device = nullptr;
   }
+  bool fire_and_forget() const {
+    return this->frame.address() == BROADCAST_ADDRESS && !this->options.allow_broadcast_read &&
+           !this->options.expect_broadcast_write_response;
+  }
   // Fire-and-forget completion for a broadcast (address 0): the frame was transmitted (on_sent already
   // fired), but a broadcast is never answered (Modbus 4.1), so the entry retires with no terminal callback.
   void complete_broadcast() {
@@ -191,7 +199,8 @@ struct ModbusDeviceCommand {
     } else if (!this->waiting_state()) {  // an already-retired shell stays put; off the wire -> RETIRED
       this->state = FrameState::RETIRED;
     }
-    this->options = {};  // reset every option
+    // Only continuous ends with the clear; the delivery flags must survive for a granted retry.
+    this->options.continuous = false;
   }
 
   // True while the entry is still waiting for a response
@@ -534,27 +543,27 @@ class ModbusClientDevice {
     return this->queue_pdu(
         helpers::create_read_pdu(FunctionCode::READ_DISCRETE_INPUTS, start_address, number_of_inputs), options);
   }
-  bool write_single_register(uint16_t start_address, uint16_t value) {
-    return this->queue_pdu(helpers::create_write_single_register_pdu(start_address, value));
+  bool write_single_register(uint16_t start_address, uint16_t value, CommandOptions options = {}) {
+    return this->queue_pdu(helpers::create_write_single_register_pdu(start_address, value), options);
   }
-  bool write_single_coil(uint16_t address, bool value) {
-    return this->queue_pdu(helpers::create_write_single_coil_pdu(address, value));
+  bool write_single_coil(uint16_t address, bool value, CommandOptions options = {}) {
+    return this->queue_pdu(helpers::create_write_single_coil_pdu(address, value), options);
   }
-  bool write_multiple_registers(uint16_t start_address, std::span<const uint16_t> values) {
+  bool write_multiple_registers(uint16_t start_address, std::span<const uint16_t> values, CommandOptions options = {}) {
     // Empty goes to the full-size builder so the rejection log names this method's limit, not the small one's.
     if (!values.empty() && values.size() <= helpers::MAX_FEW_REGISTERS)
-      return this->queue_pdu(helpers::create_write_few_registers_pdu(start_address, values));
-    return this->queue_pdu(helpers::create_write_registers_pdu(start_address, values));
+      return this->queue_pdu(helpers::create_write_few_registers_pdu(start_address, values), options);
+    return this->queue_pdu(helpers::create_write_registers_pdu(start_address, values), options);
   }
   /// Note: std::vector<bool> cannot bind to std::span<const bool>; use a contiguous bool container or the packed
   /// overload.
-  bool write_multiple_coils(uint16_t start_address, std::span<const bool> values) {
-    return this->queue_pdu(helpers::create_write_coils_pdu(start_address, values));
+  bool write_multiple_coils(uint16_t start_address, std::span<const bool> values, CommandOptions options = {}) {
+    return this->queue_pdu(helpers::create_write_coils_pdu(start_address, values), options);
   }
   /// Packed variant: a PackedBits view (the same layout on_read_coils() delivers), so
   /// read-modify-write needs no unpack/repack.
-  bool write_multiple_coils(uint16_t start_address, PackedBits bits) {
-    return this->queue_pdu(helpers::create_write_coils_pdu(start_address, bits));
+  bool write_multiple_coils(uint16_t start_address, PackedBits bits, CommandOptions options = {}) {
+    return this->queue_pdu(helpers::create_write_coils_pdu(start_address, bits), options);
   }
   /// FC 0x17: the read-back is delivered through on_read_holding_registers(), and a device exception
   /// (typically a rejected write half) arrives there too via its status - one callback handles both

esphome/components/modbus_client/__init__.py

@@ -7,7 +7,6 @@ from esphome.components import modbus
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_ADDRESS,
-    CONF_CONTINUOUS,
     CONF_COUNT,
     CONF_ID,
     CONF_ON_ERROR,
@@ -158,24 +157,6 @@ _ACTION_BASE_SCHEMA = cv.Schema(
 )
 
 
-def _no_continuous_on_write(config: ConfigType) -> ConfigType:
-    """Reject `continuous: true` on a static write PDU: continuous polling only applies to reads.
-    Only the fully-static case is decidable here; the hub strips the flag from mutating PDUs at
-    runtime, so a templated pdu or continuous falls through to that backstop."""
-    pdu = config[CONF_PDU]
-    if (
-        isinstance(pdu, list)
-        and config.get(CONF_CONTINUOUS) is True
-        and modbus.is_function_code_write(pdu[0])
-    ):
-        raise cv.Invalid(
-            f"'{CONF_CONTINUOUS}: true' does not apply to a write PDU (function code "
-            f"0x{pdu[0]:02X}); continuous polling only applies to reads",
-            path=[CONF_CONTINUOUS],
-        )
-    return config
-
-
 MODBUS_CLIENT_SEND_SCHEMA = cv.All(
     _ACTION_BASE_SCHEMA.extend(
         {
@@ -186,10 +167,12 @@ MODBUS_CLIENT_SEND_SCHEMA = cv.All(
                 )
             ),
             **modbus.command_options_schema(direction="read", templatable=True),
+            **modbus.command_options_schema(direction="write", templatable=True),
             cv.Optional(CONF_ON_RESPONSE): _handler_schema(),
         }
     ),
-    _no_continuous_on_write,
+    modbus.reject_inapplicable_command_options(CONF_PDU),
+    modbus.reject_broadcast_options_for_unicast(CONF_ADDRESS),
 )
 
 
@@ -261,8 +244,7 @@ async def register_client_action(
             var.get_not_sent_trigger(), [(_PDU_SPAN, "request")], not_sent_conf
         )
     # Wire any command options the action's schema opted into (e.g. continuous on reads). Pass the
-    # matching direction so a write action never generates a read option's setter; the write side
-    # has no options yet, so this is a no-op there.
+    # matching direction so a write action never generates a read option's setter.
     await modbus.register_templatable_command_options(
         var, config, args, command_direction
     )
@@ -279,6 +261,8 @@ async def modbus_client_send_to_code(config, action_id, template_arg, args):
     var = cg.new_Pvariable(action_id, template_arg)
     template_ = await cg.templatable(config[CONF_PDU], args, _PDU_BUFFER)
     cg.add(var.set_pdu(template_))
+    # The read set is wired by register_client_action() below.
+    await modbus.register_templatable_command_options(var, config, args, "write")
     return await register_client_action(
         var,
         config,
@@ -353,6 +337,7 @@ def _read_schema(max_count: int) -> cv.All:
             }
         ),
         _no_address_overflow(CONF_COUNT),
+        modbus.reject_broadcast_options_for_unicast(CONF_ADDRESS),
     )
 
 
@@ -364,21 +349,35 @@ def _write_multiple_schema(item: Callable[[Any], Any], max_values: int) -> cv.Al
                 cv.Required(CONF_VALUES): cv.templatable(
                     cv.All(cv.ensure_list(item), cv.Length(min=1, max=max_values))
                 ),
+                **modbus.command_options_schema(direction="write", templatable=True),
             }
         ),
         _no_address_overflow(CONF_VALUES),
+        modbus.reject_broadcast_options_for_unicast(CONF_ADDRESS),
     )
 
 
 _READ_REGISTERS_SCHEMA = _read_schema(modbus.MAX_NUM_OF_REGISTERS_TO_READ)
 
-_WRITE_SINGLE_REGISTER_SCHEMA = _TYPED_ACTION_SCHEMA.extend(
-    {cv.Required(CONF_VALUE): cv.templatable(cv.hex_uint16_t)}
+_WRITE_SINGLE_REGISTER_SCHEMA = cv.All(
+    _TYPED_ACTION_SCHEMA.extend(
+        {
+            cv.Required(CONF_VALUE): cv.templatable(cv.hex_uint16_t),
+            **modbus.command_options_schema(direction="write", templatable=True),
+        }
+    ),
+    modbus.reject_broadcast_options_for_unicast(CONF_ADDRESS),
 )
 
 # A coil is one bit, so the value is a boolean - the wire only carries 0x0000 or 0xFF00.
-_WRITE_SINGLE_COIL_SCHEMA = _TYPED_ACTION_SCHEMA.extend(
-    {cv.Required(CONF_VALUE): cv.templatable(cv.boolean)}
+_WRITE_SINGLE_COIL_SCHEMA = cv.All(
+    _TYPED_ACTION_SCHEMA.extend(
+        {
+            cv.Required(CONF_VALUE): cv.templatable(cv.boolean),
+            **modbus.command_options_schema(direction="write", templatable=True),
+        }
+    ),
+    modbus.reject_broadcast_options_for_unicast(CONF_ADDRESS),
 )
 
 
@@ -542,10 +541,15 @@ _READ_WRITE_MULTIPLE_REGISTERS_SCHEMA = cv.All(
                     cv.Length(min=1, max=modbus.MAX_NUM_OF_REGISTERS_TO_WRITE_RW),
                 )
             ),
+            # 0x17 counts as a read at address 0, so it takes allow_broadcast_read only.
+            **modbus.command_options_schema(
+                direction="read", templatable=True, function_code=0x17
+            ),
         }
     ),
     _no_address_overflow(CONF_READ_COUNT, CONF_READ_ADDRESS),
     _no_address_overflow(CONF_VALUES, CONF_WRITE_ADDRESS),
+    modbus.reject_broadcast_options_for_unicast(CONF_ADDRESS),
 )
 
 

esphome/components/modbus_client/modbus_client.h

@@ -85,18 +85,36 @@ template<typename... Ts> class ClientActionBase : public Action<Ts...>, public m
 /// builds its static struct; declaring the values here instead of per action means a new read option
 /// costs one TEMPLATABLE_VALUE plus one field below, and every read action picks it up.
 /// The read/write split mirrors _COMMAND_OPTIONS in the modbus component's Python
-/// (command_options_schema(direction="read") adds exactly these keys). When a write-side option
-/// arrives it gets a WriteCommandOptions twin, so write actions never carry read-only members.
+/// (command_options_schema(direction="read") adds exactly these keys); WriteCommandOptions is the twin.
 template<typename... Ts> class ReadCommandOptions {
  public:
   // Poll: re-queue after each success until downgraded (replay with false) or failed. The hub strips
   // it for mutating function codes at the door (see modbus::CommandOptions).
   TEMPLATABLE_VALUE(bool, continuous)
+  TEMPLATABLE_VALUE(bool, allow_broadcast_read)
 
  protected:
   /// The options for this send, with every templatable value resolved against the action's arguments.
   modbus::CommandOptions command_options_(const Ts &...x) const {
-    return {.continuous = this->continuous_.value(x...)};
+    return {.continuous = this->continuous_.value(x...),
+            .allow_broadcast_read = this->allow_broadcast_read_.value(x...)};
+  }
+};
+
+/// The write-side per-command options (command_options_schema(direction="write") adds exactly these keys).
+template<typename... Ts> class WriteCommandOptions {
+ public:
+  TEMPLATABLE_VALUE(bool, expect_broadcast_write_response)
+
+ protected:
+  /// Resolves every write option into `options`, so send's merge of both sets stays exhaustive.
+  void apply_write_command_options_(modbus::CommandOptions &options, const Ts &...x) const {
+    options.expect_broadcast_write_response = this->expect_broadcast_write_response_.value(x...);
+  }
+  modbus::CommandOptions write_command_options_(const Ts &...x) const {
+    modbus::CommandOptions options{};
+    this->apply_write_command_options_(options, x...);
+    return options;
   }
 };
 
@@ -107,8 +125,11 @@ template<typename... Ts> class ReadCommandOptions {
 /// modbus::helpers::create_*_pdu() builders and return it directly (smaller builder results convert).
 /// A PduBuffer drops bytes past modbus::MAX_PDU_SIZE without reporting it (the hub's oversize check
 /// cannot fire - that limit is the capacity), so an over-long lambda-built PDU is silently truncated.
+/// A raw PDU may be a read or a write, so this action carries both option sets.
 template<typename... Ts>
-class ModbusClientSendAction : public ClientActionBase<Ts...>, public ReadCommandOptions<Ts...> {
+class ModbusClientSendAction : public ClientActionBase<Ts...>,
+                               public ReadCommandOptions<Ts...>,
+                               public WriteCommandOptions<Ts...> {
  public:
   TEMPLATABLE_VALUE(modbus::helpers::PduBuffer, pdu)
 
@@ -116,7 +137,11 @@ class ModbusClientSendAction : public ClientActionBase<Ts...>, public ReadComman
     return &this->response_trigger_;
   }
 
-  void play(const Ts &...x) override { this->send_or_resolve_(this->pdu_.value(x...), this->command_options_(x...)); }
+  void play(const Ts &...x) override {
+    modbus::CommandOptions options = this->command_options_(x...);
+    this->apply_write_command_options_(options, x...);
+    this->send_or_resolve_(this->pdu_.value(x...), options);
+  }
 
   void on_response(std::span<const uint8_t> request_pdu, std::span<const uint8_t> response_pdu) override {
     this->response_trigger_.trigger(request_pdu, response_pdu);
@@ -218,7 +243,8 @@ template<typename... Ts> class ReadBitsAction : public TypedClientActionBase<Ts.
 
 /// modbus_client.write_single_register: on_response is the acknowledgement (the ack only echoes the
 /// request, so it carries no arguments).
-template<typename... Ts> class WriteSingleRegisterAction : public TypedClientActionBase<Ts...> {
+template<typename... Ts>
+class WriteSingleRegisterAction : public TypedClientActionBase<Ts...>, public WriteCommandOptions<Ts...> {
  public:
   TEMPLATABLE_VALUE(uint16_t, start_address)
   TEMPLATABLE_VALUE(uint16_t, value)
@@ -227,7 +253,8 @@ template<typename... Ts> class WriteSingleRegisterAction : public TypedClientAct
 
   void play(const Ts &...x) override {
     this->send_or_resolve_(
-        modbus::helpers::create_write_single_register_pdu(this->start_address_.value(x...), this->value_.value(x...)));
+        modbus::helpers::create_write_single_register_pdu(this->start_address_.value(x...), this->value_.value(x...)),
+        this->write_command_options_(x...));
   }
   void on_write_single_register(uint16_t address, uint16_t value, modbus::ResponseStatus status) override {
     if (modbus::succeeded(status))
@@ -240,7 +267,8 @@ template<typename... Ts> class WriteSingleRegisterAction : public TypedClientAct
 
 /// modbus_client.write_single_coil: on_response is the acknowledgement (no arguments). A coil holds one
 /// bit, so the value is a bool - the wire only ever carries 0x0000 or 0xFF00.
-template<typename... Ts> class WriteSingleCoilAction : public TypedClientActionBase<Ts...> {
+template<typename... Ts>
+class WriteSingleCoilAction : public TypedClientActionBase<Ts...>, public WriteCommandOptions<Ts...> {
  public:
   TEMPLATABLE_VALUE(uint16_t, start_address)
   TEMPLATABLE_VALUE(bool, value)
@@ -249,7 +277,8 @@ template<typename... Ts> class WriteSingleCoilAction : public TypedClientActionB
 
   void play(const Ts &...x) override {
     this->send_or_resolve_(
-        modbus::helpers::create_write_single_coil_pdu(this->start_address_.value(x...), this->value_.value(x...)));
+        modbus::helpers::create_write_single_coil_pdu(this->start_address_.value(x...), this->value_.value(x...)),
+        this->write_command_options_(x...));
   }
   void on_write_single_coil(uint16_t address, bool value, modbus::ResponseStatus status) override {
     if (modbus::succeeded(status))
@@ -264,7 +293,8 @@ template<typename... Ts> class WriteSingleCoilAction : public TypedClientActionB
 /// A `values:` list is emitted as a flash array and sent straight from there; only a lambda builds a
 /// vector, and only when it runs. Same split as canbus's send action, and for the same reason: a static
 /// list must not allocate on every play().
-template<typename... Ts> class WriteMultipleRegistersAction : public TypedClientActionBase<Ts...> {
+template<typename... Ts>
+class WriteMultipleRegistersAction : public TypedClientActionBase<Ts...>, public WriteCommandOptions<Ts...> {
  public:
   TEMPLATABLE_VALUE(uint16_t, start_address)
 
@@ -288,11 +318,13 @@ template<typename... Ts> class WriteMultipleRegistersAction : public TypedClient
     // the empty PDU then resolves via on_not_sent like any refused send.
     if (this->len_ >= 0) {
       this->send_or_resolve_(modbus::helpers::create_write_registers_pdu(
-          start, std::span<const uint16_t>(this->values_.data, static_cast<size_t>(this->len_))));
+                                 start, std::span<const uint16_t>(this->values_.data, static_cast<size_t>(this->len_))),
+                             this->write_command_options_(x...));
       return;
     }
     const std::vector<uint16_t> values = this->values_.func(x...);
-    this->send_or_resolve_(modbus::helpers::create_write_registers_pdu(start, std::span<const uint16_t>(values)));
+    this->send_or_resolve_(modbus::helpers::create_write_registers_pdu(start, std::span<const uint16_t>(values)),
+                           this->write_command_options_(x...));
   }
   void on_write_multiple_registers(uint16_t start_address, std::span<const uint16_t> registers,
                                    modbus::ResponseStatus status) override {
@@ -313,7 +345,8 @@ template<typename... Ts> class WriteMultipleRegistersAction : public TypedClient
 /// A `values:` list is packed into wire layout at code-generation time and stored in flash, so play()
 /// neither allocates nor packs. A lambda returns std::vector<bool> - already a bit per coil rather than
 /// a byte - and is packed into a stack buffer on the way to the builder.
-template<typename... Ts> class WriteMultipleCoilsAction : public TypedClientActionBase<Ts...> {
+template<typename... Ts>
+class WriteMultipleCoilsAction : public TypedClientActionBase<Ts...>, public WriteCommandOptions<Ts...> {
  public:
   TEMPLATABLE_VALUE(uint16_t, start_address)
 
@@ -334,13 +367,16 @@ template<typename... Ts> class WriteMultipleCoilsAction : public TypedClientActi
     const uint16_t start = this->start_address_.value(x...);
     if (this->count_ >= 0) {
       const auto count = static_cast<uint16_t>(this->count_);
-      this->send_or_resolve_(modbus::helpers::create_write_coils_pdu(
-          start,
-          modbus::PackedBits(std::span<const uint8_t>(this->values_.packed, modbus::packed_bit_bytes(count)), count)));
+      this->send_or_resolve_(
+          modbus::helpers::create_write_coils_pdu(
+              start, modbus::PackedBits(std::span<const uint8_t>(this->values_.packed, modbus::packed_bit_bytes(count)),
+                                        count)),
+          this->write_command_options_(x...));
       return;
     }
     // The builder packs and bound-checks; an over-long set is rejected and logged there.
-    this->send_or_resolve_(modbus::helpers::create_write_coils_pdu(start, this->values_.func(x...)));
+    this->send_or_resolve_(modbus::helpers::create_write_coils_pdu(start, this->values_.func(x...)),
+                           this->write_command_options_(x...));
   }
   void on_write_multiple_coils(uint16_t start_address, modbus::PackedBits bits,
                                modbus::ResponseStatus status) override {
@@ -359,7 +395,8 @@ template<typename... Ts> class WriteMultipleCoilsAction : public TypedClientActi
 
 /// modbus_client.read_write_multiple_registers (FC 0x17): writes one register block and reads another back in
 /// one transaction (write first, per Modbus 6.17). on_response delivers the read-back words as `values`.
-template<typename... Ts> class ReadWriteMultipleRegistersAction : public TypedClientActionBase<Ts...> {
+template<typename... Ts>
+class ReadWriteMultipleRegistersAction : public TypedClientActionBase<Ts...>, public ReadCommandOptions<Ts...> {
  public:
   TEMPLATABLE_VALUE(uint16_t, read_address)
   TEMPLATABLE_VALUE(uint16_t, read_count)
@@ -385,13 +422,15 @@ template<typename... Ts> class ReadWriteMultipleRegistersAction : public TypedCl
     // An out-of-range read/write count builds an empty PDU (the builder logs why), resolving via on_not_sent.
     if (this->len_ >= 0) {
       this->send_or_resolve_(modbus::helpers::create_read_write_multiple_registers_pdu(
-          read_start, read_count, write_start,
-          std::span<const uint16_t>(this->values_.data, static_cast<size_t>(this->len_))));
+                                 read_start, read_count, write_start,
+                                 std::span<const uint16_t>(this->values_.data, static_cast<size_t>(this->len_))),
+                             this->command_options_(x...));
       return;
     }
     const std::vector<uint16_t> values = this->values_.func(x...);
     this->send_or_resolve_(modbus::helpers::create_read_write_multiple_registers_pdu(
-        read_start, read_count, write_start, std::span<const uint16_t>(values)));
+                               read_start, read_count, write_start, std::span<const uint16_t>(values)),
+                           this->command_options_(x...));
   }
   // The 0x17 response carries only the read block, so the hub dispatch delivers it as a holding-register read.
   void on_read_registers(modbus::EntityType entity_type, uint16_t start_address, std::span<const uint16_t> registers,

esphome/components/modbus_controller/__init__.py

@@ -103,12 +103,20 @@ def _warn_removed_options(config: ConfigType) -> ConfigType:
 
 
 def _reject_broadcast_address(config: ConfigType) -> ConfigType:
-    """A modbus_controller polls one device, so its address cannot be the broadcast address (0):
-    a broadcast is never answered (Modbus 4.1), so no register could ever read back."""
+    """Address 0 is rejected unless allow_broadcast_read, which in turn requires address 0."""
+    if config[modbus.CONF_ALLOW_BROADCAST_READ]:
+        if config.get(CONF_ADDRESS) != modbus.BROADCAST_ADDRESS:
+            raise cv.Invalid(
+                f"'{modbus.CONF_ALLOW_BROADCAST_READ}' only applies to the broadcast address; "
+                f"set 'address: 0' or remove the option.",
+                [modbus.CONF_ALLOW_BROADCAST_READ],
+            )
+        return config
     modbus.reject_broadcast_address(
         config.get(CONF_ADDRESS),
         "a modbus_controller device address",
-        "Assign the unit address of the device you want to poll.",
+        "Assign the unit address of the device you want to poll, or set allow_broadcast_read if "
+        "it answers address 0.",
         [CONF_ADDRESS],
     )
     return config
@@ -346,12 +354,52 @@ def _reject_continuous_write_custom_pdu(config: ConfigType) -> None:
         )
 
 
+def _reject_broadcastable_custom_pdu(config: ConfigType) -> None:
+    """A broadcastable custom_pdu under an address-0 controller is a real broadcast, never answered."""
+    pdu = config.get(CONF_CUSTOM_PDU)
+    if pdu is None or not modbus.is_function_code_broadcastable(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_ADDRESS) == modbus.BROADCAST_ADDRESS
+        and controller.get(modbus.CONF_ALLOW_BROADCAST_READ) is True
+    ):
+        raise cv.Invalid(
+            f"a '{CONF_CUSTOM_PDU}' with function code 0x{pdu[0] & 0x7F:02X} is a real broadcast at "
+            f"address 0 and is never answered, so it can't be polled through the "
+            f"'{controller[CONF_ID]}' modbus_controller; 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."""
+    """Final-validate for the platforms that accept custom_pdu."""
     migrate_custom_command(config)
     _reject_continuous_write_custom_pdu(config)
+    _reject_broadcastable_custom_pdu(config)
+
+
+def _reject_write_option_off_broadcast(config: ConfigType) -> None:
+    if not any(config.get(key) is True for key in modbus.broadcast_only_option_keys()):
+        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_ADDRESS) != modbus.BROADCAST_ADDRESS:
+        raise cv.Invalid(
+            f"'{modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE}' only applies when the "
+            f"'{controller[CONF_ID]}' modbus_controller is at address 0; remove the option.",
+            [modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE],
+        )
+
+
+def validate_writer_item(config: ConfigType) -> None:
+    """Final-validate for the writer platforms (number, output, select, switch)."""
+    if CONF_CUSTOM_PDU in config or CONF_CUSTOM_COMMAND in config:
+        validate_custom_pdu_item(config)
+    _reject_write_option_off_broadcast(config)
 
 
 def _final_validate(config: ConfigType) -> None:
@@ -448,11 +496,7 @@ async def to_code(config: ConfigType) -> None:
     await cg.register_component(var, config)
     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")
-        )
-    )
+    modbus.add_command_options(var, "set_read_options", config, direction="read")
     await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
 
 

esphome/components/modbus_controller/modbus_controller.cpp

@@ -24,7 +24,7 @@ void WriterDevice::warn_write_buffer_deprecated(const LogString *platform, uint1
 bool WriterDevice::send_raw_frame_deprecated(std::span<const uint8_t> frame) {
   if (frame.empty())
     return false;
-  return this->parent_->queue_pdu(frame[0], frame.subspan(1), this);
+  return this->parent_->queue_pdu(frame[0], frame.subspan(1), this, this->write_options_);
 }
 
 void ControllerDevice::set_controller(ModbusController *controller) {
@@ -234,10 +234,13 @@ void ModbusCommandItem::on_sent(std::span<const uint8_t> request_pdu) {
   // (frame[0]), which may differ from this controller's. (unqueue_command() is a no-op for a poll.)
   // A custom polling command sends its PDU to this controller's own address, so only a factory custom
   // command (a raw frame staged in payload) can carry a different address byte.
+  // An address-0 read with allow_broadcast_read is answered, so it keeps its terminal callback.
   uint8_t wire_address = this->address_;
   if (this->function_code_ == FunctionCode::CUSTOM && !this->payload.empty())
     wire_address = this->payload.data()[0];
-  if (wire_address == modbus::BROADCAST_ADDRESS)
+  const bool answered = this->controller_->read_options().allow_broadcast_read &&
+                        !modbus::helpers::is_function_code_broadcastable(request_pdu[0]);
+  if (wire_address == modbus::BROADCAST_ADDRESS && !answered)
     this->controller_->unqueue_command(this);
 }
 
@@ -285,8 +288,8 @@ 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()) {
+  // One-shots never poll, so only the broadcast flag is passed (the hub strips it from writes).
+  if (!item->send({.allow_broadcast_read = this->read_options_.allow_broadcast_read})) {
     // 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()),
              item->register_address());
@@ -340,7 +343,7 @@ void ModbusController::update() {
   if (this->can_send()) {
     for (auto &poll : this->polling_devices_) {
       ESP_LOGVV(TAG, "Updating range 0x%X", poll.register_address());
-      // read_options_ carries the controller's continuous flag (the offline probe above sends it too).
+      // read_options_ carries the controller's read-side flags (the offline probe above sends them too).
       // A refusal is already logged by the hub; note the affected range for controller-level diagnostics.
       if (!poll.queue(this->read_options_)) {
         ESP_LOGD(TAG, "Poll refused by hub for range 0x%X", poll.register_address());

esphome/components/modbus_controller/modbus_controller.h

@@ -280,10 +280,11 @@ class ControllerDevice : protected modbus::ModbusClientDevice {
 
   void notify_online_(std::span<const uint8_t> request_pdu);
 
-  /// Write-path state owned by WriterEntity's forwarders, stored here so both bools land in the base's
-  /// tail padding instead of adding a word to every writer entity. The warn flag leaves in 2027.3.0.
-  bool dispatched_{false};
-  bool write_buffer_deprecated_warned_{false};
+  /// Write-path state for WriterEntity's forwarders, packed into the base's tail padding. The warn flag
+  /// leaves in 2027.3.0.
+  bool dispatched_ : 1 {false};
+  bool write_buffer_deprecated_warned_ : 1 {false};
+  modbus::CommandOptions write_options_{};
   ModbusController *controller_{nullptr};
 };
 
@@ -305,6 +306,8 @@ class WriterDevice final : public ControllerDevice {
   bool dispatched() const { return this->dispatched_; }
   void set_dispatched() { this->dispatched_ = true; }
   void clear_dispatched() { this->dispatched_ = false; }
+  modbus::CommandOptions write_options() const { return this->write_options_; }
+  void set_write_options(modbus::CommandOptions options) { this->write_options_ = options; }
   /// Warn once per entity that filling the write_lambda buffer parameter is deprecated (the entity is now the
   /// command - call a write helper / queue_pdu() on `item` instead). The buffer parameter is removed in 2027.3.0.
   void warn_write_buffer_deprecated(const LogString *platform, uint16_t address);
@@ -326,27 +329,29 @@ class WriterEntity {
   /// Whether the lambda called a request helper since the last clear_dispatched_(). Deliberately records
   /// the call, not the hub's accept/refuse: a refused lambda write must not fall through to the default write.
   bool dispatched() const { return this->device_.dispatched(); }
+  void set_write_options(modbus::CommandOptions options) { this->device_.set_write_options(options); }
   bool write_single_register(uint16_t address, uint16_t value) {
     this->device_.set_dispatched();
-    return this->device_.write_single_register(address, value);
+    return this->device_.write_single_register(address, value, this->device_.write_options());
   }
   bool write_single_coil(uint16_t address, bool value) {
     this->device_.set_dispatched();
-    return this->device_.write_single_coil(address, value);
+    return this->device_.write_single_coil(address, value, this->device_.write_options());
   }
   bool write_multiple_registers(uint16_t address, std::span<const uint16_t> values) {
     this->device_.set_dispatched();
-    return this->device_.write_multiple_registers(address, values);
+    return this->device_.write_multiple_registers(address, values, this->device_.write_options());
   }
   bool write_multiple_coils(uint16_t address, std::span<const bool> values) {
     this->device_.set_dispatched();
-    return this->device_.write_multiple_coils(address, values);
+    return this->device_.write_multiple_coils(address, values, this->device_.write_options());
   }
   bool write_multiple_coils(uint16_t address, modbus::PackedBits bits) {
     this->device_.set_dispatched();
-    return this->device_.write_multiple_coils(address, bits);
+    return this->device_.write_multiple_coils(address, bits, this->device_.write_options());
   }
-  bool queue_pdu(std::span<const uint8_t> pdu, modbus::CommandOptions options = {}) {
+  bool queue_pdu(std::span<const uint8_t> pdu) { return this->queue_pdu(pdu, this->device_.write_options()); }
+  bool queue_pdu(std::span<const uint8_t> pdu, modbus::CommandOptions options) {
     this->device_.set_dispatched();
     return this->device_.queue_pdu(pdu, options);
   }

esphome/components/modbus_controller/number/__init__.py

@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import number
+from esphome.components import modbus, number
 from esphome.components.modbus.helpers import (
     MODBUS_WRITE_REGISTER_TYPE,
     SENSOR_VALUE_TYPE,
@@ -23,8 +23,8 @@ from .. import (
     add_modbus_base_properties,
     modbus_calc_properties,
     modbus_controller_ns,
-    validate_custom_pdu_item,
     validate_range_reuse_migration,
+    validate_writer_item,
 )
 from ..const import (
     CONF_BITMASK,
@@ -84,6 +84,7 @@ CONFIG_SCHEMA = cv.All(
             cv.Optional(CONF_STEP, default=1): cv.positive_float,
             cv.Optional(CONF_MULTIPLY, default=1.0): cv.float_,
             cv.Optional(CONF_USE_WRITE_MULTIPLE, default=False): cv.boolean,
+            **modbus.command_options_schema(direction="write"),
         }
     ),
     validate_min_max,
@@ -91,7 +92,7 @@ CONFIG_SCHEMA = cv.All(
     validate_range_reuse_migration,
 )
 
-FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
+FINAL_VALIDATE_SCHEMA = validate_writer_item
 
 
 async def to_code(config: ConfigType) -> None:
@@ -122,6 +123,7 @@ async def to_code(config: ConfigType) -> None:
     cg.add(parent.add_sensor_item(var))
     await add_modbus_base_properties(var, config, ModbusNumber)
     cg.add(var.set_use_write_mutiple(config[CONF_USE_WRITE_MULTIPLE]))
+    modbus.add_command_options(var, "set_write_options", config, direction="write")
     if CONF_WRITE_LAMBDA in config:
         template_ = await cg.process_lambda(
             config[CONF_WRITE_LAMBDA],

esphome/components/modbus_controller/output/__init__.py

@@ -1,7 +1,7 @@
 import logging
 
 import esphome.codegen as cg
-from esphome.components import output
+from esphome.components import modbus, output
 from esphome.components.modbus.helpers import (
     SENSOR_VALUE_TYPE,
     PduBuffer,
@@ -18,6 +18,7 @@ from .. import (
     modbus_calc_properties,
     modbus_controller_ns,
     reject_odd_holding_write_offset,
+    validate_writer_item,
 )
 from ..const import (
     CONF_CUSTOM_COMMAND,
@@ -79,6 +80,7 @@ CONFIG_SCHEMA = cv.All(
                     ),
                     cv.Optional(CONF_WRITE_LAMBDA): cv.returning_lambda,
                     cv.Optional(CONF_USE_WRITE_MULTIPLE, default=False): cv.boolean,
+                    **modbus.command_options_schema(direction="write"),
                 }
             ),
             "holding": cv.All(
@@ -98,6 +100,7 @@ CONFIG_SCHEMA = cv.All(
                         cv.Optional(CONF_WRITE_LAMBDA): cv.returning_lambda,
                         cv.Optional(CONF_MULTIPLY, default=1.0): cv.float_,
                         cv.Optional(CONF_USE_WRITE_MULTIPLE, default=False): cv.boolean,
+                        **modbus.command_options_schema(direction="write"),
                     }
                 ),
                 reject_odd_holding_write_offset,
@@ -111,6 +114,9 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
+FINAL_VALIDATE_SCHEMA = validate_writer_item
+
+
 async def to_code(config: ConfigType) -> None:
     byte_offset = modbus_calc_properties(config)
     # Binary Output
@@ -153,6 +159,7 @@ async def to_code(config: ConfigType) -> None:
     await output.register_output(var, config)
     parent = await cg.get_variable(config[CONF_MODBUS_CONTROLLER_ID])
     cg.add(var.set_use_write_mutiple(config[CONF_USE_WRITE_MULTIPLE]))
+    modbus.add_command_options(var, "set_write_options", config, direction="write")
     cg.add(var.set_parent(parent))
     if write_template:
         cg.add(var.set_write_template(write_template))

esphome/components/modbus_controller/select/__init__.py

@@ -2,7 +2,7 @@ from collections.abc import Callable
 from typing import Any
 
 import esphome.codegen as cg
-from esphome.components import select
+from esphome.components import modbus, select
 from esphome.components.modbus.helpers import SENSOR_VALUE_TYPE, RegisterValues
 import esphome.config_validation as cv
 from esphome.const import CONF_ADDRESS, CONF_ID, CONF_LAMBDA, CONF_OPTIMISTIC
@@ -15,6 +15,7 @@ from .. import (
     modbus_controller_ns,
     validate_range_reuse_migration,
     validate_skip_updates_deprecated,
+    validate_writer_item,
 )
 from ..const import (
     CONF_FORCE_NEW_RANGE,
@@ -77,6 +78,7 @@ CONFIG_SCHEMA = cv.All(
             cv.Optional(CONF_REGISTER_COUNT): cv.positive_int,
             cv.Required(CONF_OPTIONSMAP): ensure_option_map(),
             cv.Optional(CONF_USE_WRITE_MULTIPLE, default=False): cv.boolean,
+            **modbus.command_options_schema(direction="write"),
             cv.Optional(CONF_OPTIMISTIC, default=False): cv.boolean,
             cv.Optional(CONF_LAMBDA): cv.returning_lambda,
             cv.Optional(CONF_WRITE_LAMBDA): cv.returning_lambda,
@@ -86,6 +88,9 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
+FINAL_VALIDATE_SCHEMA = validate_writer_item
+
+
 async def to_code(config: ConfigType) -> None:
     options_map = config[CONF_OPTIONSMAP]
 
@@ -104,6 +109,7 @@ async def to_code(config: ConfigType) -> None:
     cg.add(parent.add_sensor_item(var))
     cg.add(var.set_parent(parent))
     cg.add(var.set_use_write_mutiple(config[CONF_USE_WRITE_MULTIPLE]))
+    modbus.add_command_options(var, "set_write_options", config, direction="write")
     cg.add(var.set_optimistic(config[CONF_OPTIMISTIC]))
 
     if CONF_LAMBDA in config:

esphome/components/modbus_controller/switch/__init__.py

@@ -1,5 +1,5 @@
 import esphome.codegen as cg
-from esphome.components import switch
+from esphome.components import modbus, switch
 from esphome.components.modbus.helpers import MODBUS_REGISTER_TYPE, PduBuffer
 import esphome.config_validation as cv
 from esphome.const import CONF_ADDRESS, CONF_ASSUMED_STATE, CONF_ID
@@ -13,9 +13,9 @@ from .. import (
     modbus_calc_properties,
     modbus_controller_ns,
     reject_odd_holding_write_offset,
-    validate_custom_pdu_item,
     validate_modbus_register,
     validate_range_reuse_migration,
+    validate_writer_item,
 )
 from ..const import (
     CONF_BITMASK,
@@ -51,6 +51,7 @@ CONFIG_SCHEMA = cv.All(
             cv.Optional(CONF_ASSUMED_STATE, default=False): cv.boolean,
             cv.Optional(CONF_REGISTER_TYPE): cv.enum(MODBUS_REGISTER_TYPE),
             cv.Optional(CONF_USE_WRITE_MULTIPLE, default=False): cv.boolean,
+            **modbus.command_options_schema(direction="write"),
             cv.Optional(CONF_WRITE_LAMBDA): cv.returning_lambda,
         }
     ),
@@ -59,7 +60,7 @@ CONFIG_SCHEMA = cv.All(
     validate_range_reuse_migration,
 )
 
-FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
+FINAL_VALIDATE_SCHEMA = validate_writer_item
 
 
 async def to_code(config: ConfigType) -> None:
@@ -78,6 +79,7 @@ async def to_code(config: ConfigType) -> None:
     paren = await cg.get_variable(config[CONF_MODBUS_CONTROLLER_ID])
     cg.add(var.set_parent(paren))
     cg.add(var.set_use_write_mutiple(config[CONF_USE_WRITE_MULTIPLE]))
+    modbus.add_command_options(var, "set_write_options", config, direction="write")
     assumed_state = config[CONF_ASSUMED_STATE]
     cg.add(var.set_assumed_state(assumed_state))
     if not assumed_state:

esphome/components/number/__init__.py

@@ -174,6 +174,10 @@ NumberInRangeCondition = number_ns.class_(
 
 NumberMode = number_ns.enum("NumberMode")
 
+# Schema default that also matches the C++ initializer in number_traits.h; codegen
+# skips the setter when the config equals it.
+DEFAULT_MODE = "AUTO"
+
 NUMBER_MODES = {
     "AUTO": NumberMode.NUMBER_MODE_AUTO,
     "BOX": NumberMode.NUMBER_MODE_BOX,
@@ -216,7 +220,7 @@ _NUMBER_SCHEMA = (
                 CONF_UNIT_OF_MEASUREMENT, visibility=cv.Visibility.ADVANCED
             ): validate_unit_of_measurement,
             cv.Optional(
-                CONF_MODE, default="AUTO", visibility=cv.Visibility.ADVANCED
+                CONF_MODE, default=DEFAULT_MODE, visibility=cv.Visibility.ADVANCED
             ): cv.enum(NUMBER_MODES, upper=True),
             cv.Optional(
                 CONF_DEVICE_CLASS, visibility=cv.Visibility.ADVANCED
@@ -286,10 +290,10 @@ async def setup_number_core_(
     cg.add(var.traits.set_max_value(max_value))
     cg.add(var.traits.set_step(step))
 
-    # Only set if non-default to avoid bloating setup() function
-    # (mode_ is initialized to NUMBER_MODE_AUTO in the header)
-    if config[CONF_MODE] != NumberMode.NUMBER_MODE_AUTO:
-        cg.add(var.traits.set_mode(config[CONF_MODE]))
+    # Skip the setter when the config matches the C++ initializer (DEFAULT_MODE).
+    # The validated value is the enum key string, not the C++ enum expression.
+    if (mode := config[CONF_MODE]) != DEFAULT_MODE:
+        cg.add(var.traits.set_mode(mode))
 
     CORE.add_job(_build_number_automations, var, config)
 

esphome/components/number/number_traits.h

@@ -31,7 +31,7 @@ class NumberTraits {
   float min_value_ = NAN;
   float max_value_ = NAN;
   float step_ = NAN;
-  NumberMode mode_{NUMBER_MODE_AUTO};
+  NumberMode mode_{NUMBER_MODE_AUTO};  // Keep in sync with DEFAULT_MODE in __init__.py
 };
 
 }  // namespace esphome::number

esphome/components/output/__init__.py

@@ -53,12 +53,17 @@ async def setup_output_platform_(obj, config):
     if CONF_POWER_SUPPLY in config:
         power_supply_ = await cg.get_variable(config[CONF_POWER_SUPPLY])
         cg.add(obj.set_power_supply(power_supply_))
-    if CONF_MAX_POWER in config:
+    # The C++ initializers are max_power 1.0 and min_power 0.0; skip the setter when
+    # the config matches them. The define stays whenever the key is present because
+    # platforms such as ac_dimmer read the scaling fields directly.
+    if (max_power := config.get(CONF_MAX_POWER)) is not None:
         cg.add_define("USE_OUTPUT_FLOAT_POWER_SCALING")
-        cg.add(obj.set_max_power(config[CONF_MAX_POWER]))
-    if CONF_MIN_POWER in config:
+        if max_power != 1.0:
+            cg.add(obj.set_max_power(max_power))
+    if (min_power := config.get(CONF_MIN_POWER)) is not None:
         cg.add_define("USE_OUTPUT_FLOAT_POWER_SCALING")
-        cg.add(obj.set_min_power(config[CONF_MIN_POWER]))
+        if min_power != 0.0:
+            cg.add(obj.set_min_power(min_power))
     # Only emit when zero_means_zero is actually enabled. The schema defaults to False
     # so this key is always present; emitting unconditionally would force
     # USE_OUTPUT_FLOAT_POWER_SCALING on for every output, defeating the gate.

esphome/components/output/float_output.h

@@ -123,6 +123,7 @@ class FloatOutput : public BinaryOutput {
   virtual void write_state(float state) = 0;
 
 #ifdef USE_OUTPUT_FLOAT_POWER_SCALING
+  // Codegen skips the setters for these values; keep in sync with output/__init__.py
   float max_power_{1.0f};
   float min_power_{0.0f};
   bool zero_means_zero_{false};

esphome/components/pmsa003i/pmsa003i.cpp

@@ -88,7 +88,11 @@ void PMSA003IComponent::update() {
 bool PMSA003IComponent::read_data_(PM25AQIData *data) {
   uint8_t buffer[COUNT_DATA_BYTES];
 
-  this->read_bytes_raw(buffer, COUNT_DATA_BYTES);
+  const i2c::ErrorCode error = this->read(buffer, COUNT_DATA_BYTES);
+  if (error != i2c::ERROR_OK) {
+    ESP_LOGW(TAG, "I2C error %d", error);
+    return false;
+  }
 
   // https://github.com/adafruit/Adafruit_PM25AQI
 

esphome/components/sensor/__init__.py

@@ -344,7 +344,9 @@ _SENSOR_SCHEMA = (
                 cv.requires_component("mqtt"),
                 cv.Any(None, cv.positive_time_period_milliseconds),
             ),
-            cv.Optional(CONF_FILTERS): validate_filters,
+            cv.Optional(
+                CONF_FILTERS, visibility=cv.Visibility.ADVANCED
+            ): validate_filters,
             cv.Optional(CONF_ON_VALUE): automation.validate_automation({}),
             cv.Optional(CONF_ON_RAW_VALUE): automation.validate_automation({}),
             cv.Optional(CONF_ON_VALUE_RANGE): automation.validate_automation(

esphome/components/template/water_heater/template_water_heater.cpp

@@ -1,6 +1,8 @@
 #include "template_water_heater.h"
 #include "esphome/core/log.h"
 
+#include <cmath>
+
 namespace esphome::template_ {
 
 static const char *const TAG = "template.water_heater";
@@ -45,9 +47,12 @@ water_heater::WaterHeaterTraits TemplateWaterHeater::traits() {
 void TemplateWaterHeater::loop() {
   bool changed = false;
 
+  // NAN is passed through so a source that has no value yet shows as unknown, but NAN never
+  // equals NAN, so an already-NAN value must not count as a change or it would republish forever.
   auto curr_temp = this->current_temperature_f_.call();
   if (curr_temp.has_value()) {
-    if (*curr_temp != this->current_temperature_) {
+    if (*curr_temp != this->current_temperature_ &&
+        !(std::isnan(*curr_temp) && std::isnan(this->current_temperature_))) {
       this->current_temperature_ = *curr_temp;
       changed = true;
     }
@@ -55,7 +60,8 @@ void TemplateWaterHeater::loop() {
 
   auto target_temp = this->target_temperature_f_.call();
   if (target_temp.has_value()) {
-    if (*target_temp != this->target_temperature_) {
+    if (*target_temp != this->target_temperature_ &&
+        !(std::isnan(*target_temp) && std::isnan(this->target_temperature_))) {
       this->target_temperature_ = *target_temp;
       changed = true;
     }

esphome/components/text_sensor/__init__.py

@@ -148,7 +148,9 @@ _TEXT_SENSOR_SCHEMA = (
             cv.Optional(
                 CONF_DEVICE_CLASS, visibility=cv.Visibility.ADVANCED
             ): validate_device_class,
-            cv.Optional(CONF_FILTERS): validate_filters,
+            cv.Optional(
+                CONF_FILTERS, visibility=cv.Visibility.ADVANCED
+            ): validate_filters,
             cv.Optional(CONF_ON_VALUE): automation.validate_automation({}),
             cv.Optional(CONF_ON_RAW_VALUE): automation.validate_automation({}),
         }

esphome/components/web_server/__init__.py

@@ -56,6 +56,10 @@ CONF_SORTING_GROUPS = "sorting_groups"
 CONF_SORTING_WEIGHT = "sorting_weight"
 CONF_ALLOWED_ORIGINS = "allowed_origins"
 
+# Schema default that also matches the C++ initializer in web_server_base.h; codegen
+# skips the setter when the config equals it.
+DEFAULT_PORT = 80
+
 
 web_server_ns = cg.esphome_ns.namespace("web_server")
 WebServer = web_server_ns.class_("WebServer", cg.Component, cg.Controller)
@@ -251,7 +255,7 @@ CONFIG_SCHEMA = cv.All(
     cv.Schema(
         {
             cv.GenerateID(): cv.declare_id(WebServer),
-            cv.Optional(CONF_PORT, default=80): cv.port,
+            cv.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
             cv.Optional(CONF_VERSION, default=2): cv.one_of(1, 2, 3, int=True),
             cv.Optional(CONF_CSS_URL): cv.string,
             cv.Optional(CONF_CSS_INCLUDE): cv.file_,
@@ -379,9 +383,11 @@ async def to_code(config: ConfigType) -> None:
 
     version = config[CONF_VERSION]
 
-    cg.add(paren.set_port(config[CONF_PORT]))
+    # Skip the setter when the config matches the C++ initializer (DEFAULT_PORT).
+    if (port := config[CONF_PORT]) != DEFAULT_PORT:
+        cg.add(paren.set_port(port))
     cg.add_define("USE_WEBSERVER")
-    cg.add_define("USE_WEBSERVER_PORT", config[CONF_PORT])
+    cg.add_define("USE_WEBSERVER_PORT", port)
     cg.add_define("USE_WEBSERVER_VERSION", version)
     if version >= 2:
         # Don't compress the index HTML as the data sizes are almost the same.
@@ -395,9 +401,11 @@ async def to_code(config: ConfigType) -> None:
     # Captive portal will still be able to perform OTA updates even when this is set
     if config.get(CONF_OTA) is False:
         cg.add_define("USE_WEBSERVER_OTA_DISABLED")
-    cg.add(var.set_expose_log(config[CONF_LOG]))
+    # expose_log_ is true in C++; only emit the setter to turn it off.
     if config[CONF_LOG]:
         request_log_listener()  # Request a log listener slot for web server log streaming
+    else:
+        cg.add(var.set_expose_log(False))
     if config[CONF_ENABLE_PRIVATE_NETWORK_ACCESS]:
         cg.add_define("USE_WEBSERVER_PRIVATE_NETWORK_ACCESS")
     if (allowed_origins := config.get(CONF_ALLOWED_ORIGINS)) is not None:
@@ -433,7 +441,9 @@ async def to_code(config: ConfigType) -> None:
         path = CORE.relative_config_path(config[CONF_JS_INCLUDE])
         with path.open(encoding="utf-8") as js_file:
             add_resource_as_progmem("JS_INCLUDE", js_file.read())
-    cg.add(var.set_include_internal(config[CONF_INCLUDE_INTERNAL]))
+    # include_internal_ is false in C++; only emit the setter to turn it on.
+    if config[CONF_INCLUDE_INTERNAL]:
+        cg.add(var.set_include_internal(True))
     if CONF_LOCAL in config and config[CONF_LOCAL]:
         cg.add_define("USE_WEBSERVER_LOCAL")
     if config[CONF_COMPRESSION] == "gzip":

esphome/components/web_server_base/web_server_base.h

@@ -170,7 +170,7 @@ class WebServerBase final {
 
  protected:
   uint8_t initialized_{0};
-  uint16_t port_{80};
+  uint16_t port_{80};  // Keep in sync with DEFAULT_PORT in web_server/__init__.py
   AsyncWebServer *server_{nullptr};
   std::vector<AsyncWebHandler *> handlers_;
 #ifdef USE_WEBSERVER_AUTH

esphome/components/wifi/__init__.py

@@ -167,6 +167,9 @@ MAX_WIFI_NETWORKS = 127
 # get best-effort connection attempts. Longer timeout ensures we exhaust all options
 # before falling back to AP mode. Aligned with improv wifi_timeout default.
 DEFAULT_AP_TIMEOUT = "90s"
+DEFAULT_REBOOT_TIMEOUT = "15min"
+# Both defaults also match the C++ initializers in wifi_component.h; codegen skips
+# the setter when the config equals them.
 
 wifi_ns = cg.esphome_ns.namespace("wifi")
 EAPAuth = wifi_ns.struct("EAPAuth")
@@ -285,7 +288,9 @@ WIFI_NETWORK_BASE = cv.Schema(
         cv.Optional(CONF_SSID): cv.sensitive(cv.ssid),
         cv.Optional(CONF_PASSWORD): cv.sensitive(validate_password),
         cv.Optional(CONF_CHANNEL): validate_channel,
-        cv.Optional(CONF_MANUAL_IP): STA_MANUAL_IP_SCHEMA,
+        cv.Optional(
+            CONF_MANUAL_IP, visibility=cv.Visibility.ADVANCED
+        ): STA_MANUAL_IP_SCHEMA,
     }
 )
 
@@ -484,12 +489,14 @@ CONFIG_SCHEMA = cv.All(
             ),
             cv.Optional(CONF_SSID): cv.sensitive(cv.ssid),
             cv.Optional(CONF_PASSWORD): cv.sensitive(validate_password),
-            cv.Optional(CONF_MANUAL_IP): STA_MANUAL_IP_SCHEMA,
+            cv.Optional(
+                CONF_MANUAL_IP, visibility=cv.Visibility.ADVANCED
+            ): STA_MANUAL_IP_SCHEMA,
             cv.Optional(CONF_EAP): EAP_AUTH_SCHEMA,
             cv.Optional(CONF_AP): wifi_network_ap,
             cv.Optional(CONF_DOMAIN, default=".local"): cv.domain_name,
             cv.Optional(
-                CONF_REBOOT_TIMEOUT, default="15min"
+                CONF_REBOOT_TIMEOUT, default=DEFAULT_REBOOT_TIMEOUT
             ): cv.positive_time_period_milliseconds,
             cv.SplitDefault(
                 CONF_POWER_SAVE_MODE,
@@ -599,7 +606,8 @@ def wifi_network(config, ap, static_ip):
         cg.add(ap.set_channel(config[CONF_CHANNEL]))
     if static_ip is not None:
         cg.add(ap.set_manual_ip(manual_ip(static_ip)))
-    if CONF_PRIORITY in config:
+    # priority_ is 0 in C++; skip the setter when the config matches it.
+    if config.get(CONF_PRIORITY, 0) != 0:
         cg.add(ap.set_priority(config[CONF_PRIORITY]))
 
     return ap
@@ -648,7 +656,9 @@ async def to_code(config):
             WiFiAP(),
             lambda ap: cg.add(var.set_ap(wifi_network(conf, ap, ip_config))),
         )
-        cg.add(var.set_ap_timeout(conf[CONF_AP_TIMEOUT]))
+        # Skip the setter when the config matches the C++ initializer.
+        if (ap_timeout := conf[CONF_AP_TIMEOUT]) != cv.time_period(DEFAULT_AP_TIMEOUT):
+            cg.add(var.set_ap_timeout(ap_timeout))
         cg.add_define("USE_WIFI_AP")
 
     # ESP32: register the WiFi stack with the esp32 sdkconfig reconciler, which
@@ -664,10 +674,27 @@ async def to_code(config):
     if has_manual_ip:
         cg.add_define("USE_WIFI_MANUAL_IP")
 
-    cg.add(var.set_reboot_timeout(config[CONF_REBOOT_TIMEOUT]))
-    cg.add(var.set_power_save_mode(config[CONF_POWER_SAVE_MODE]))
-    if CONF_MIN_AUTH_MODE in config:
-        cg.add(var.set_min_auth_mode(config[CONF_MIN_AUTH_MODE]))
+    # The C++ initializers are DEFAULT_REBOOT_TIMEOUT, power save NONE and minimum
+    # auth WPA2; skip the setters when the config matches them.
+    if (reboot_timeout := config[CONF_REBOOT_TIMEOUT]) != cv.time_period(
+        DEFAULT_REBOOT_TIMEOUT
+    ):
+        cg.add(var.set_reboot_timeout(reboot_timeout))
+    if (power_save_mode := config[CONF_POWER_SAVE_MODE]) != "NONE":
+        if reasons := CORE.data.get(POWER_SAVE_OFF_REASONS_KEY):
+            _LOGGER.warning(
+                "power_save_mode %s is not applied: %s",
+                power_save_mode,
+                "; ".join(reasons),
+            )
+        else:
+            cg.add(var.set_power_save_mode(power_save_mode))
+    # From here on force_power_save_off() can no longer take effect
+    CORE.data[POWER_SAVE_APPLIED_KEY] = True
+    if (
+        min_auth_mode := config.get(CONF_MIN_AUTH_MODE)
+    ) is not None and min_auth_mode != "WPA2":
+        cg.add(var.set_min_auth_mode(min_auth_mode))
     fast_connect = config[CONF_FAST_CONNECT]
     if fast_connect[CONF_ENABLED]:
         cg.add_define("USE_WIFI_FAST_CONNECT")
@@ -825,6 +852,8 @@ async def wifi_disable_to_code(config, action_id, template_arg, args):
 
 KEEP_SCAN_RESULTS_KEY = "wifi_keep_scan_results"
 RUNTIME_POWER_SAVE_KEY = "wifi_runtime_power_save"
+POWER_SAVE_OFF_REASONS_KEY = "wifi_power_save_off_reasons"
+POWER_SAVE_APPLIED_KEY = "wifi_power_save_applied"
 RUNTIME_ROAMING_SUPPRESSION_KEY = "wifi_runtime_roaming_suppression"
 # Keys for listener counts
 IP_STATE_LISTENERS_KEY = "wifi_ip_state_listeners"
@@ -857,6 +886,25 @@ def request_wifi_scan_results_lock() -> None:
     CORE.data[SCAN_RESULTS_LOCK_KEY] = True
 
 
+def force_power_save_off(reason: str) -> None:
+    """Keep the station out of WiFi power save regardless of power_save_mode.
+
+    Components whose platform cannot run power save safely call this from their
+    final validation (FINAL_VALIDATE_SCHEMA), which always runs before any code
+    generation. Every distinct reason is kept; when the configured mode is not
+    NONE, wifi's code generation logs them and skips the mode. Calling it once
+    wifi has generated its code is too late and raises.
+    """
+    if POWER_SAVE_APPLIED_KEY in CORE.data:
+        raise EsphomeError(
+            "wifi.force_power_save_off() must be called from final validation, "
+            "before wifi generates its code"
+        )
+    reasons: list[str] = CORE.data.setdefault(POWER_SAVE_OFF_REASONS_KEY, [])
+    if reason not in reasons:
+        reasons.append(reason)
+
+
 def enable_runtime_power_save_control():
     """Enable runtime WiFi power save control.
 

esphome/components/wifi/wifi_component.h

@@ -913,11 +913,11 @@ class WiFiComponent final : public Component {
   float output_power_{NAN};
   uint32_t action_started_;
   uint32_t last_connected_{0};
-  uint32_t reboot_timeout_{};
+  uint32_t reboot_timeout_{900000};  // Keep in sync with DEFAULT_REBOOT_TIMEOUT in __init__.py
   uint32_t roaming_last_check_{0};
   uint32_t roaming_scan_end_{0};  // Timestamp when last roaming scan completed
 #ifdef USE_WIFI_AP
-  uint32_t ap_timeout_{};
+  uint32_t ap_timeout_{90000};  // Keep in sync with DEFAULT_AP_TIMEOUT in __init__.py
 #endif
 
   // 1-byte enums and integers

esphome/const.py

@@ -4,7 +4,7 @@ from enum import Enum
 
 from esphome.enum import StrEnum
 
-__version__ = "2026.9.0b4"
+__version__ = "2026.9.0b5"
 
 ALLOWED_NAME_CHARS = "abcdefghijklmnopqrstuvwxyz0123456789-_"
 VALID_SUBSTITUTIONS_CHARACTERS = (

tests/component_tests/bk72xx_ble/config/test_power_save.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: bk-power-save
+
+bk72xx:
+  board: cb2s
+
+wifi:
+  ssid: test
+  password: testtest
+  power_save_mode: high
+
+bk72xx_ble:

tests/component_tests/bk72xx_ble/test_power_save.py

@@ -0,0 +1,20 @@
+"""bk72xx_ble keeps WiFi power save off: the Beken SDK's MCU sleep does not
+wake up once the station is stopped while the BLE controller runs."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+
+def test_power_save_mode_is_not_applied_with_ble(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    main_cpp = generate_main(component_config_path("test_power_save.yaml"))
+
+    assert "bk72xx_ble::BK72xxBLE" in main_cpp
+    assert "set_power_save_mode(" not in main_cpp
+    assert "power_save_mode HIGH is not applied" in caplog.text
+    assert "issues/18592" in caplog.text

tests/component_tests/config_validation/test_advanced_visibility.py

@@ -0,0 +1,53 @@
+"""Power-user fields are marked as advanced on the shared schemas.
+
+``filters``, ``manual_ip`` and the GPIO switch interlock options are knobs
+whose defaults suit nearly every user, so a schema-aware editor should keep
+them behind its "advanced settings" disclosure rather than on the main form.
+"""
+
+from __future__ import annotations
+
+import importlib
+
+import pytest
+
+from esphome.components import binary_sensor, ethernet, sensor, text_sensor, wifi
+import esphome.config_validation as cv
+
+
+def _markers(schema: cv.Schema) -> dict[str, object]:
+    s = schema
+    if hasattr(s, "validators"):
+        # cv.All -> the schema is the first validator.
+        s = s.validators[0]
+    return {str(k): k for k in s.schema}
+
+
+def _gpio_switch_schema() -> cv.Schema:
+    return importlib.import_module("esphome.components.gpio.switch").CONFIG_SCHEMA
+
+
+@pytest.mark.parametrize(
+    ("label", "schema_factory", "fields"),
+    [
+        ("sensor", sensor.sensor_schema, ["filters"]),
+        ("binary_sensor", binary_sensor.binary_sensor_schema, ["filters"]),
+        ("text_sensor", text_sensor.text_sensor_schema, ["filters"]),
+        ("wifi_network", lambda: wifi.WIFI_NETWORK_BASE, ["manual_ip"]),
+        ("wifi", lambda: wifi.CONFIG_SCHEMA, ["manual_ip"]),
+        ("ethernet", lambda: ethernet.BASE_SCHEMA, ["manual_ip"]),
+        ("gpio_switch", _gpio_switch_schema, ["interlock", "interlock_wait_time"]),
+    ],
+)
+def test_power_user_fields_are_advanced(
+    label: str, schema_factory, fields: list[str]
+) -> None:
+    markers = _markers(schema_factory())
+    for field in fields:
+        assert markers[field].visibility is cv.Visibility.ADVANCED, f"{label}.{field}"
+
+
+def test_interlock_wait_time_keeps_its_default() -> None:
+    """Marking the field advanced must not drop its default."""
+    markers = _markers(_gpio_switch_schema())
+    assert markers["interlock_wait_time"].default() == "0ms"

tests/component_tests/esp8266_pwm/config/frequency.yaml

@@ -0,0 +1,19 @@
+---
+esphome:
+  name: test
+
+esp8266:
+  board: d1_mini
+
+output:
+  - platform: esp8266_pwm
+    id: default_frequency
+    pin: GPIO4
+    frequency: 1kHz
+  - platform: esp8266_pwm
+    id: custom_frequency
+    pin: GPIO5
+    frequency: 2kHz
+  - platform: esp8266_pwm
+    id: schema_default_frequency
+    pin: GPIO12

tests/component_tests/esp8266_pwm/test_esp8266_pwm.py

@@ -0,0 +1,16 @@
+"""Tests for the esp8266_pwm output codegen."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+
+def test_default_frequency_is_not_emitted(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """The 1 kHz default already lives in the C++ initializer."""
+    main_cpp = generate_main(component_config_path("frequency.yaml"))
+
+    assert "default_frequency->set_frequency(" not in main_cpp
+    assert "schema_default_frequency->set_frequency(" not in main_cpp
+    assert "custom_frequency->set_frequency(2000.0f);" in main_cpp

tests/component_tests/light/config/transitions.yaml

@@ -0,0 +1,29 @@
+---
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+output:
+  - platform: ledc
+    id: out_a
+    pin: GPIO4
+  - platform: ledc
+    id: out_b
+    pin: GPIO5
+
+light:
+  - platform: monochromatic
+    id: plain_light
+    output: out_a
+    flash_transition_length: 0s
+  - platform: monochromatic
+    id: fancy_light
+    output: out_b
+    flash_transition_length: 500ms
+    effects:
+      - pulse:
+  - platform: monochromatic
+    id: bare_light
+    output: out_a

tests/component_tests/light/test_default_setters.py

@@ -0,0 +1,19 @@
+"""Tests that light codegen skips setters for default values."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+
+def test_default_flash_length_and_empty_effects_are_not_emitted(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """A 0 ms flash transition and an empty effect list match the C++ defaults."""
+    main_cpp = generate_main(component_config_path("transitions.yaml"))
+
+    assert "plain_light->set_flash_transition_length(" not in main_cpp
+    assert "plain_light->add_effects(" not in main_cpp
+    assert "bare_light->set_flash_transition_length(" not in main_cpp
+    assert "bare_light->add_effects(" not in main_cpp
+    assert "fancy_light->set_flash_transition_length(500);" in main_cpp
+    assert "fancy_light->add_effects({" in main_cpp

tests/component_tests/logger/test_logger.py

@@ -52,3 +52,35 @@ def test_logger_pre_setup_before_other_components(generate_main):
             f"Component allocation '{alloc.group()}' at position {alloc.start()} "
             f"appears before logger pre_setup() at position {logger_pre_setup.start()}"
         )
+
+
+def test_default_uart_selection_is_not_emitted(generate_main):
+    """UART0 is the C++ initializer on ESP8266, so the setter is skipped."""
+    main_cpp = generate_main("tests/component_tests/logger/test_logger.yaml")
+
+    assert "set_uart_selection(" not in main_cpp
+
+
+def test_custom_uart_selection_is_emitted(generate_main):
+    """A non default UART still reaches the setter before pre_setup()."""
+    main_cpp = generate_main("tests/component_tests/logger/test_logger_uart1.yaml")
+
+    assert "set_uart_selection(logger::UART_SELECTION_UART1);" in main_cpp
+
+
+def test_libretiny_default_uart_selection_is_not_emitted(generate_main):
+    """DEFAULT is the C++ initializer on LibreTiny, so the setter is skipped."""
+    main_cpp = generate_main(
+        "tests/component_tests/logger/test_logger_libretiny_default.yaml"
+    )
+
+    assert "set_uart_selection(" not in main_cpp
+
+
+def test_libretiny_uart0_is_emitted(generate_main):
+    """UART0 is not the LibreTiny initializer, so it must still be set."""
+    main_cpp = generate_main(
+        "tests/component_tests/logger/test_logger_libretiny_uart0.yaml"
+    )
+
+    assert "set_uart_selection(logger::UART_SELECTION_UART0);" in main_cpp

tests/component_tests/logger/test_logger_libretiny_default.yaml

@@ -0,0 +1,8 @@
+---
+esphome:
+  name: test
+
+rtl87xx:
+  board: generic-rtl8710bn-2mb-788k
+
+logger:

tests/component_tests/logger/test_logger_libretiny_uart0.yaml

@@ -0,0 +1,9 @@
+---
+esphome:
+  name: test
+
+rtl87xx:
+  board: generic-rtl8710bn-2mb-788k
+
+logger:
+  hardware_uart: UART0

tests/component_tests/logger/test_logger_uart1.yaml

@@ -0,0 +1,9 @@
+---
+esphome:
+  name: test
+
+esp8266:
+  board: d1_mini_lite
+
+logger:
+  hardware_uart: UART1

tests/component_tests/modbus/test_modbus.py

@@ -33,7 +33,6 @@ def test_server_schema_rejects_address_zero() -> None:
 
 
 def test_client_schema_still_accepts_address_zero() -> None:
-    # Not rejected for clients today, but not supported either: a client broadcast gets no reply and
-    # stalls the hub for the full send-wait.
+    # A client may address 0: writes are broadcast, and reads are allowed with allow_broadcast_read.
     schema = modbus.modbus_device_schema(0x01)
     assert schema({CONF_MODBUS_ID: "hub", CONF_ADDRESS: 0})[CONF_ADDRESS] == 0

tests/component_tests/modbus_client/test_modbus_client.py

@@ -7,7 +7,7 @@ guard is a safety property: these tests pin it to every handler slot.
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components import modbus_client
+from esphome.components import modbus, modbus_client
 from esphome.components.modbus_client import (
     CONF_ON_NO_RESPONSE,
     CONF_ON_NOT_SENT,
@@ -126,7 +126,7 @@ def test_on_no_response_retry_lambda_accepted() -> None:
 
 def test_continuous_on_write_pdu_rejected() -> None:
     """A literal write-code PDU with continuous: true is rejected at config time (reads only)."""
-    with pytest.raises(cv.Invalid, match="does not apply to a write PDU"):
+    with pytest.raises(cv.Invalid, match="does not apply to function code"):
         MODBUS_CLIENT_SEND_SCHEMA(
             {
                 CONF_ADDRESS: 0x01,
@@ -185,3 +185,145 @@ def test_multi_conf_no_default_is_set() -> None:
     """
     assert modbus_client.MULTI_CONF is True
     assert modbus_client.MULTI_CONF_NO_DEFAULT is True
+
+
+@pytest.mark.parametrize("key", [CONF_CONTINUOUS, modbus.CONF_ALLOW_BROADCAST_READ])
+def test_send_rejects_read_option_on_static_write_pdu(key: str) -> None:
+    # A read option set true on a static write PDU is refused at validation, naming the key.
+    config = {
+        CONF_ADDRESS: 1,
+        CONF_PDU: [0x06, 0x00, 0x10, 0x00, 0x01],
+        key: True,
+    }
+    with pytest.raises(
+        cv.Invalid, match=f"'{key}: true' does not apply to function code"
+    ):
+        MODBUS_CLIENT_SEND_SCHEMA(config)
+
+
+def test_send_accepts_allow_broadcast_read_on_read_pdu() -> None:
+    # allow_broadcast_read defaults to False and is accepted on a read PDU to address 0.
+    config = MODBUS_CLIENT_SEND_SCHEMA(
+        {CONF_ADDRESS: 0, CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02]}
+    )
+    assert config[modbus.CONF_ALLOW_BROADCAST_READ] is False
+    config = MODBUS_CLIENT_SEND_SCHEMA(
+        {
+            CONF_ADDRESS: 0,
+            CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02],
+            modbus.CONF_ALLOW_BROADCAST_READ: True,
+        }
+    )
+    assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
+
+
+def test_send_rejects_write_option_on_static_read_pdu() -> None:
+    # The write-side option is refused on a static read PDU, the mirror of the read-option check.
+    key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
+    with pytest.raises(
+        cv.Invalid, match=f"'{key}: true' does not apply to function code"
+    ):
+        MODBUS_CLIENT_SEND_SCHEMA(
+            {CONF_ADDRESS: 0, CONF_PDU: [0x03, 0x00, 0x10, 0x00, 0x02], key: True}
+        )
+
+
+def test_send_accepts_write_option_on_static_write_pdu() -> None:
+    config = MODBUS_CLIENT_SEND_SCHEMA(
+        {
+            CONF_ADDRESS: 0,
+            CONF_PDU: [0x06, 0x00, 0x10, 0x00, 0x01],
+            modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
+        }
+    )
+    assert config[modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE] is True
+
+
+def test_write_actions_offer_write_option_only() -> None:
+    # Every write action takes expect_broadcast_write_response and none of the read options.
+    from esphome.components.modbus_client import (
+        _WRITE_MULTIPLE_COILS_SCHEMA,
+        _WRITE_MULTIPLE_REGISTERS_SCHEMA,
+        _WRITE_SINGLE_COIL_SCHEMA,
+        _WRITE_SINGLE_REGISTER_SCHEMA,
+        CONF_START_ADDRESS,
+        CONF_VALUE,
+        CONF_VALUES,
+    )
+
+    write_key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
+    base = {CONF_ADDRESS: 0, CONF_START_ADDRESS: 0x10, write_key: True}
+    for schema, extra in (
+        (_WRITE_SINGLE_REGISTER_SCHEMA, {CONF_VALUE: 1}),
+        (_WRITE_SINGLE_COIL_SCHEMA, {CONF_VALUE: True}),
+        (_WRITE_MULTIPLE_REGISTERS_SCHEMA, {CONF_VALUES: [1, 2]}),
+        (_WRITE_MULTIPLE_COILS_SCHEMA, {CONF_VALUES: [True, False]}),
+    ):
+        config = schema({**base, **extra})
+        assert config[write_key] is True
+        assert modbus.CONF_ALLOW_BROADCAST_READ not in config
+        with pytest.raises(cv.Invalid):
+            schema({**base, **extra, modbus.CONF_ALLOW_BROADCAST_READ: True})
+
+
+def test_send_options_follow_the_hub_classification() -> None:
+    # A vendor code is broadcastable, so it takes the write-side flag and refuses the read-side one;
+    # 0x17 is a read for broadcast purposes, so the reverse holds.
+    write_key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
+    read_key = modbus.CONF_ALLOW_BROADCAST_READ
+    assert MODBUS_CLIENT_SEND_SCHEMA(
+        {CONF_ADDRESS: 0, CONF_PDU: [0x41, 0x01], write_key: True}
+    )[write_key]
+    with pytest.raises(cv.Invalid, match=f"'{read_key}: true' does not apply"):
+        MODBUS_CLIENT_SEND_SCHEMA(
+            {CONF_ADDRESS: 0, CONF_PDU: [0x41, 0x01], read_key: True}
+        )
+    pdu_0x17 = [0x17, 0x00, 0x10, 0x00, 0x01, 0x00, 0x20, 0x00, 0x01, 0x02, 0x00, 0x01]
+    assert MODBUS_CLIENT_SEND_SCHEMA(
+        {CONF_ADDRESS: 0, CONF_PDU: pdu_0x17, read_key: True}
+    )[read_key]
+    with pytest.raises(cv.Invalid, match=f"'{write_key}: true' does not apply"):
+        MODBUS_CLIENT_SEND_SCHEMA(
+            {CONF_ADDRESS: 0, CONF_PDU: pdu_0x17, write_key: True}
+        )
+
+
+def test_read_write_multiple_offers_allow_broadcast_read_only() -> None:
+    from esphome.components.modbus_client import (
+        _READ_WRITE_MULTIPLE_REGISTERS_SCHEMA,
+        CONF_READ_ADDRESS,
+        CONF_VALUES,
+        CONF_WRITE_ADDRESS,
+    )
+
+    config = _READ_WRITE_MULTIPLE_REGISTERS_SCHEMA(
+        {
+            CONF_ADDRESS: 0,
+            CONF_READ_ADDRESS: 0x10,
+            CONF_WRITE_ADDRESS: 0x20,
+            CONF_VALUES: [1],
+            modbus.CONF_ALLOW_BROADCAST_READ: True,
+        }
+    )
+    assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
+    assert CONF_CONTINUOUS not in config
+    assert modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE not in config
+
+
+@pytest.mark.parametrize(
+    "key",
+    [modbus.CONF_ALLOW_BROADCAST_READ, modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE],
+)
+def test_broadcast_options_rejected_on_literal_unicast_address(key: str) -> None:
+    # A broadcast-only option on a literal non-zero address would be silently dropped by the hub.
+    if key == modbus.CONF_ALLOW_BROADCAST_READ:
+        pdu = [0x03, 0x00, 0x10, 0x00, 0x01]
+    else:
+        pdu = [0x06, 0x00, 0x10, 0x00, 0x01]
+    with pytest.raises(cv.Invalid, match="only applies to the broadcast address"):
+        MODBUS_CLIENT_SEND_SCHEMA({CONF_ADDRESS: 1, CONF_PDU: pdu, key: True})
+    # A templated address is not decidable at validation and passes through.
+    config = MODBUS_CLIENT_SEND_SCHEMA(
+        {CONF_ADDRESS: Lambda("return 1;"), CONF_PDU: pdu, key: True}
+    )
+    assert config[key] is True

tests/component_tests/modbus_controller/test_broadcast_address.py

@@ -0,0 +1,79 @@
+"""A modbus_controller cannot poll the broadcast address (0) unless allow_broadcast_read says the
+device answers it."""
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components import modbus
+from esphome.components.modbus_controller import CONFIG_SCHEMA
+from esphome.const import CONF_ADDRESS
+from esphome.types import ConfigType
+
+
+def _controller(address: int, **extra: object) -> ConfigType:
+    return CONFIG_SCHEMA({modbus.CONF_MODBUS_ID: "bus", CONF_ADDRESS: address, **extra})
+
+
+def test_address_zero_rejected_by_default() -> None:
+    with pytest.raises(cv.Invalid, match="broadcast address"):
+        _controller(0)
+
+
+def test_address_zero_accepted_with_allow_broadcast_read() -> None:
+    config = _controller(0, **{modbus.CONF_ALLOW_BROADCAST_READ: True})
+    assert config[CONF_ADDRESS] == 0
+    assert config[modbus.CONF_ALLOW_BROADCAST_READ] is True
+
+
+def test_allow_broadcast_read_defaults_false() -> None:
+    assert _controller(1)[modbus.CONF_ALLOW_BROADCAST_READ] is False
+
+
+def test_writer_entity_takes_expect_broadcast_write_response() -> None:
+    # The write-side option lives on the writing platforms, not the controller.
+    from esphome.components.modbus_controller.const import CONF_MODBUS_CONTROLLER_ID
+    from esphome.components.modbus_controller.switch import (
+        CONFIG_SCHEMA as SWITCH_SCHEMA,
+    )
+    from esphome.const import CONF_NAME
+
+    key = modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE
+    base = {
+        CONF_MODBUS_CONTROLLER_ID: "ctl",
+        CONF_NAME: "Switch",
+        "register_type": "coil",
+        CONF_ADDRESS: 0x20,
+    }
+    assert SWITCH_SCHEMA(base)[key] is False
+    assert SWITCH_SCHEMA({**base, CONF_NAME: "Switch 2", key: True})[key] is True
+    with pytest.raises(cv.Invalid):
+        CONFIG_SCHEMA({modbus.CONF_MODBUS_ID: "bus", CONF_ADDRESS: 1, key: True})
+
+
+def test_allow_broadcast_read_requires_address_zero() -> None:
+    # The option only means something at address 0; elsewhere it would be silently inert.
+    with pytest.raises(cv.Invalid, match="only applies to the broadcast address"):
+        _controller(5, **{modbus.CONF_ALLOW_BROADCAST_READ: True})
+
+
+def test_add_command_options_skips_defaults() -> None:
+    # The setter is only emitted when an option differs from its C++ default.
+    import esphome.codegen as cg
+    from esphome.const import CONF_CONTINUOUS
+
+    var = cg.MockObj("ctl")
+    emitted: list = []
+    original = cg.add
+    cg.add = emitted.append
+    try:
+        modbus.add_command_options(
+            var, "set_read_options", {CONF_CONTINUOUS: False}, direction="read"
+        )
+        assert emitted == []
+        modbus.add_command_options(
+            var, "set_read_options", {CONF_CONTINUOUS: True}, direction="read"
+        )
+        assert len(emitted) == 1
+        assert "set_read_options" in str(emitted[0])
+    finally:
+        cg.add = original

tests/component_tests/modbus_controller/test_custom_pdu.py

@@ -9,6 +9,7 @@ test cannot: a write-coded custom_pdu polled continuously is rejected there.
 import pytest
 from voluptuous import Invalid, MultipleInvalid
 
+from esphome.components import modbus
 from esphome.components.modbus_controller import (
     ModbusItemBaseSchema,
     validate_custom_pdu_item,
@@ -55,14 +56,21 @@ def test_custom_pdu_rejects_non_byte_values() -> None:
         ModbusItemBaseSchema({CONF_CUSTOM_PDU: [0x0103, 0x002A]})
 
 
-def _controller_full_config(*, continuous: bool) -> Config:
+def _controller_full_config(
+    *, continuous: bool, allow_broadcast_read: bool = False
+) -> 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
+    final-validate to resolve the controller (and its option flags) 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}
+        {
+            CONF_ID: ctl_id,
+            CONF_ADDRESS: 0 if allow_broadcast_read else 1,
+            CONF_CONTINUOUS: continuous,
+            modbus.CONF_ALLOW_BROADCAST_READ: allow_broadcast_read,
+        }
     ]
     config.declare_ids.append((ctl_id, ["modbus_controller", 0, CONF_ID]))
     return config
@@ -98,3 +106,64 @@ def test_continuous_read_custom_pdu_allowed(reset_full_config) -> None:
             CONF_CUSTOM_PDU: [0x03, 0x00, 0x2A, 0x00, 0x01],
         }
     )
+
+
+def test_broadcastable_custom_pdu_rejected_under_broadcast_controller(
+    reset_full_config,
+) -> None:
+    """A vendor-coded custom_pdu under an allow_broadcast_read controller would be a real broadcast,
+    never answered, so it is rejected at final validate."""
+    fv.full_config.set(
+        _controller_full_config(continuous=False, allow_broadcast_read=True)
+    )
+    with pytest.raises(Invalid, match="is a real broadcast at address 0"):
+        validate_custom_pdu_item(
+            {
+                CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
+                CONF_CUSTOM_PDU: [0x41, 0x00, 0x03],
+            }
+        )
+
+
+def test_read_custom_pdu_allowed_under_broadcast_controller(reset_full_config) -> None:
+    """A read-coded custom_pdu (0x03) is answered under allow_broadcast_read, so it is fine."""
+    fv.full_config.set(
+        _controller_full_config(continuous=False, allow_broadcast_read=True)
+    )
+    validate_custom_pdu_item(
+        {
+            CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
+            CONF_CUSTOM_PDU: [0x03, 0x00, 0x2A, 0x00, 0x01],
+        }
+    )
+
+
+def test_write_option_rejected_under_unicast_controller(reset_full_config) -> None:
+    """expect_broadcast_write_response on a writer entity whose controller is not at address 0 is
+    rejected at final validate, where the controller's address is known."""
+    from esphome.components.modbus_controller import validate_writer_item
+
+    fv.full_config.set(_controller_full_config(continuous=False))
+    with pytest.raises(
+        Invalid, match="only applies when the 'ctl' modbus_controller is at address 0"
+    ):
+        validate_writer_item(
+            {
+                CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
+                modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
+            }
+        )
+
+
+def test_write_option_allowed_under_broadcast_controller(reset_full_config) -> None:
+    from esphome.components.modbus_controller import validate_writer_item
+
+    fv.full_config.set(
+        _controller_full_config(continuous=False, allow_broadcast_read=True)
+    )
+    validate_writer_item(
+        {
+            CONF_MODBUS_CONTROLLER_ID: ID("ctl"),
+            modbus.CONF_EXPECT_BROADCAST_WRITE_RESPONSE: True,
+        }
+    )

tests/component_tests/number/config/mode.yaml

@@ -0,0 +1,28 @@
+---
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+number:
+  - platform: template
+    id: auto_number
+    min_value: 0
+    max_value: 10
+    step: 1
+    optimistic: true
+  - platform: template
+    id: box_number
+    min_value: 0
+    max_value: 10
+    step: 1
+    mode: box
+    optimistic: true
+  - platform: template
+    id: explicit_auto_number
+    min_value: 0
+    max_value: 10
+    step: 1
+    mode: auto
+    optimistic: true

tests/component_tests/number/test_number.py

@@ -0,0 +1,16 @@
+"""Tests for the number component codegen."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+
+def test_default_mode_is_not_emitted(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Mode auto is the C++ initializer, so only a non default mode is set."""
+    main_cpp = generate_main(component_config_path("mode.yaml"))
+
+    assert "auto_number->traits.set_mode(" not in main_cpp
+    assert "explicit_auto_number->traits.set_mode(" not in main_cpp
+    assert "box_number->traits.set_mode(number::NUMBER_MODE_BOX);" in main_cpp

tests/component_tests/output/config/ac_dimmer_min_power_zero.yaml

@@ -0,0 +1,13 @@
+---
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+output:
+  - platform: ac_dimmer
+    id: dimmer
+    gate_pin: GPIO4
+    zero_cross_pin: GPIO5
+    min_power: 0%

tests/component_tests/output/config/power_limits.yaml

@@ -0,0 +1,18 @@
+---
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+output:
+  - platform: ledc
+    id: default_power
+    pin: GPIO4
+    max_power: 100%
+    min_power: 0%
+  - platform: ledc
+    id: custom_power
+    pin: GPIO5
+    max_power: 90%
+    min_power: 1%

tests/component_tests/output/test_output.py

@@ -0,0 +1,31 @@
+"""Tests for the output platform codegen."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+from esphome.core import CORE
+
+
+def test_default_power_limits_are_not_emitted(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """max_power 100% and min_power 0% already live in the C++ initializers."""
+    main_cpp = generate_main(component_config_path("power_limits.yaml"))
+
+    assert "default_power->set_max_power(" not in main_cpp
+    assert "default_power->set_min_power(" not in main_cpp
+    assert "custom_power->set_max_power(0.9f);" in main_cpp
+    assert "custom_power->set_min_power(0.01f);" in main_cpp
+    assert "USE_OUTPUT_FLOAT_POWER_SCALING" in {d.name for d in CORE.defines}
+
+
+def test_default_min_power_keeps_scaling_fields(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """ac_dimmer reads min_power_ directly, so the define must stay on for min_power 0%."""
+    main_cpp = generate_main(component_config_path("ac_dimmer_min_power_zero.yaml"))
+
+    assert "dimmer->set_min_power(" not in main_cpp
+    assert "USE_OUTPUT_FLOAT_POWER_SCALING" in {d.name for d in CORE.defines}

tests/component_tests/web_server/config/bare.yaml

@@ -0,0 +1,12 @@
+---
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+wifi:
+  ssid: test
+  password: testtest
+
+web_server:

tests/component_tests/web_server/config/custom.yaml

@@ -0,0 +1,15 @@
+---
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+wifi:
+  ssid: test
+  password: testtest
+
+web_server:
+  port: 8080
+  log: false
+  include_internal: true

tests/component_tests/web_server/config/defaults.yaml

@@ -0,0 +1,15 @@
+---
+esphome:
+  name: test
+
+esp32:
+  board: esp32dev
+
+wifi:
+  ssid: test
+  password: testtest
+
+web_server:
+  port: 80
+  log: true
+  include_internal: false

tests/component_tests/web_server/test_default_setters.py

@@ -0,0 +1,35 @@
+"""Tests that web_server only emits setters for non default values."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+
+@pytest.mark.parametrize("config_file", ["bare.yaml", "defaults.yaml"])
+def test_default_values_are_not_emitted(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    config_file: str,
+) -> None:
+    """Port 80, log on and include_internal off already live in the C++ initializers.
+
+    Both the schema defaults and the same values written explicitly take the skip path.
+    """
+    main_cpp = generate_main(component_config_path(config_file))
+
+    assert "set_port(" not in main_cpp
+    assert "set_expose_log(" not in main_cpp
+    assert "set_include_internal(" not in main_cpp
+
+
+def test_custom_values_are_emitted(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Non default values still reach the C++ setters."""
+    main_cpp = generate_main(component_config_path("custom.yaml"))
+
+    assert "set_port(8080);" in main_cpp
+    assert "set_expose_log(false);" in main_cpp
+    assert "set_include_internal(true);" in main_cpp

tests/component_tests/wifi/config/bare.yaml

@@ -0,0 +1,12 @@
+---
+esphome:
+  name: test
+
+esp8266:
+  board: d1_mini
+
+wifi:
+  ssid: test
+  password: testtest
+  ap:
+    ssid: fallback

tests/component_tests/wifi/config/custom.yaml

@@ -0,0 +1,18 @@
+---
+esphome:
+  name: test
+
+esp8266:
+  board: d1_mini
+
+wifi:
+  networks:
+    - ssid: test
+      password: testtest
+      priority: 5
+  ap:
+    ssid: fallback
+    ap_timeout: 2min
+  reboot_timeout: 0s
+  power_save_mode: light
+  min_auth_mode: wpa

tests/component_tests/wifi/config/defaults.yaml

@@ -0,0 +1,18 @@
+---
+esphome:
+  name: test
+
+esp8266:
+  board: d1_mini
+
+wifi:
+  networks:
+    - ssid: test
+      password: testtest
+      priority: 0
+  ap:
+    ssid: fallback
+    ap_timeout: 90s
+  reboot_timeout: 15min
+  power_save_mode: none
+  min_auth_mode: wpa2

tests/component_tests/wifi/test_default_setters.py

@@ -0,0 +1,39 @@
+"""Tests that wifi codegen skips setters for default values."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+
+@pytest.mark.parametrize("config_file", ["bare.yaml", "defaults.yaml"])
+def test_default_values_are_not_emitted(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    config_file: str,
+) -> None:
+    """Priority 0, 90 s AP timeout, 15 min reboot, power save none, WPA2 are C++ defaults.
+
+    Both the schema defaults and the same values written explicitly take the skip path.
+    """
+    main_cpp = generate_main(component_config_path(config_file))
+
+    assert "set_priority(" not in main_cpp
+    assert "set_ap_timeout(" not in main_cpp
+    assert "set_reboot_timeout(" not in main_cpp
+    assert "set_power_save_mode(" not in main_cpp
+    assert "set_min_auth_mode(" not in main_cpp
+
+
+def test_custom_values_are_emitted(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Non default values still reach the C++ setters."""
+    main_cpp = generate_main(component_config_path("custom.yaml"))
+
+    assert "set_priority(5);" in main_cpp
+    assert "set_ap_timeout(120000);" in main_cpp
+    assert "set_reboot_timeout(0);" in main_cpp
+    assert "set_power_save_mode(wifi::WIFI_POWER_SAVE_LIGHT);" in main_cpp
+    assert "set_min_auth_mode(wifi::WIFI_MIN_AUTH_MODE_WPA);" in main_cpp

tests/component_tests/wifi/test_power_save_off.py

@@ -0,0 +1,46 @@
+"""Tests for wifi.force_power_save_off(), the hook platforms use to keep the
+station out of power save."""
+
+from collections.abc import Callable
+from pathlib import Path
+
+import pytest
+
+from esphome.components import wifi
+from esphome.core import CORE, EsphomeError
+
+
+def test_reasons_accumulate_without_duplicates() -> None:
+    """Every caller's reason is kept once; a repeated reason is not duplicated."""
+    wifi.force_power_save_off("first")
+    wifi.force_power_save_off("first")
+    wifi.force_power_save_off("second")
+
+    assert CORE.data[wifi.POWER_SAVE_OFF_REASONS_KEY] == ["first", "second"]
+
+
+def test_forced_off_skips_the_setter_and_warns(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """With a reason recorded, power_save_mode is reported and not applied."""
+    wifi.force_power_save_off("the platform cannot sleep")
+
+    main_cpp = generate_main(component_config_path("custom.yaml"))
+
+    assert "set_power_save_mode(" not in main_cpp
+    assert (
+        "power_save_mode LIGHT is not applied: the platform cannot sleep" in caplog.text
+    )
+
+
+def test_call_after_wifi_codegen_raises(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Once wifi has generated its code the hook cannot take effect any more."""
+    generate_main(component_config_path("custom.yaml"))
+
+    with pytest.raises(EsphomeError, match="before wifi generates its code"):
+        wifi.force_power_save_off("too late")

tests/components/ac_dimmer/common.yaml

@@ -4,3 +4,4 @@ output:
     gate_pin: ${gate_pin}
     zero_cross_pin: ${zero_cross_pin}
     zero_cross_interrupt_type: ANY
+    min_power: 0%

tests/components/bk72xx_ble/validate-power-save.bk72xx-ard.yaml

@@ -0,0 +1,9 @@
+# A wifi power_save_mode other than NONE is forced off with a warning while
+# bk72xx_ble is configured (esphome#18592); this config must still validate.
+packages:
+  bk72xx_ble: !include common.yaml
+
+wifi:
+  ssid: MySSID
+  password: password1
+  power_save_mode: high

tests/components/modbus/modbus_client_hub_test.cpp

@@ -792,6 +792,261 @@ TEST(ModbusClientHubBroadcast, RefusesReadBroadcast) {
   EXPECT_EQ(device.sent_count_, 0);  // never transmitted
 }
 
+// allow_broadcast_read lifts the refusal for a device that answers address 0: the read is queued, sent,
+// and waits for a reply like a unicast read, so a reply from address 0 completes it with on_response.
+TEST(ModbusClientHubBroadcast, AllowBroadcastReadWaitsAndAcceptsReplyFromZero) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
+
+  const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};  // read holding registers 0x0010, count 2
+  ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
+  EXPECT_TRUE(hub.queued(0).options.allow_broadcast_read);
+  EXPECT_FALSE(hub.queued(0).fire_and_forget());
+
+  hub.send_next_for_test();
+  EXPECT_EQ(device.sent_count_, 1);
+  EXPECT_TRUE(hub.waiting());  // not fire-and-forget: the reply is expected
+  EXPECT_EQ(hub.entries(), 1u);
+
+  const uint8_t reply[] = {0x03, 0x04, 0x00, 0x01, 0x00, 0x02};
+  hub.receive_frame_for_test(BROADCAST_ADDRESS, reply);
+  EXPECT_EQ(device.response_count_, 1);
+  EXPECT_EQ(device.last_response_size_, sizeof(reply));
+  EXPECT_FALSE(hub.waiting());
+  EXPECT_EQ(hub.entries(), 0u);
+}
+
+// The address-0 read waits like a unicast one, so the reply must come from address 0 too: a reply from
+// another unit id is an unexpected frame and interrupts the transaction as it would for any address.
+TEST(ModbusClientHubBroadcast, AllowBroadcastReadRejectsReplyFromOtherAddress) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
+
+  const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
+  ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
+  hub.send_next_for_test();
+  ASSERT_TRUE(hub.waiting());
+
+  const uint8_t reply[] = {0x03, 0x04, 0x00, 0x01, 0x00, 0x02};
+  hub.receive_frame_for_test(0x07, reply);
+  EXPECT_EQ(device.response_count_, 0);
+  EXPECT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED);
+}
+
+// An address-scoped clear must not turn a live address-0 entry back into a fire-and-forget broadcast: a
+// retry granted after the clear is re-sent with the flag intact, so it still waits and gets its terminal.
+TEST(ModbusClientHubBroadcast, AllowBroadcastReadSurvivesClearBeforeRetry) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  RetryingDevice device(&hub, BROADCAST_ADDRESS, true);
+
+  const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
+  ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
+  hub.send_next_for_test();
+  ASSERT_TRUE(hub.waiting());
+
+  hub.clear_tx_queue_for_address(BROADCAST_ADDRESS);
+  EXPECT_EQ(hub.waiting_command().state, FrameState::WAITING_RETIRED);
+  EXPECT_TRUE(hub.waiting_command().options.allow_broadcast_read);
+
+  hub.timeout_waiting();  // retry granted: the entry is READY again
+  ASSERT_EQ(hub.queued_frames(), 1u);
+  EXPECT_FALSE(hub.queued(0).fire_and_forget());
+
+  hub.send_next_for_test();
+  EXPECT_TRUE(hub.waiting());  // the retry still waits for its reply
+  EXPECT_EQ(hub.entries(), 1u);
+}
+
+// The function code check is unchanged by the relaxed address match: a mismatched reply still interrupts.
+TEST(ModbusClientHubBroadcast, AllowBroadcastReadStillRejectsWrongFunctionCode) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
+
+  const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
+  ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
+  hub.send_next_for_test();
+  ASSERT_TRUE(hub.waiting());
+
+  const uint8_t wrong_reply[] = {0x04, 0x04, 0x00, 0x01, 0x00, 0x02};
+  hub.receive_frame_for_test(BROADCAST_ADDRESS, wrong_reply);  // right address, wrong function code
+  EXPECT_EQ(device.response_count_, 0);
+  EXPECT_EQ(hub.waiting_command().state, FrameState::INTERRUPTED);
+}
+
+// A silent device leaves the read to the normal send-wait timeout, so on_no_response is delivered.
+TEST(ModbusClientHubBroadcast, AllowBroadcastReadTimesOutLikeUnicast) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
+
+  const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
+  ASSERT_TRUE(device.queue_pdu(read, {.allow_broadcast_read = true}));
+  hub.send_next_for_test();
+  ASSERT_TRUE(hub.waiting());
+
+  hub.timeout_waiting();
+  EXPECT_EQ(device.no_response_count_, 1);
+  EXPECT_EQ(device.response_count_, 0);
+  EXPECT_FALSE(hub.waiting());
+  EXPECT_EQ(hub.entries(), 0u);
+}
+
+// allow_broadcast_read is stripped from a broadcastable code (a write or custom code to address 0 is a real broadcast,
+// still fire-and-forget) and from a unicast frame (nothing to allow).
+TEST(ModbusClientHubBroadcast, AllowBroadcastReadIgnoredForWritesAndUnicast) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice broadcast_device(&hub, BROADCAST_ADDRESS);
+  BroadcastProbeDevice unicast_device(&hub, 0x01);
+
+  const uint8_t write[] = {0x06, 0x00, 0x10, 0x00, 0x01};
+  ASSERT_TRUE(broadcast_device.queue_pdu(write, {.allow_broadcast_read = true}));
+  EXPECT_FALSE(hub.queued(0).options.allow_broadcast_read);
+  EXPECT_TRUE(hub.queued(0).fire_and_forget());
+  hub.send_next_for_test();
+  EXPECT_EQ(broadcast_device.sent_count_, 1);
+  EXPECT_FALSE(hub.waiting());
+  EXPECT_EQ(hub.entries(), 0u);
+
+  const uint8_t custom[] = {0x41, 0x01, 0x02};
+  ASSERT_TRUE(broadcast_device.queue_pdu(custom, {.allow_broadcast_read = true}));
+  EXPECT_FALSE(hub.queued(0).options.allow_broadcast_read);
+  EXPECT_TRUE(hub.queued(0).fire_and_forget());
+  hub.send_next_for_test();
+  EXPECT_FALSE(hub.waiting());
+  EXPECT_EQ(hub.entries(), 0u);
+
+  const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
+  ASSERT_TRUE(unicast_device.queue_pdu(read, {.allow_broadcast_read = true}));
+  EXPECT_FALSE(hub.queued(0).options.allow_broadcast_read);
+  EXPECT_FALSE(hub.queued(0).fire_and_forget());
+}
+
+// expect_broadcast_write_response is the write-side twin: a write to address 0 waits for its reply instead
+// of retiring at transmission, and the reply (from address 0) completes it.
+TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseWaitsAndAcceptsReply) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
+
+  const uint8_t write[] = {0x06, 0x00, 0x10, 0x00, 0x01};
+  ASSERT_TRUE(device.write_single_register(0x0010, 0x0001, {.expect_broadcast_write_response = true}));
+  EXPECT_TRUE(hub.queued(0).options.expect_broadcast_write_response);
+  EXPECT_FALSE(hub.queued(0).fire_and_forget());
+
+  hub.send_next_for_test();
+  EXPECT_EQ(device.sent_count_, 1);
+  EXPECT_TRUE(hub.waiting());
+  EXPECT_EQ(hub.entries(), 1u);
+
+  hub.receive_frame_for_test(BROADCAST_ADDRESS, write);  // the echo, as address 0
+  EXPECT_EQ(device.response_count_, 1);
+  EXPECT_EQ(device.last_response_size_, sizeof(write));
+  EXPECT_FALSE(hub.waiting());
+  EXPECT_EQ(hub.entries(), 0u);
+}
+
+// Two requests for the same address-0 write may disagree on expect_broadcast_write_response (a
+// broadcastable frame is accepted either way), but a write duplicate is refused at its cap of one in
+// flight rather than absorbed, so the queued entry's delivery mode is never changed under it.
+TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseDuplicateRefusedNotMerged) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
+
+  ASSERT_TRUE(device.write_single_register(0x0010, 0x0001));  // fire-and-forget as queued
+  EXPECT_TRUE(hub.queued(0).fire_and_forget());
+  EXPECT_FALSE(device.write_single_register(0x0010, 0x0001, {.expect_broadcast_write_response = true}));
+  EXPECT_EQ(hub.entries(), 1u);
+  EXPECT_TRUE(hub.queued(0).fire_and_forget());  // the refused request left the entry untouched
+
+  hub.send_next_for_test();
+  EXPECT_FALSE(hub.waiting());
+  EXPECT_EQ(hub.entries(), 0u);
+}
+
+// A custom-code poll at address 0 is a fire-and-forget broadcast that a one-shot duplicate downgrades and
+// is absorbed into; if that duplicate wants the reply, the entry waits for it instead of retiring at the
+// send, so the absorbed request still gets its terminal callback.
+TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseMergesIntoDowngradedPoll) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
+
+  const uint8_t custom[] = {0x41, 0x01, 0x02};
+  ASSERT_TRUE(device.queue_pdu(custom, {.continuous = true}));
+  EXPECT_TRUE(hub.queued(0).fire_and_forget());
+  ASSERT_TRUE(device.queue_pdu(custom, {.expect_broadcast_write_response = true}));  // downgrades, absorbed
+  EXPECT_EQ(hub.entries(), 1u);
+  EXPECT_FALSE(hub.queued(0).options.continuous);
+  EXPECT_FALSE(hub.queued(0).fire_and_forget());
+
+  hub.send_next_for_test();
+  EXPECT_TRUE(hub.waiting());
+  hub.receive_frame_for_test(BROADCAST_ADDRESS, custom);
+  EXPECT_EQ(device.response_count_, 1);
+}
+
+// A silent device leaves an expected write response to the normal send-wait timeout.
+TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseTimesOutLikeUnicast) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice device(&hub, BROADCAST_ADDRESS);
+
+  ASSERT_TRUE(device.write_single_coil(0x0010, true, {.expect_broadcast_write_response = true}));
+  hub.send_next_for_test();
+  ASSERT_TRUE(hub.waiting());
+
+  hub.timeout_waiting();
+  EXPECT_EQ(device.no_response_count_, 1);
+  EXPECT_EQ(device.response_count_, 0);
+  EXPECT_FALSE(hub.waiting());
+  EXPECT_EQ(hub.entries(), 0u);
+}
+
+// expect_broadcast_write_response is stripped from a read (allow_broadcast_read is the read-side flag, so
+// the broadcast guard still refuses it) and from a unicast frame (nothing to expect).
+TEST(ModbusClientHubBroadcast, ExpectBroadcastWriteResponseIgnoredForReadsAndUnicast) {
+  NullUART uart;
+  NoResponseProbeHub hub;
+  hub.set_uart_parent(&uart);
+  hub.setup();
+  BroadcastProbeDevice broadcast_device(&hub, BROADCAST_ADDRESS);
+  BroadcastProbeDevice unicast_device(&hub, 0x01);
+
+  const uint8_t read[] = {0x03, 0x00, 0x10, 0x00, 0x02};
+  EXPECT_FALSE(broadcast_device.queue_pdu(read, {.expect_broadcast_write_response = true}));
+  EXPECT_EQ(hub.entries(), 0u);
+
+  ASSERT_TRUE(unicast_device.write_single_register(0x0010, 0x0001, {.expect_broadcast_write_response = true}));
+  EXPECT_FALSE(hub.queued(0).options.expect_broadcast_write_response);
+  EXPECT_FALSE(hub.queued(0).fire_and_forget());
+}
+
 // The counterpart to RefusesReadBroadcast: a custom (user-defined) function code carries no reply the
 // hub knows how to expect, so a broadcast of one is accepted and completes fire-and-forget like a write.
 TEST(ModbusClientHubBroadcast, AcceptsCustomBroadcast) {

tests/components/modbus_client/common.yaml

@@ -79,7 +79,8 @@ button:
     name: "Typed Actions"
     on_press:
       - modbus_client.write_single_register:
-          address: 0x01
+          address: !lambda "return 1;"
+          expect_broadcast_write_response: true
           start_address: 0x0102
           value: !lambda "return 42;"
           on_response:
@@ -93,6 +94,7 @@ button:
           start_address: 0x10
           count: 2
           continuous: true
+          allow_broadcast_read: !lambda "return false;"
           on_response:
             then:
               - lambda: 'ESP_LOGI("modbus_client.test", "first=%u n=%u", values[0], (unsigned) values.size());'

tests/components/modbus_client/validate-broadcast.esp32-idf.yaml

@@ -0,0 +1,36 @@
+# Config-only: actions that address the broadcast address (0) and wait for a reply, for a device that
+# answers it. Never compiled, so the extra action objects do not inflate the memory-impact baseline.
+packages:
+  modbus: !include ../../test_build_components/common/modbus/esp32-idf.yaml
+
+button:
+  - platform: template
+    name: Broadcast probe
+    on_press:
+      - modbus_client.read_holding_registers:
+          address: 0
+          allow_broadcast_read: true
+          start_address: 0x10
+          count: 1
+          on_response:
+            then:
+              - lambda: 'ESP_LOGI("modbus_client.test", "broadcast read first=%u", values[0]);'
+      - modbus_client.write_single_register:
+          address: 0
+          expect_broadcast_write_response: true
+          start_address: 0x0102
+          value: 42
+          on_response:
+            then:
+              - logger.log: "broadcast write acked"
+      - modbus_client.read_write_multiple_registers:
+          address: 0
+          allow_broadcast_read: true
+          read_address: 0x10
+          read_count: 1
+          write_address: 0x20
+          values: [1]
+      - modbus_client.send:
+          address: 0
+          expect_broadcast_write_response: true
+          pdu: [0x41, 0x01]

tests/components/modbus_controller/common.yaml

@@ -6,7 +6,6 @@ modbus_controller:
     on_online:
       then:
         logger.log: "Module Online"
-
 binary_sensor:
   - platform: modbus_controller
     modbus_controller_id: modbus_controller1

tests/components/modbus_controller/validate-broadcast.esp32-idf.yaml

@@ -0,0 +1,29 @@
+# Config-only: a controller polling the broadcast address (0), for a device that answers it, with a
+# writer entity expecting the reply to its broadcast writes. Never compiled, so the extra entities do
+# not inflate the memory-impact baseline.
+packages:
+  modbus: !include ../../test_build_components/common/modbus/esp32-idf.yaml
+
+modbus_controller:
+  - id: modbus_controller_broadcast
+    address: 0
+    allow_broadcast_read: true
+    modbus_id: modbus_bus
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_broadcast
+    id: modbus_broadcast_sensor
+    name: Broadcast Read Sensor
+    register_type: holding
+    address: 0x0010
+    value_type: U_WORD
+
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_broadcast
+    id: modbus_broadcast_switch
+    name: Broadcast Write Switch
+    register_type: coil
+    address: 0x20
+    expect_broadcast_write_response: true

tests/integration/fixtures/water_heater_template_unknown_temperature.yaml

@@ -0,0 +1,16 @@
+esphome:
+  name: wh-template-unknown-test
+host:
+api:
+logger:
+
+water_heater:
+  - platform: template
+    id: unknown_boiler
+    name: Unknown Boiler
+    # Both temperatures stay unknown, as they do before an upstream component reports a value.
+    current_temperature: !lambda "return NAN;"
+    target_temperature: !lambda "return NAN;"
+    supported_modes:
+      - "off"
+      - eco

tests/integration/test_water_heater_template.py

@@ -155,3 +155,36 @@ async def test_water_heater_template(
         client.water_heater_command(test_water_heater.key, mode=WaterHeaterMode.ECO)
         eco_state = await wait_for_state()
         assert eco_state.mode == WaterHeaterMode.ECO
+
+
+@pytest.mark.asyncio
+async def test_water_heater_template_unknown_temperature(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test a template water heater whose temperature lambdas stay unknown.
+
+    NAN never compares equal to itself, so a lambda that keeps returning NAN must not be
+    mistaken for a changed value and republish the state on every loop iteration.
+    """
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        state_count = 0
+
+        def on_state(state: aioesphomeapi.EntityState) -> None:
+            nonlocal state_count
+            if isinstance(state, WaterHeaterState):
+                state_count += 1
+
+        entities, _ = await client.list_entities_services()
+        water_heater_infos = [e for e in entities if isinstance(e, WaterHeaterInfo)]
+        assert len(water_heater_infos) == 1
+
+        client.subscribe_states(on_state)
+
+        # Let the device run for a while; only the single initial state may arrive.
+        await asyncio.sleep(1.0)
+        assert state_count <= 1, (
+            f"Expected at most 1 state publish, got {state_count} - "
+            "an unknown (NAN) temperature is republishing every loop"
+        )

tests/unit_tests/components/file/test_image.py

@@ -5,8 +5,13 @@ from __future__ import annotations
 from pathlib import Path
 from unittest.mock import patch
 
+import pytest
+
+from esphome import yaml_util
 from esphome.components.file import image as file_image
-from esphome.external_files import RemoteFile
+from esphome.const import CONF_PATH
+from esphome.core import CORE
+from esphome.external_files import RemoteFile, url_cache_key
 from esphome.loader import get_component, get_platform
 
 
@@ -55,6 +60,42 @@ def test_prefetch_files_yields_remote_refs(setup_core: Path) -> None:
     assert files[1].url == "https://example.com/img.png"
 
 
+def test_validated_file_values_hash_alike_across_data_dirs(
+    setup_core: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A CLI and an add-on data dir dump validated image files identically."""
+    url = "https://example.com/img.png"
+    (setup_core / "img.png").touch()
+    dumps: list[str] = []
+    for data_dir in (
+        setup_core / ".esphome",
+        setup_core.parent / f"{setup_core.name}-data",
+    ):
+        monkeypatch.setenv("ESPHOME_DATA_DIR", str(data_dir))
+        with patch("esphome.components.file.image.external_files.download_content"):
+            config = {
+                "remote": file_image.validate_file_shorthand(url),
+                "mdi": file_image.validate_file_shorthand("mdi:home"),
+                "local": file_image.validate_file_shorthand("img.png"),
+                "local_schema": file_image.LOCAL_SCHEMA({CONF_PATH: "img.png"}),
+            }
+        dumps.append(
+            yaml_util.dump(
+                config,
+                sort_keys=True,
+                relative_to=CORE.config_dir,
+                data_dir=CORE.data_dir,
+            )
+        )
+    assert dumps[0] == dumps[1]
+    assert dumps[0].splitlines() == [
+        "local: img.png",
+        "local_schema: img.png",
+        "mdi: .esphome/image/mdi/home.svg",
+        f"remote: .esphome/image/{url_cache_key(url)}",
+    ]
+
+
 def test_extractor_matches_validator_path(setup_core: Path) -> None:
     """The path the validator downloads to equals the extractor's path."""
     with patch(