diff --git a/esphome/components/modbus/helpers.py b/esphome/components/modbus/helpers.py
index e7eaacee0ca..ec95b820455 100644
--- a/esphome/components/modbus/helpers.py
+++ b/esphome/components/modbus/helpers.py
@@ -3,6 +3,9 @@ import esphome.codegen as cg
 modbus_ns = cg.esphome_ns.namespace("modbus")
 modbus_helpers_ns = modbus_ns.namespace("helpers")
 
+RegisterValues = modbus_ns.class_("RegisterValues")
+PduBuffer = modbus_helpers_ns.class_("PduBuffer")
+
 FunctionCode_ns = modbus_ns.namespace("FunctionCode")
 FunctionCode = FunctionCode_ns.enum("FunctionCode")
 
diff --git a/esphome/components/modbus_controller/__init__.py b/esphome/components/modbus_controller/__init__.py
index 924a260d37a..e87eccb32ca 100644
--- a/esphome/components/modbus_controller/__init__.py
+++ b/esphome/components/modbus_controller/__init__.py
@@ -191,7 +191,7 @@ ModbusItemBaseSchema = cv.Schema(
 )
 
 
-def validate_modbus_register(config):
+def validate_modbus_register(config: ConfigType) -> ConfigType:
     # custom_command is the deprecated alias for custom_pdu (migrated later in final validate); treat
     # either as "a custom frame is configured" so the address/register_type rules match.
     has_custom = CONF_CUSTOM_PDU in config or CONF_CUSTOM_COMMAND in config
@@ -278,7 +278,7 @@ def _final_validate(config: ConfigType) -> None:
 FINAL_VALIDATE_SCHEMA = _final_validate
 
 
-def modbus_calc_properties(config):
+def modbus_calc_properties(config: ConfigType) -> tuple[int, int]:
     byte_offset = 0
     reg_count = 0
     if CONF_OFFSET in config:
@@ -307,8 +307,12 @@ def modbus_calc_properties(config):
 
 
 async def add_modbus_base_properties(
-    var, config, sensor_type, lambda_param_type=cg.float_, lambda_return_type=float
-):
+    var: cg.MockObj,
+    config: ConfigType,
+    sensor_type: cg.MockObjClass,
+    lambda_param_type: cg.MockObj = cg.float_,
+    lambda_return_type: Any = float,
+) -> None:
     if CONF_CUSTOM_PDU in config:
         cg.add(var.set_custom_pdu(config[CONF_CUSTOM_PDU]))
 
@@ -347,8 +351,11 @@ _CALLBACK_AUTOMATIONS = (
 )
 
 
-async def to_code(config):
-    var = cg.new_Pvariable(config[CONF_ID])
+async def to_code(config: ConfigType) -> None:
+    # Await the hub first, so no entity can bind to a controller that doesn't have one yet.
+    hub = await cg.get_variable(config[modbus.CONF_MODBUS_ID])
+    var = cg.new_Pvariable(config[CONF_ID], hub, config[CONF_ADDRESS])
+    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(
@@ -356,17 +363,22 @@ async def to_code(config):
             modbus.command_options_expression(config, direction="read")
         )
     )
-    await register_modbus_device(var, config)
     await automation.build_callback_automations(var, config, _CALLBACK_AUTOMATIONS)
 
 
-async def register_modbus_device(var, config):
+async def register_modbus_device(var: cg.MockObj, config: ConfigType) -> cg.MockObj:
+    # Remove before 2027.3.0
+    _LOGGER.warning(
+        "'modbus_controller.register_modbus_device' is deprecated, use "
+        "'modbus.register_modbus_client_device' and set the address on your own "
+        "class instead. Will be removed in 2027.3.0"
+    )
     cg.add(var.set_address(config[CONF_ADDRESS]))
     await cg.register_component(var, config)
     return await modbus.register_modbus_client_device(var, config)
 
 
-def function_code_to_register(function_code):
+def function_code_to_register(function_code: str) -> cg.MockObj:
     FUNCTION_CODE_TYPE_MAP = {
         "read_coils": EntityType.COIL,
         "read_discrete_inputs": EntityType.DISCRETE_INPUT,
diff --git a/esphome/components/modbus_controller/modbus_controller.cpp b/esphome/components/modbus_controller/modbus_controller.cpp
index 20b8f516e9a..9d7b719e153 100644
--- a/esphome/components/modbus_controller/modbus_controller.cpp
+++ b/esphome/components/modbus_controller/modbus_controller.cpp
@@ -10,6 +10,73 @@ static const char *const TAG = "modbus_controller";
 
 void ModbusController::setup() { this->create_polling_commands_(); }
 
+void WriterDevice::warn_write_buffer_deprecated(const LogString *platform, uint16_t address) {
+  if (this->write_buffer_deprecated_warned_)
+    return;
+  this->write_buffer_deprecated_warned_ = true;
+  ESP_LOGW(TAG,
+           "Modbus %s (address 0x%X): filling the write_lambda buffer parameter is deprecated; call a write helper / "
+           "queue_pdu() on the entity (item) instead. The buffer parameter is removed in 2027.3.0",
+           LOG_STR_ARG(platform), address);
+}
+
+bool WriterDevice::send_raw_frame_deprecated(std::span<const uint8_t> frame) {
+  if (frame.empty())
+    return false;
+  this->dispatched_ = true;
+  return this->parent_->queue_pdu(frame[0], frame.subspan(1), this);
+}
+
+void WriterDevice::set_controller(ModbusController *controller) {
+  this->controller_ = controller;
+  this->set_parent(controller->hub());
+  this->set_address(controller->device_address());
+}
+
+void WriterDevice::notify_online_(std::span<const uint8_t> request_pdu) {
+  if (this->controller_ != nullptr)
+    this->controller_->set_online(true, fc_of(request_pdu), addr_of(request_pdu));
+}
+
+void WriterDevice::on_response(std::span<const uint8_t> request_pdu, std::span<const uint8_t> response_pdu) {
+  this->notify_online_(request_pdu);
+  this->dispatch_response_(request_pdu, response_pdu, std::nullopt);
+}
+
+void WriterDevice::on_error(std::span<const uint8_t> request_pdu, modbus::ExceptionCode exception_code) {
+  ESP_LOGW(TAG, "Modbus error function code: 0x%X register 0x%X exception: %d", fc_of(request_pdu),
+           addr_of(request_pdu), static_cast<uint8_t>(exception_code));
+  this->notify_online_(request_pdu);  // an exception is still a legitimate reply -> device is online
+  this->dispatch_response_(request_pdu, {}, exception_code);
+}
+
+// Fired once per wire transmission (including hub re-queues from a retry), so the on_command_sent trigger
+// reflects when the frame actually went out, not when it was queued.
+void WriterDevice::on_sent(std::span<const uint8_t> request_pdu) {
+  if (this->controller_ != nullptr)
+    this->controller_->command_sent(fc_of(request_pdu), addr_of(request_pdu));
+}
+
+void WriterDevice::on_not_sent(std::span<const uint8_t> request_pdu) {
+  // Only the offline teardown reaches this (a supersede retires silently), so the frame is genuinely
+  // lost; a dropped write was already published optimistically, so surface it.
+  if (modbus::helpers::is_function_code_write(fc_of(request_pdu))) {
+    ESP_LOGW(TAG, "Write not sent: function 0x%X register 0x%X", fc_of(request_pdu), addr_of(request_pdu));
+  } else {
+    ESP_LOGD(TAG, "Request not sent: function 0x%X register 0x%X", fc_of(request_pdu), addr_of(request_pdu));
+  }
+}
+
+bool WriterDevice::on_no_response(std::span<const uint8_t> request_pdu) {
+  if (this->controller_ == nullptr)
+    return false;
+  this->controller_->increment_non_response_count();
+  if (this->controller_->can_send())
+    return true;  // the hub re-queues the frame it is holding; on_sent fires again on the retry
+  this->controller_->set_online(false, fc_of(request_pdu), addr_of(request_pdu));
+  return false;
+}
+
 ModbusCommandItem::ModbusCommandItem(ModbusController &controller, modbus::ModbusClientHub *parent, uint8_t address,
                                      RegisterRange &&range)
     : modbus::ModbusClientDevice(parent, address),
diff --git a/esphome/components/modbus_controller/modbus_controller.h b/esphome/components/modbus_controller/modbus_controller.h
index 1db07f1ee8d..1f36d5a7c87 100644
--- a/esphome/components/modbus_controller/modbus_controller.h
+++ b/esphome/components/modbus_controller/modbus_controller.h
@@ -232,6 +232,115 @@ struct RegisterRange {
   SensorSet sensors;  // all sensors of this range
 };
 
+/// A hub device owned by a writer entity (switch/number/select/output) through WriterEntity.
+/// Centralises the feedback to the controller - online/offline tracking, retry counting and the
+/// on_command_sent trigger - and records every dispatch, so a write lambda can tell "I sent it myself"
+/// from "use the default write". The hub base is inherited protected, so the public members below are
+/// the entity's whole request API and nothing can bypass the recording or re-target the device.
+class WriterDevice final : protected modbus::ModbusClientDevice {
+ protected:
+  void on_response(std::span<const uint8_t> request_pdu, std::span<const uint8_t> response_pdu) override;
+  void on_error(std::span<const uint8_t> request_pdu, modbus::ExceptionCode exception_code) override;
+  void on_sent(std::span<const uint8_t> request_pdu) override;
+  void on_not_sent(std::span<const uint8_t> request_pdu) override;
+  bool on_no_response(std::span<const uint8_t> request_pdu) override;
+
+  void notify_online_(std::span<const uint8_t> request_pdu);
+  /// Function code / register address decoded from a request PDU ([fc, addr_hi, addr_lo, ...]).
+  static int fc_of(std::span<const uint8_t> pdu) { return pdu.empty() ? 0 : (pdu[0] & modbus::FUNCTION_CODE_MASK); }
+  static int addr_of(std::span<const uint8_t> pdu) {
+    return pdu.size() >= 3 ? modbus::helpers::get_data<uint16_t>(pdu.data(), 1) : 0;
+  }
+
+  /// Declared before controller_ so they land in the padding after ModbusClientDevice::custom_response_warned_
+  /// instead of adding a word to every entity that owns a device.
+  /// dispatched_: a frame was queued since the last clear_dispatched_().
+  /// write_buffer_deprecated_warned_: warn-once for the legacy write_lambda buffer parameter.
+  bool dispatched_{false};
+  bool write_buffer_deprecated_warned_{false};
+  ModbusController *controller_{nullptr};
+
+ public:
+  /// Whether a frame was queued to the hub since the last clear_dispatched_().
+  bool dispatched() const { return this->dispatched_; }
+
+  bool write_single_register(uint16_t address, uint16_t value) {
+    this->dispatched_ = true;
+    return modbus::ModbusClientDevice::write_single_register(address, value);
+  }
+  bool write_single_coil(uint16_t address, bool value) {
+    this->dispatched_ = true;
+    return modbus::ModbusClientDevice::write_single_coil(address, value);
+  }
+  bool write_multiple_registers(uint16_t address, std::span<const uint16_t> values) {
+    this->dispatched_ = true;
+    return modbus::ModbusClientDevice::write_multiple_registers(address, values);
+  }
+  bool write_multiple_coils(uint16_t address, std::span<const bool> values) {
+    this->dispatched_ = true;
+    return modbus::ModbusClientDevice::write_multiple_coils(address, values);
+  }
+  bool write_multiple_coils(uint16_t address, modbus::PackedBits bits) {
+    this->dispatched_ = true;
+    return modbus::ModbusClientDevice::write_multiple_coils(address, bits);
+  }
+  bool queue_pdu(std::span<const uint8_t> pdu, modbus::CommandOptions options = {}) {
+    this->dispatched_ = true;
+    return modbus::ModbusClientDevice::queue_pdu(pdu, options);
+  }
+  /// Send a legacy raw frame (address + function code + data) to the frame's own address.
+  /// Serves only the deprecated write_lambda buffer path. Remove before 2027.3.0.
+  bool send_raw_frame_deprecated(std::span<const uint8_t> frame);
+
+  void clear_tx_queue_for_device() { modbus::ModbusClientDevice::clear_tx_queue_for_device(); }
+
+  // Entity plumbing, public because the owning WriterEntity holds the only reachable instance (device_ is
+  // protected there and the hub sees just the masked base) - reachability is the access gate, not a friend.
+  void set_controller(ModbusController *controller);
+  void clear_dispatched() { this->dispatched_ = false; }
+  /// 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);
+};
+
+/// Gives a writer entity the write API of the WriterDevice it owns. The device is a member, not a base:
+/// the mixin declares no virtual function, so an entity mixing it in gains no second vtable and all the
+/// writer platforms share the single WriterDevice vtable instead of each emitting its own copy.
+/// The forwarders keep `item->write_*()` working unchanged inside a write_lambda.
+class WriterEntity {
+ public:
+  bool dispatched() const { return this->device_.dispatched(); }
+  bool write_single_register(uint16_t address, uint16_t value) {
+    return this->device_.write_single_register(address, value);
+  }
+  bool write_single_coil(uint16_t address, bool value) { return this->device_.write_single_coil(address, value); }
+  bool write_multiple_registers(uint16_t address, std::span<const uint16_t> values) {
+    return this->device_.write_multiple_registers(address, values);
+  }
+  bool write_multiple_coils(uint16_t address, std::span<const bool> values) {
+    return this->device_.write_multiple_coils(address, values);
+  }
+  bool write_multiple_coils(uint16_t address, modbus::PackedBits bits) {
+    return this->device_.write_multiple_coils(address, bits);
+  }
+  bool queue_pdu(std::span<const uint8_t> pdu, modbus::CommandOptions options = {}) {
+    return this->device_.queue_pdu(pdu, options);
+  }
+  void clear_tx_queue_for_device() { this->device_.clear_tx_queue_for_device(); }
+
+ protected:
+  bool send_raw_frame_deprecated_(std::span<const uint8_t> frame) {
+    return this->device_.send_raw_frame_deprecated(frame);
+  }
+  void set_controller_(ModbusController *controller) { this->device_.set_controller(controller); }
+  void clear_dispatched_() { this->device_.clear_dispatched(); }
+  void warn_write_buffer_deprecated_(const LogString *platform, uint16_t address) {
+    this->device_.warn_write_buffer_deprecated(platform, address);
+  }
+
+  WriterDevice device_;
+};
+
 /// A single modbus command. Each command is its own ModbusClientDevice: it sends its frame to the hub
 /// and the hub routes the response back to this object's on_modbus_* callbacks, so the controller no
 /// longer has to match responses to a FIFO queue.
@@ -398,17 +507,16 @@ inline bool offline_retry_due(uint16_t update_counter, uint16_t module_offline_a
 
 class ModbusController final : public PollingComponent {
  public:
+  // The controller is not itself a modbus device - its commands and writer entities send as their own
+  // devices, built against this hub + address.
+  ModbusController(modbus::ModbusClientHub *hub, uint8_t address) : hub_(hub), address_(address) {}
+
   void dump_config() override;
   // No loop() override: the hub owns transmit/receive timing and each command routes its own
   // response, so the controller never joins the looping components at all.
   void setup() override;
   void update() override;
 
-  // The controller is not itself a modbus device - its commands and writer entities send as their own
-  // devices. It only owns the hub + address so those senders can be built against them.
-  void set_parent(modbus::ModbusClientHub *hub) { this->hub_ = hub; }
-  void set_address(uint8_t address) { this->address_ = address; }
-
   /// The hub and modbus address this controller talks to. Used to build commands/entities that send as
   /// their own device.
   modbus::ModbusClientHub *hub() const { return this->hub_; }
diff --git a/esphome/components/modbus_controller/number/__init__.py b/esphome/components/modbus_controller/number/__init__.py
index a43e10a51e1..6a5b7041b8a 100644
--- a/esphome/components/modbus_controller/number/__init__.py
+++ b/esphome/components/modbus_controller/number/__init__.py
@@ -3,6 +3,7 @@ from esphome.components import number
 from esphome.components.modbus.helpers import (
     MODBUS_WRITE_REGISTER_TYPE,
     SENSOR_VALUE_TYPE,
+    RegisterValues,
 )
 import esphome.config_validation as cv
 from esphome.const import (
@@ -13,6 +14,7 @@ from esphome.const import (
     CONF_MULTIPLY,
     CONF_STEP,
 )
+from esphome.types import ConfigType
 
 from .. import (
     ModbusItemBaseSchema,
@@ -43,7 +45,7 @@ ModbusNumber = modbus_controller_ns.class_(
 )
 
 
-def validate_min_max(config):
+def validate_min_max(config: ConfigType) -> ConfigType:
     if config[CONF_MAX_VALUE] <= config[CONF_MIN_VALUE]:
         raise cv.Invalid("max_value must be greater than min_value")
     if config[CONF_MIN_VALUE] < -16777215:
@@ -53,7 +55,7 @@ def validate_min_max(config):
     return config
 
 
-def validate_modbus_number(config):
+def validate_modbus_number(config: ConfigType) -> ConfigType:
     # custom_command is the deprecated alias for custom_pdu (migrated later in final validate).
     has_custom = CONF_CUSTOM_PDU in config or CONF_CUSTOM_COMMAND in config
     if not has_custom and CONF_ADDRESS not in config:
@@ -89,7 +91,7 @@ CONFIG_SCHEMA = cv.All(
 FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     byte_offset, reg_count = modbus_calc_properties(config)
     var = cg.new_Pvariable(
         config[CONF_ID],
@@ -124,7 +126,7 @@ async def to_code(config):
             [
                 (ModbusNumber.operator("ptr"), "item"),
                 (cg.float_, "x"),
-                (cg.std_vector.template(cg.uint16).operator("ref"), "payload"),
+                (RegisterValues.operator("ref"), "payload"),
             ],
             return_type=cg.optional.template(float),
         )
diff --git a/esphome/components/modbus_controller/number/modbus_number.cpp b/esphome/components/modbus_controller/number/modbus_number.cpp
index 7903b2e3177..e890a2a9ac6 100644
--- a/esphome/components/modbus_controller/number/modbus_number.cpp
+++ b/esphome/components/modbus_controller/number/modbus_number.cpp
@@ -1,4 +1,3 @@
-#include <vector>
 #include "modbus_number.h"
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
@@ -29,62 +28,73 @@ void ModbusNumber::parse_and_publish(std::span<const uint8_t> data) {
 }
 
 void ModbusNumber::control(float value) {
-  optional<ModbusCommandItem> write_cmd;
-  std::vector<uint16_t> data;
+  this->clear_dispatched_();
+  // A new write supersedes this entity's own not-yet-sent writes: drop them (and detach any in-flight one)
+  // so a rapidly-changing value writes the latest, not every intermediate.
+  this->clear_tx_queue_for_device();
+  modbus::RegisterValues data;
   float write_value = value;
-  // Is there are lambda configured?
   if (this->write_transform_func_.has_value()) {
-    // data is passed by reference
-    // the lambda can fill the empty vector directly
-    // in that case the return value is ignored
+    // The lambda may drive the write itself via item->write_*(), override the value (return a value), or
+    // (deprecated) fill `data` with the register words to write.
     auto val = (*this->write_transform_func_)(this, value, data);
-    if (val.has_value()) {
-      ESP_LOGV(TAG, "Value overwritten by lambda");
-      write_value = val.value();
-    } else {
+    if (this->dispatched()) {
+      this->publish_state(value);
+      return;
+    }
+    if (!data.empty()) {
+      // Deprecated buffer path (frozen): the lambda filled a legacy raw frame as words; pack it big-endian.
+      this->warn_write_buffer_deprecated_(LOG_STR("number"), this->start_address);
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
+      char hex_buf[format_hex_pretty_uint16_size(MODBUS_NUMBER_MAX_LOG_REGISTERS)];
+#endif
+      ESP_LOGV(TAG, "Modbus Number write raw: %s",
+               format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
+      // Sized to hold RegisterValues at capacity, so a full buffer can never truncate into a valid frame.
+      StaticVector<uint8_t, modbus::MAX_NUM_OF_REGISTERS_TO_READ * 2> bytes;
+      for (uint16_t word : data) {
+        const auto word_bytes = decode_value(word);
+        bytes.push_back(word_bytes[0]);
+        bytes.push_back(word_bytes[1]);
+      }
+      if (!this->send_raw_frame_deprecated_(std::span<const uint8_t>(bytes.data(), bytes.size()))) {
+        ESP_LOGW(TAG, "Modbus write for '%s' was refused by the hub; state not published", this->get_name().c_str());
+        return;
+      }
+      this->publish_state(value);
+      return;
+    }
+    if (!val.has_value()) {
       ESP_LOGV(TAG, "Communication handled by lambda - exiting control");
       return;
     }
+    ESP_LOGV(TAG, "Value overwritten by lambda");
+    write_value = val.value();
   } else {
     write_value = this->multiply_by_ * write_value;
   }
 
-  if (!data.empty()) {
-#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
-    char hex_buf[format_hex_pretty_uint16_size(MODBUS_NUMBER_MAX_LOG_REGISTERS)];
-#endif
-    ESP_LOGV(TAG, "Modbus Number write raw: %s",
-             format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
-    write_cmd.emplace(ModbusCommandItem::create_custom_command(
-        this->parent_, data,
-        [this](modbus::EntityType register_type, uint16_t start_address, std::span<const uint8_t> data) {
-          this->parent_->on_write_register_response(register_type, this->start_address, data);
-        }));
-  } else {
-    std::vector<uint16_t> payload;
-    modbus::helpers::float_to_payload(payload, write_value, this->sensor_value_type);
+  modbus::helpers::float_to_payload(data, write_value, this->sensor_value_type);
+  // float_to_payload() appends nothing for RAW, so an empty payload must be caught before data[0] below.
+  if (data.empty()) {
+    ESP_LOGW(TAG, "No payload was created for updating number");
+    return;
+  }
 
-    ESP_LOGD(TAG,
-             "Updating register: connected Sensor=%s start address=0x%X register count=%d new value=%.02f (val=%.02f)",
-             this->get_name().c_str(), this->start_address, this->register_count, value, write_value);
+  ESP_LOGD(TAG,
+           "Updating register: connected Sensor=%s start address=0x%X register count=%d new value=%.02f (val=%.02f)",
+           this->get_name().c_str(), this->start_address, this->register_count, value, write_value);
 
-    // Create and send the write command
-    if (this->register_count == 1 && !this->use_write_multiple_) {
-      write_cmd.emplace(
-          ModbusCommandItem::create_write_single_command(this->parent_, this->write_address(), payload[0]));
-    } else {
-      write_cmd.emplace(ModbusCommandItem::create_write_multiple_command(this->parent_, this->write_address(),
-                                                                         this->register_count, payload));
-    }
-    // publish new value
-    write_cmd->on_data_func = [this, value](modbus::EntityType register_type, uint16_t start_address,
-                                            std::span<const uint8_t> data) {
-      // gets called when the write command is ack'd from the device
-      this->parent_->on_write_register_response(register_type, start_address, data);
-      this->publish_state(value);
-    };
+  bool queued;
+  if (this->register_count == 1 && !this->use_write_multiple_) {
+    queued = this->write_single_register(this->write_address(), data[0]);
+  } else {
+    queued = this->write_multiple_registers(this->write_address(), data);
+  }
+  if (!queued) {
+    ESP_LOGW(TAG, "Modbus write for '%s' was refused by the hub; state not published", this->get_name().c_str());
+    return;
   }
-  this->parent_->queue_command(std::move(*write_cmd));
   this->publish_state(value);
 }
 void ModbusNumber::dump_config() { LOG_NUMBER(TAG, "Modbus Number", this); }
diff --git a/esphome/components/modbus_controller/number/modbus_number.h b/esphome/components/modbus_controller/number/modbus_number.h
index 538a982f80c..59c76e18f2a 100644
--- a/esphome/components/modbus_controller/number/modbus_number.h
+++ b/esphome/components/modbus_controller/number/modbus_number.h
@@ -10,7 +10,7 @@ namespace esphome::modbus_controller {
 
 using value_to_data_t = std::function<float>(float);
 
-class ModbusNumber final : public number::Number, public Component, public SensorItem {
+class ModbusNumber final : public number::Number, public Component, public SensorItem, public WriterEntity {
  public:
   ModbusNumber(modbus::EntityType register_type, uint16_t start_address, uint8_t offset, uint32_t bitmask,
                SensorValueType value_type, int register_count, bool force_new_range) {
@@ -26,11 +26,11 @@ class ModbusNumber final : public number::Number, public Component, public Senso
   void dump_config() override;
   void parse_and_publish(std::span<const uint8_t> data) override;
   float get_setup_priority() const override { return setup_priority::HARDWARE; }
-  void set_parent(ModbusController *parent) { this->parent_ = parent; }
+  void set_parent(ModbusController *parent) { this->set_controller_(parent); }
   void set_write_multiply(float factor) { this->multiply_by_ = factor; }
 
   using transform_func_t = optional<float> (*)(ModbusNumber *, float, std::span<const uint8_t>);
-  using write_transform_func_t = optional<float> (*)(ModbusNumber *, float, std::vector<uint16_t> &);
+  using write_transform_func_t = optional<float> (*)(ModbusNumber *, float, modbus::RegisterValues &);
   void set_template(transform_func_t f) { this->transform_func_ = f; }
   void set_write_template(write_transform_func_t f) { this->write_transform_func_ = f; }
   void set_use_write_mutiple(bool use_write_multiple) { this->use_write_multiple_ = use_write_multiple; }
@@ -39,7 +39,6 @@ class ModbusNumber final : public number::Number, public Component, public Senso
   void control(float value) override;
   optional<transform_func_t> transform_func_{nullopt};
   optional<write_transform_func_t> write_transform_func_{nullopt};
-  ModbusController *parent_{nullptr};
   float multiply_by_{1.0};
   bool use_write_multiple_{false};
 };
diff --git a/esphome/components/modbus_controller/output/__init__.py b/esphome/components/modbus_controller/output/__init__.py
index 178c99caa11..34a0f488ec0 100644
--- a/esphome/components/modbus_controller/output/__init__.py
+++ b/esphome/components/modbus_controller/output/__init__.py
@@ -1,8 +1,13 @@
 import esphome.codegen as cg
 from esphome.components import output
-from esphome.components.modbus.helpers import SENSOR_VALUE_TYPE
+from esphome.components.modbus.helpers import (
+    SENSOR_VALUE_TYPE,
+    PduBuffer,
+    RegisterValues,
+)
 import esphome.config_validation as cv
 from esphome.const import CONF_ADDRESS, CONF_ID, CONF_MULTIPLY
+from esphome.types import ConfigType
 
 from .. import (
     ModbusItemBaseSchema,
@@ -73,7 +78,7 @@ CONFIG_SCHEMA = cv.typed_schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     byte_offset, reg_count = modbus_calc_properties(config)
     # Binary Output
     write_template = None
@@ -89,7 +94,7 @@ async def to_code(config):
                 [
                     (ModbusBinaryOutput.operator("ptr"), "item"),
                     (cg.bool_, "x"),
-                    (cg.std_vector.template(cg.uint8).operator("ref"), "payload"),
+                    (PduBuffer.operator("ref"), "payload"),
                 ],
                 return_type=cg.optional.template(bool),
             )
@@ -109,7 +114,7 @@ async def to_code(config):
                 [
                     (ModbusFloatOutput.operator("ptr"), "item"),
                     (cg.float_, "x"),
-                    (cg.std_vector.template(cg.uint16).operator("ref"), "payload"),
+                    (RegisterValues.operator("ref"), "payload"),
                 ],
                 return_type=cg.optional.template(float),
             )
diff --git a/esphome/components/modbus_controller/output/modbus_output.cpp b/esphome/components/modbus_controller/output/modbus_output.cpp
index 48249f4387e..b05d3889fd8 100644
--- a/esphome/components/modbus_controller/output/modbus_output.cpp
+++ b/esphome/components/modbus_controller/output/modbus_output.cpp
@@ -2,6 +2,8 @@
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
+#include <array>
+
 namespace esphome::modbus_controller {
 
 static const char *const TAG = "modbus_controller.output";
@@ -13,25 +15,33 @@ static constexpr size_t MODBUS_OUTPUT_MAX_LOG_BYTES = 64;
  *
  */
 void ModbusFloatOutput::write_state(float value) {
-  std::vector<uint16_t> data;
+  this->clear_dispatched_();
+  // A new write supersedes this entity's own not-yet-sent writes: drop them (and detach any in-flight one)
+  // so a rapidly-changing value writes the latest, not every intermediate.
+  this->clear_tx_queue_for_device();
+  modbus::RegisterValues data;
   auto original_value = value;
-  // Is there are lambda configured?
   if (this->write_transform_func_.has_value()) {
-    // data is passed by reference
-    // the lambda can fill the empty vector directly
-    // in that case the return value is ignored
+    // The lambda may drive the write itself via item->write_*(), override the value (return a value), or
+    // (deprecated) fill `data` with the register words to write.
     auto val = (*this->write_transform_func_)(this, value, data);
-    if (val.has_value()) {
-      ESP_LOGV(TAG, "Value overwritten by lambda");
-      value = val.value();
-    } else {
+    if (this->dispatched()) {
+      return;
+    }
+    if (!data.empty()) {
+      // Deprecated buffer path (frozen): the lambda supplied the register words for the shared write below.
+      this->warn_write_buffer_deprecated_(LOG_STR("float output"), this->start_address);
+    } else if (!val.has_value()) {
       ESP_LOGV(TAG, "Communication handled by lambda - exiting control");
       return;
+    } else {
+      ESP_LOGV(TAG, "Value overwritten by lambda");
+      value = val.value();
     }
   } else {
     value = this->multiply_by_ * value;
   }
-  // lambda didn't set payload
+
   if (data.empty()) {
     modbus::helpers::float_to_payload(data, value, this->sensor_value_type);
   }
@@ -57,16 +67,15 @@ void ModbusFloatOutput::write_state(float value) {
     return;
   }
 
-  // Create and send the write command
-  optional<ModbusCommandItem> write_cmd;
+  bool queued;
   if (this->register_count == 1 && !this->use_write_multiple_) {
-    write_cmd.emplace(
-        ModbusCommandItem::create_write_single_command(this->parent_, this->start_address + this->offset, data[0]));
+    queued = this->write_single_register(this->write_address(), data[0]);
   } else {
-    write_cmd.emplace(ModbusCommandItem::create_write_multiple_command(
-        this->parent_, this->start_address + this->offset, data.size(), data));
+    queued = this->write_multiple_registers(this->write_address(), data);
+  }
+  if (!queued) {
+    ESP_LOGW(TAG, "Modbus output write (address 0x%X) was refused by the hub", this->write_address());
   }
-  this->parent_->queue_command(std::move(*write_cmd));
 }
 
 void ModbusFloatOutput::dump_config() {
@@ -81,50 +90,52 @@ void ModbusFloatOutput::dump_config() {
 
 // ModbusBinaryOutput
 void ModbusBinaryOutput::write_state(bool state) {
-  // This will be called every time the user requests a state change.
-  optional<ModbusCommandItem> cmd;
-  std::vector<uint8_t> data;
+  this->clear_dispatched_();
+  // A new write supersedes this entity's own not-yet-sent writes: drop them (and detach any in-flight one)
+  // so a rapidly-changing value writes the latest, not every intermediate.
+  this->clear_tx_queue_for_device();
+  modbus::helpers::PduBuffer data;
 
-  // Is there are lambda configured?
   if (this->write_transform_func_.has_value()) {
-    // data is passed by reference
-    // the lambda can fill the empty vector directly
-    // in that case the return value is ignored
+    // The lambda may drive the write itself via item->write_*/queue_pdu(), override the value (return a value),
+    // or (deprecated) fill `data` with a custom PDU.
     auto val = (*this->write_transform_func_)(this, state, data);
-    if (val.has_value()) {
-      ESP_LOGV(TAG, "Value overwritten by lambda");
-      state = val.value();
-    } else {
+    if (this->dispatched()) {
+      return;
+    }
+    if (!data.empty()) {
+      this->warn_write_buffer_deprecated_(LOG_STR("binary output"), this->start_address);
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
+      char hex_buf[format_hex_pretty_size(MODBUS_OUTPUT_MAX_LOG_BYTES)];
+#endif
+      ESP_LOGV(TAG, "Modbus binary output write raw: %s",
+               format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
+      // The lambda filled a legacy raw frame (device address + function code + data).
+      if (!this->send_raw_frame_deprecated_(data)) {
+        ESP_LOGW(TAG, "Modbus output write (address 0x%X) was refused by the hub", this->write_address());
+      }
+      return;
+    }
+    if (!val.has_value()) {
       ESP_LOGV(TAG, "Communication handled by lambda - exiting control");
       return;
     }
+    ESP_LOGV(TAG, "Value overwritten by lambda");
+    state = val.value();
   }
-  if (!data.empty()) {
-#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
-    char hex_buf[format_hex_pretty_size(MODBUS_OUTPUT_MAX_LOG_BYTES)];
-#endif
-    ESP_LOGV(TAG, "Modbus binary output write raw: %s",
-             format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
-    cmd.emplace(ModbusCommandItem::create_custom_command(
-        this->parent_, data,
-        [this](modbus::EntityType register_type, uint16_t start_address, std::span<const uint8_t> data) {
-          this->parent_->on_write_register_response(register_type, this->start_address, data);
-        }));
+  ESP_LOGV(TAG, "Write new state: value is %s, type is %d address = %X, offset = %x", ONOFF(state),
+           (int) this->register_type, this->start_address, this->offset);
+  // offset for coil and discrete inputs is the coil/register number not bytes
+  bool queued;
+  if (this->use_write_multiple_) {
+    std::array<bool, 1> states{state};
+    queued = this->write_multiple_coils(this->write_address(), states);
   } else {
-    ESP_LOGV(TAG, "Write new state: value is %s, type is %d address = %X, offset = %x", ONOFF(state),
-             (int) this->register_type, this->start_address, this->offset);
-
-    // offset for coil and discrete inputs is the coil/register number not bytes
-    if (this->use_write_multiple_) {
-      std::vector<bool> states{state};
-      cmd.emplace(
-          ModbusCommandItem::create_write_multiple_coils(this->parent_, this->start_address + this->offset, states));
-    } else {
-      cmd.emplace(
-          ModbusCommandItem::create_write_single_coil(this->parent_, this->start_address + this->offset, state));
-    }
+    queued = this->write_single_coil(this->write_address(), state);
+  }
+  if (!queued) {
+    ESP_LOGW(TAG, "Modbus output write (address 0x%X) was refused by the hub", this->write_address());
   }
-  this->parent_->queue_command(std::move(*cmd));
 }
 
 void ModbusBinaryOutput::dump_config() {
diff --git a/esphome/components/modbus_controller/output/modbus_output.h b/esphome/components/modbus_controller/output/modbus_output.h
index e79c442aa4a..b942dcea628 100644
--- a/esphome/components/modbus_controller/output/modbus_output.h
+++ b/esphome/components/modbus_controller/output/modbus_output.h
@@ -8,26 +8,24 @@
 
 namespace esphome::modbus_controller {
 
-class ModbusFloatOutput final : public output::FloatOutput, public Component, public SensorItem {
+class ModbusFloatOutput final : public output::FloatOutput, public Component, public SensorItem, public WriterEntity {
  public:
   ModbusFloatOutput(uint16_t start_address, uint8_t offset, SensorValueType value_type, int register_count) {
     this->register_type = modbus::EntityType::HOLDING;
-    this->set_address(start_address);
-    this->set_offset_from_start_address(offset);
+    this->set_address(start_address + offset);
+    this->set_offset_from_start_address(0);
     this->bitmask = 0xFFFFFFFF;
     this->register_count = register_count;
     this->sensor_value_type = value_type;
-    this->set_address(this->start_address + offset);
-    this->set_offset_from_start_address(0);
   }
   void dump_config() override;
 
-  void set_parent(ModbusController *parent) { this->parent_ = parent; }
+  void set_parent(ModbusController *parent) { this->set_controller_(parent); }
   void set_write_multiply(float factor) { this->multiply_by_ = factor; }
   // Do nothing
   void parse_and_publish(std::span<const uint8_t> data) override{};
 
-  using write_transform_func_t = optional<float> (*)(ModbusFloatOutput *, float, std::vector<uint16_t> &);
+  using write_transform_func_t = optional<float> (*)(ModbusFloatOutput *, float, modbus::RegisterValues &);
   void set_write_template(write_transform_func_t f) { this->write_transform_func_ = f; }
   void set_use_write_mutiple(bool use_write_multiple) { this->use_write_multiple_ = use_write_multiple; }
 
@@ -35,29 +33,28 @@ class ModbusFloatOutput final : public output::FloatOutput, public Component, pu
   void write_state(float value) override;
   optional<write_transform_func_t> write_transform_func_{nullopt};
 
-  ModbusController *parent_{nullptr};
   float multiply_by_{1.0};
   bool use_write_multiple_{false};
 };
 
-class ModbusBinaryOutput final : public output::BinaryOutput, public Component, public SensorItem {
+class ModbusBinaryOutput final : public output::BinaryOutput, public Component, public SensorItem, public WriterEntity {
  public:
   ModbusBinaryOutput(uint16_t start_address, uint8_t offset) {
     this->register_type = modbus::EntityType::COIL;
-    this->set_address(start_address);
+    // A coil offset is a coil count; fold it into the address.
+    this->set_address(start_address + offset);
     this->bitmask = 0xFFFFFFFF;
     this->sensor_value_type = SensorValueType::BIT;
     this->register_count = 1;
-    this->set_address(this->start_address + offset);
     this->set_offset_from_start_address(0);
   }
   void dump_config() override;
 
-  void set_parent(ModbusController *parent) { this->parent_ = parent; }
+  void set_parent(ModbusController *parent) { this->set_controller_(parent); }
   // Do nothing
   void parse_and_publish(std::span<const uint8_t> data) override{};
 
-  using write_transform_func_t = optional<bool> (*)(ModbusBinaryOutput *, bool, std::vector<uint8_t> &);
+  using write_transform_func_t = optional<bool> (*)(ModbusBinaryOutput *, bool, modbus::helpers::PduBuffer &);
   void set_write_template(write_transform_func_t f) { this->write_transform_func_ = f; }
   void set_use_write_mutiple(bool use_write_multiple) { this->use_write_multiple_ = use_write_multiple; }
 
@@ -65,7 +62,6 @@ class ModbusBinaryOutput final : public output::BinaryOutput, public Component,
   void write_state(bool state) override;
   optional<write_transform_func_t> write_transform_func_{nullopt};
 
-  ModbusController *parent_{nullptr};
   bool use_write_multiple_{false};
 };
 
diff --git a/esphome/components/modbus_controller/select/__init__.py b/esphome/components/modbus_controller/select/__init__.py
index 1d77f9235d0..07893e33036 100644
--- a/esphome/components/modbus_controller/select/__init__.py
+++ b/esphome/components/modbus_controller/select/__init__.py
@@ -1,8 +1,16 @@
+from collections.abc import Callable
+from typing import Any
+
 import esphome.codegen as cg
 from esphome.components import select
-from esphome.components.modbus.helpers import SENSOR_VALUE_TYPE, TYPE_REGISTER_MAP
+from esphome.components.modbus.helpers import (
+    SENSOR_VALUE_TYPE,
+    TYPE_REGISTER_MAP,
+    RegisterValues,
+)
 import esphome.config_validation as cv
 from esphome.const import CONF_ADDRESS, CONF_ID, CONF_LAMBDA, CONF_OPTIMISTIC
+from esphome.types import ConfigType
 
 from .. import (
     ModbusController,
@@ -29,8 +37,8 @@ ModbusSelect = modbus_controller_ns.class_(
 )
 
 
-def ensure_option_map():
-    def validator(value):
+def ensure_option_map() -> Callable[[Any], dict[str, int]]:
+    def validator(value: Any) -> dict[str, int]:
         cv.check_not_templatable(value)
         option = cv.All(cv.string_strict)
         mapping = cv.All(cv.int_range(-(2**63), 2**63 - 1))
@@ -47,7 +55,7 @@ def ensure_option_map():
     return validator
 
 
-def register_count_value_type_min(value):
+def register_count_value_type_min(value: ConfigType) -> ConfigType:
     reg_count = value.get(CONF_REGISTER_COUNT)
     if reg_count is not None:
         value_type = value[CONF_VALUE_TYPE]
@@ -87,7 +95,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     value_type = config[CONF_VALUE_TYPE]
     reg_count = config.get(CONF_REGISTER_COUNT)
     if reg_count is None:
@@ -132,7 +140,7 @@ async def to_code(config):
                 (ModbusSelect.operator("const_ptr"), "item"),
                 (cg.std_string.operator("const").operator("ref"), "x"),
                 (cg.int64, "value"),
-                (cg.std_vector.template(cg.uint16).operator("ref"), "payload"),
+                (RegisterValues.operator("ref"), "payload"),
             ],
             return_type=cg.optional.template(cg.int64),
         )
diff --git a/esphome/components/modbus_controller/select/modbus_select.cpp b/esphome/components/modbus_controller/select/modbus_select.cpp
index 0a9383b1b0f..c1cc241d6b7 100644
--- a/esphome/components/modbus_controller/select/modbus_select.cpp
+++ b/esphome/components/modbus_controller/select/modbus_select.cpp
@@ -46,35 +46,43 @@ void ModbusSelect::control(size_t index) {
   const char *option = this->option_at(index);
   ESP_LOGD(TAG, "Found value %lld for option '%s'", *mapval, option);
 
-  std::vector<uint16_t> data;
+  this->clear_dispatched_();
+  // A new write supersedes this entity's own not-yet-sent writes: drop them (and detach any in-flight one)
+  // so a rapidly-changing value writes the latest, not every intermediate.
+  this->clear_tx_queue_for_device();
+  modbus::RegisterValues data;
 
   if (this->write_transform_func_.has_value()) {
-    // Transform func requires string parameter for backward compatibility
+    // The lambda may drive the write itself via item->write_*(), override the mapping value (return a value),
+    // or (deprecated) fill `data` with the register words to write. Transform func requires string parameter
+    // for backward compatibility.
     auto val = (*this->write_transform_func_)(this, std::string(option), *mapval, data);
-    if (val.has_value()) {
-      mapval = val;
-      ESP_LOGV(TAG, "write_lambda returned mapping value %lld", *mapval);
-    } else {
+    if (this->dispatched()) {
+      if (this->optimistic_)
+        this->publish_state(index);
+      return;
+    }
+    if (!data.empty()) {
+      // Deprecated buffer path (frozen): the lambda supplied the register words for the shared write below.
+      this->warn_write_buffer_deprecated_(LOG_STR("select"), this->start_address);
+    } else if (!val.has_value()) {
       ESP_LOGD(TAG, "Communication handled by write_lambda - exiting control");
       return;
+    } else {
+      mapval = val;
+      ESP_LOGV(TAG, "write_lambda returned mapping value %lld", *mapval);
     }
   }
 
   if (data.empty()) {
     modbus::helpers::number_to_payload(data, *mapval, this->sensor_value_type);
-  } else {
-    ESP_LOGV(TAG, "Using payload from write lambda");
+    // number_to_payload() appends nothing for RAW.
+    if (data.empty()) {
+      ESP_LOGW(TAG, "No payload was created for updating select");
+      return;
+    }
   }
 
-  if (data.empty()) {
-    ESP_LOGW(TAG, "No payload was created for updating select");
-    return;
-  }
-
-  // The command declares register_count registers, so the payload must be exactly that many words:
-  // a value type narrower than the declared width is zero-padded (the config deliberately allows
-  // register_count larger than the value type). Anything else would put a byte count on the wire
-  // that disagrees with the quantity field, which conformant devices reject.
   // register_count declares the READ range width - it may pull neighboring registers into one poll -
   // so a write covers exactly the registers the value occupies: the quantity comes from the payload,
   // never from register_count (padding to it would zero registers the user only declared for reading).
@@ -86,16 +94,17 @@ void ModbusSelect::control(size_t index) {
   }
 
   const uint16_t write_address = this->write_address();
-  optional<ModbusCommandItem> write_cmd;
+  bool queued;
   if ((this->register_count == 1) && (!this->use_write_multiple_)) {
-    write_cmd.emplace(ModbusCommandItem::create_write_single_command(this->parent_, write_address, data[0]));
+    queued = this->write_single_register(write_address, data[0]);
   } else {
-    write_cmd.emplace(
-        ModbusCommandItem::create_write_multiple_command(this->parent_, write_address, data.size(), data));
+    queued = this->write_multiple_registers(write_address, data);
   }
 
-  this->parent_->queue_command(std::move(*write_cmd));
-
+  if (!queued) {
+    ESP_LOGW(TAG, "Modbus write for '%s' was refused by the hub; state not published", this->get_name().c_str());
+    return;
+  }
   if (this->optimistic_)
     this->publish_state(index);
 }
diff --git a/esphome/components/modbus_controller/select/modbus_select.h b/esphome/components/modbus_controller/select/modbus_select.h
index 41ebd4f6580..c6ac76a45b2 100644
--- a/esphome/components/modbus_controller/select/modbus_select.h
+++ b/esphome/components/modbus_controller/select/modbus_select.h
@@ -9,7 +9,7 @@
 
 namespace esphome::modbus_controller {
 
-class ModbusSelect final : public Component, public select::Select, public SensorItem {
+class ModbusSelect final : public Component, public select::Select, public SensorItem, public WriterEntity {
  public:
   ModbusSelect(SensorValueType sensor_value_type, uint16_t start_address, uint8_t register_count, bool force_new_range,
                std::vector<int64_t> mapping) {
@@ -26,9 +26,9 @@ class ModbusSelect final : public Component, public select::Select, public Senso
 
   using transform_func_t = optional<std::string> (*)(ModbusSelect *const, int64_t, std::span<const uint8_t>);
   using write_transform_func_t = optional<int64_t> (*)(ModbusSelect *const, const std::string &, int64_t,
-                                                       std::vector<uint16_t> &);
+                                                       modbus::RegisterValues &);
 
-  void set_parent(ModbusController *const parent) { this->parent_ = parent; }
+  void set_parent(ModbusController *const parent) { this->set_controller_(parent); }
   void set_use_write_mutiple(bool use_write_multiple) { this->use_write_multiple_ = use_write_multiple; }
   void set_optimistic(bool optimistic) { this->optimistic_ = optimistic; }
   void set_template(transform_func_t f) { this->transform_func_ = f; }
@@ -40,7 +40,6 @@ class ModbusSelect final : public Component, public select::Select, public Senso
 
  protected:
   std::vector<int64_t> mapping_{};
-  ModbusController *parent_{nullptr};
   bool use_write_multiple_{false};
   bool optimistic_{false};
   optional<transform_func_t> transform_func_{nullopt};
diff --git a/esphome/components/modbus_controller/switch/__init__.py b/esphome/components/modbus_controller/switch/__init__.py
index dedd2ceedf4..c52067f9415 100644
--- a/esphome/components/modbus_controller/switch/__init__.py
+++ b/esphome/components/modbus_controller/switch/__init__.py
@@ -1,8 +1,9 @@
 import esphome.codegen as cg
 from esphome.components import switch
-from esphome.components.modbus.helpers import MODBUS_REGISTER_TYPE
+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
+from esphome.types import ConfigType
 
 from .. import (
     ModbusItemBaseSchema,
@@ -48,7 +49,7 @@ CONFIG_SCHEMA = cv.All(
 FINAL_VALIDATE_SCHEMA = validate_custom_pdu_item
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     byte_offset, _ = modbus_calc_properties(config)
     var = cg.new_Pvariable(
         config[CONF_ID],
@@ -74,7 +75,7 @@ async def to_code(config):
             [
                 (ModbusSwitch.operator("ptr"), "item"),
                 (cg.bool_, "x"),
-                (cg.std_vector.template(cg.uint8).operator("ref"), "payload"),
+                (PduBuffer.operator("ref"), "payload"),
             ],
             return_type=cg.optional.template(bool),
         )
diff --git a/esphome/components/modbus_controller/switch/modbus_switch.cpp b/esphome/components/modbus_controller/switch/modbus_switch.cpp
index 810d904d855..c942ff1e6f3 100644
--- a/esphome/components/modbus_controller/switch/modbus_switch.cpp
+++ b/esphome/components/modbus_controller/switch/modbus_switch.cpp
@@ -3,6 +3,8 @@
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
+#include <array>
+
 namespace esphome::modbus_controller {
 
 static const char *const TAG = "modbus_controller.switch";
@@ -58,57 +60,64 @@ void ModbusSwitch::parse_and_publish(std::span<const uint8_t> data) {
 }
 
 void ModbusSwitch::write_state(bool state) {
-  // This will be called every time the user requests a state change.
-  optional<ModbusCommandItem> cmd;
-  std::vector<uint8_t> data;
-  // Is there are lambda configured?
+  this->clear_dispatched_();
+  // A new write supersedes this entity's own not-yet-sent writes: drop them (and detach any in-flight one)
+  // so a rapidly-changing value writes the latest, not every intermediate.
+  this->clear_tx_queue_for_device();
+  modbus::helpers::PduBuffer data;
   if (this->write_transform_func_.has_value()) {
-    // data is passed by reference
-    // the lambda can fill the empty vector directly
-    // in that case the return value is ignored
+    // The lambda may drive the write itself via item->write_*/queue_pdu(), override the written value (return a
+    // value), or (deprecated) fill `data` with a custom PDU.
     auto val = (*this->write_transform_func_)(this, state, data);
-    if (val.has_value()) {
-      ESP_LOGV(TAG, "Value overwritten by lambda");
-      state = val.value();
-    } else {
+    if (this->dispatched()) {
+      this->publish_state(state);
+      return;
+    }
+    if (!data.empty()) {
+      this->warn_write_buffer_deprecated_(LOG_STR("switch"), this->start_address);
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
+      char hex_buf[format_hex_pretty_size(MODBUS_SWITCH_MAX_LOG_BYTES)];
+#endif
+      ESP_LOGV(TAG, "Modbus Switch write raw: %s",
+               format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
+      // The lambda filled a legacy raw frame (device address + function code + data).
+      if (!this->send_raw_frame_deprecated_(data)) {
+        ESP_LOGW(TAG, "Modbus write for '%s' was refused by the hub; state not published", this->get_name().c_str());
+        return;
+      }
+      this->publish_state(state);
+      return;
+    }
+    if (!val.has_value()) {
       ESP_LOGV(TAG, "Communication handled by lambda - exiting control");
       return;
     }
+    ESP_LOGV(TAG, "Value overwritten by lambda");
+    state = val.value();
   }
-  if (!data.empty()) {
-#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
-    char hex_buf[format_hex_pretty_size(MODBUS_SWITCH_MAX_LOG_BYTES)];
-#endif
-    ESP_LOGV(TAG, "Modbus Switch write raw: %s",
-             format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
-    cmd.emplace(ModbusCommandItem::create_custom_command(
-        this->parent_, data,
-        [this](modbus::EntityType register_type, uint16_t start_address, std::span<const uint8_t> data) {
-          this->parent_->on_write_register_response(register_type, this->start_address, data);
-        }));
-  } else {
-    ESP_LOGV(TAG, "write_state '%s': new value = %s type = %d address = %X offset = %x", this->get_name().c_str(),
-             ONOFF(state), (int) this->register_type, this->start_address, this->offset);
-    if (this->register_type == modbus::EntityType::COIL) {
-      // offset for coil and discrete inputs is the coil/register number not bytes
-      if (this->use_write_multiple_) {
-        std::vector<bool> states{state};
-        cmd.emplace(ModbusCommandItem::create_write_multiple_coils(this->parent_, this->write_address(), states));
-      } else {
-        cmd.emplace(ModbusCommandItem::create_write_single_coil(this->parent_, this->write_address(), state));
-      }
+  ESP_LOGV(TAG, "write_state '%s': new value = %s type = %d address = %X offset = %x", this->get_name().c_str(),
+           ONOFF(state), (int) this->register_type, this->start_address, this->offset);
+  bool queued;
+  if (this->register_type == EntityType::COIL) {
+    // offset for coil and discrete inputs is the coil/register number not bytes
+    if (this->use_write_multiple_) {
+      std::array<bool, 1> states{state};
+      queued = this->write_multiple_coils(this->write_address(), states);
     } else {
-      if (this->use_write_multiple_) {
-        std::vector<uint16_t> bool_states(1, state ? (0xFFFF & this->bitmask) : 0);
-        cmd.emplace(
-            ModbusCommandItem::create_write_multiple_command(this->parent_, this->write_address(), 1, bool_states));
-      } else {
-        cmd.emplace(ModbusCommandItem::create_write_single_command(this->parent_, this->write_address(),
-                                                                   state ? 0xFFFF & this->bitmask : 0u));
-      }
+      queued = this->write_single_coil(this->write_address(), state);
+    }
+  } else {
+    if (this->use_write_multiple_) {
+      std::array<uint16_t, 1> states{static_cast<uint16_t>(state ? (0xFFFF & this->bitmask) : 0)};
+      queued = this->write_multiple_registers(this->write_address(), states);
+    } else {
+      queued = this->write_single_register(this->write_address(), state ? 0xFFFF & this->bitmask : 0u);
     }
   }
-  this->parent_->queue_command(std::move(*cmd));
+  if (!queued) {
+    ESP_LOGW(TAG, "Modbus write for '%s' was refused by the hub; state not published", this->get_name().c_str());
+    return;
+  }
   this->publish_state(state);
 }
 // ModbusSwitch end
diff --git a/esphome/components/modbus_controller/switch/modbus_switch.h b/esphome/components/modbus_controller/switch/modbus_switch.h
index c21a1939bc0..1d3d03919fd 100644
--- a/esphome/components/modbus_controller/switch/modbus_switch.h
+++ b/esphome/components/modbus_controller/switch/modbus_switch.h
@@ -8,7 +8,7 @@
 
 namespace esphome::modbus_controller {
 
-class ModbusSwitch final : public Component, public switch_::Switch, public SensorItem {
+class ModbusSwitch final : public Component, public switch_::Switch, public SensorItem, public WriterEntity {
  public:
   ModbusSwitch(modbus::EntityType register_type, uint16_t start_address, uint8_t offset, uint32_t bitmask,
                bool force_new_range) {
@@ -30,17 +30,16 @@ class ModbusSwitch final : public Component, public switch_::Switch, public Sens
   void set_assumed_state(bool assumed_state);
   void set_state(bool state) { this->state = state; }
   void parse_and_publish(std::span<const uint8_t> data) override;
-  void set_parent(ModbusController *parent) { this->parent_ = parent; }
+  void set_parent(ModbusController *parent) { this->set_controller_(parent); }
 
   using transform_func_t = optional<bool> (*)(ModbusSwitch *, bool, std::span<const uint8_t>);
-  using write_transform_func_t = optional<bool> (*)(ModbusSwitch *, bool, std::vector<uint8_t> &);
+  using write_transform_func_t = optional<bool> (*)(ModbusSwitch *, bool, modbus::helpers::PduBuffer &);
   void set_template(transform_func_t f) { this->publish_transform_func_ = f; }
   void set_write_template(write_transform_func_t f) { this->write_transform_func_ = f; }
   void set_use_write_mutiple(bool use_write_multiple) { this->use_write_multiple_ = use_write_multiple; }
 
  protected:
   bool assumed_state() override;
-  ModbusController *parent_{nullptr};
   bool use_write_multiple_{false};
   optional<transform_func_t> publish_transform_func_{nullopt};
   optional<write_transform_func_t> write_transform_func_{nullopt};
diff --git a/tests/components/modbus_controller/command_payload_test.cpp b/tests/components/modbus_controller/command_payload_test.cpp
index c125a44da59..a0a59f51067 100644
--- a/tests/components/modbus_controller/command_payload_test.cpp
+++ b/tests/components/modbus_controller/command_payload_test.cpp
@@ -13,7 +13,7 @@ namespace esphome::modbus_controller::testing {
 // malformed. Built at its true byte count, the oversize frame is refused by the hub's size check with
 // a log instead.
 TEST(ModbusCommandPayload, CoilWritePayloadIsExactSizedNotTruncated) {
-  ModbusController controller;
+  ModbusController controller(nullptr, 1);
   std::vector<bool> coils(modbus::MAX_NUM_OF_COILS_TO_WRITE + 1, true);
   auto cmd = ModbusCommandItem::create_write_multiple_coils(&controller, 0x10, coils);
   EXPECT_EQ(cmd.payload.size(), modbus::packed_bit_bytes(coils.size()));
@@ -21,7 +21,7 @@ TEST(ModbusCommandPayload, CoilWritePayloadIsExactSizedNotTruncated) {
 
 // LSB-first packing with zeroed pad bits, matching the wire layout the PDU builders produce.
 TEST(ModbusCommandPayload, CoilWritePacksLsbFirstWithZeroPad) {
-  ModbusController controller;
+  ModbusController controller(nullptr, 1);
   const std::vector<bool> coils{true, false, true, true};
   auto cmd = ModbusCommandItem::create_write_multiple_coils(&controller, 0x10, coils);
   ASSERT_EQ(cmd.payload.size(), 1u);
diff --git a/tests/integration/fixtures/uart_mock_modbus_lambda_write.yaml b/tests/integration/fixtures/uart_mock_modbus_lambda_write.yaml
new file mode 100644
index 00000000000..86e17ea0d73
--- /dev/null
+++ b/tests/integration/fixtures/uart_mock_modbus_lambda_write.yaml
@@ -0,0 +1,97 @@
+esphome:
+  name: uart-mock-modbus-lambda-write
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: reg30
+    type: uint16_t
+    initial_value: "0"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_controller
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_controller
+    id: modbus_controller_1
+    update_interval: 1s
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    id: modbus_server_1
+    registers:
+      - address: 0x30
+        value_type: U_WORD
+        read_lambda: return id(reg30);
+        write_lambda: id(reg30) = x; return true;
+
+# A COIL-type switch (assumed_state, write-only) whose write_lambda ignores its own coil type and instead
+# drives a HOLDING-REGISTER write on the mock server through the entity itself: `item` IS the command, so
+# item->write_single_register() sends a register write from a coil entity (cross-type). Returning nothing
+# (an empty optional) tells the write path the lambda already dispatched the frame - no default coil write.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "cross_switch"
+    register_type: coil
+    address: 0x00
+    assumed_state: true
+    write_lambda: |-
+      item->write_single_register(0x30, x ? 1234 : 0);
+      return {};
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_30"
+    address: 0x30
+    register_type: holding
+    value_type: U_WORD
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)
diff --git a/tests/integration/test_uart_mock_modbus.py b/tests/integration/test_uart_mock_modbus.py
index 707637cfc28..3dfeda9b37e 100644
--- a/tests/integration/test_uart_mock_modbus.py
+++ b/tests/integration/test_uart_mock_modbus.py
@@ -969,10 +969,10 @@ async def test_uart_mock_modbus_client_read_write(
 
 @pytest.mark.xfail(
     strict=True,
-    reason="Byte-accurate register-offset writes require the modbus_controller "
-    "entity-device change; on dev the byte offset is folded into the address "
-    "(writes 0x12 instead of 0x11). The write and read assertions both flip via "
-    "the same switch-constructor fold. Remove this marker when that change merges.",
+    reason="Byte-accurate register-offset writes land in the follow-up offset fix; "
+    "until then the byte offset is folded into the address (writes 0x12 instead of "
+    "0x11). The write and read assertions both flip via the same switch-constructor "
+    "fold. Remove this marker when that change merges.",
 )
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_register_offset(
@@ -1029,14 +1029,42 @@ async def test_uart_mock_modbus_register_offset(
         )
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="The deprecated write buffer requires the modbus_controller "
-    "entity-device change; on dev a nullopt-returning write_lambda early-returns "
-    "before the buffer is used, so the write never happens. The warn-once "
-    "assertion matches the log substring 'write_lambda buffer'. Remove this "
-    "marker when that change merges.",
-)
+@pytest.mark.asyncio
+async def test_uart_mock_modbus_lambda_write(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test a write_lambda that drives the write through the entity itself (item is the command).
+
+    `cross_switch` is a coil-type switch whose write_lambda ignores its own type and calls
+    item->write_single_register(0x30, ...) - a register write issued from a coil entity. The lambda
+    returns an empty optional, so the write path detects the lambda already dispatched a frame and does
+    not fall back to the default coil write. Success is reg_30 reading back the value the lambda wrote,
+    which proves both the new item->write_* path and cross-type flexibility.
+    """
+
+    tracker = SensorTracker(["reg_30"])
+    initial = tracker.expect("reg_30", 0)
+    wrote_30 = tracker.expect("reg_30", 1234)
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        entities = await tracker.setup_and_start_scenario(client)
+        await tracker.await_change(initial, "reg_30", timeout=4.0)
+
+        switch = find_entity(entities, "cross_switch", SwitchInfo)
+        assert switch is not None, "cross_switch not found"
+        client.switch_command(switch.key, True)
+
+        # The coil switch's lambda wrote register 0x30 via item->write_single_register(); reg_30 must
+        # read back 1234. If the entity-as-command dispatch were broken, no register write would go out
+        # and this would time out.
+        await tracker.await_change(wrote_30, "reg_30", timeout=4.0)
+
+
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_deprecated_write_buffer(
     yaml_config: str,
@@ -1046,9 +1074,9 @@ async def test_uart_mock_modbus_deprecated_write_buffer(
     """Test the deprecated write_lambda buffer path still works, and warns once per entity.
 
     buf_number's write_lambda fills the old `payload` buffer with a legacy raw frame as words (device
-    address + function code + data) instead of calling item->write_*. Two writes must both land with the
-    legacy raw-frame semantics, and the one-time deprecation warning must fire exactly once per entity
-    regardless of how many writes happen.
+    address + function code + data) and returns {} instead of calling item->write_*. Both writes must
+    land - a filled buffer is sent, as the docs have always described - and the one-time deprecation
+    warning must fire exactly once per entity regardless of how many writes happen.
     """
 
     warn_count = 0