Merge remote-tracking branch 'upstream/dev' into noise-session-resume

esphome/components/api/api.proto

@@ -893,6 +893,20 @@ message NoiseEncryptionSetKeyResponse {
   bool success = 1;
 }
 
+// Single-use session resume ticket, sent unsolicited by the device after a
+// Noise connection authenticates. A client presents it in the ClientHello of
+// its next connection to skip the curve25519 handshake; the device then
+// issues a fresh ticket on that connection. Never sent on plaintext
+// connections. Clients that do not understand it drop it silently.
+// Contents are secret; the device generator redacts this message from dump_to
+message NoiseResumeTicket {
+  option (id) = 152;
+  option (source) = SOURCE_SERVER;
+  option (ifdef) = "USE_API_NOISE";
+
+  bytes ticket = 1;  // session_id(8) || secret(32)
+}
+
 // ==================== HOMEASSISTANT.SERVICE ====================
 message SubscribeHomeassistantServicesRequest {
   option (id) = 34;

esphome/components/api/api_connection.cpp

@@ -1779,6 +1779,9 @@ void APIConnection::complete_authentication_() {
     this->send_time_request();
   }
 #endif
+#ifdef USE_API_NOISE
+  this->send_resume_ticket_();
+#endif
 #ifdef USE_ZWAVE_PROXY
   if (zwave_proxy::global_zwave_proxy != nullptr) {
     zwave_proxy::global_zwave_proxy->api_connection_authenticated(this);
@@ -1786,6 +1789,27 @@ void APIConnection::complete_authentication_() {
 #endif
 }
 
+#ifdef USE_API_NOISE
+void APIConnection::send_resume_ticket_() {
+#ifdef USE_API_PLAINTEXT
+  // Only encrypted transports get a ticket: on dual-mode builds a plaintext
+  // connection has no frame footer
+  if (this->helper_->frame_footer_size() == 0) {
+    return;
+  }
+#endif
+  noise::ResumeTicket ticket;
+  if (!this->parent_->get_noise_ctx().resume_cache().issue(ticket)) {
+    return;
+  }
+  NoiseResumeTicket msg;
+  msg.set_ticket(reinterpret_cast<const uint8_t *>(&ticket), sizeof(ticket));
+  // A dropped ticket is harmless: the client does a full handshake next time
+  static_cast<void>(this->send_message(msg));
+  noise_clean(&ticket, sizeof(ticket));
+}
+#endif
+
 bool APIConnection::send_hello_response_(const HelloRequest &msg) {
   // Copy client name with truncation if needed (set_client_name handles truncation)
   this->helper_->set_client_name(msg.client_info.c_str(), msg.client_info.size());

esphome/components/api/api_connection.h

@@ -381,6 +381,11 @@ class APIConnection final : public APIServerConnectionBase {
   // Helper function to handle authentication completion
   void complete_authentication_();
 
+#ifdef USE_API_NOISE
+  // Issue a fresh single-use session resume ticket over the encrypted channel
+  void send_resume_ticket_();
+#endif
+
   // Pattern B helpers: send response and return success/failure
   bool send_hello_response_(const HelloRequest &msg);
   bool send_disconnect_response_();

esphome/components/api/api_frame_helper_noise.cpp

@@ -271,8 +271,8 @@ APIError APINoiseFrameHelper::state_action_client_hello_() {
   if (aerr != APIError::OK) {
     return handle_handshake_frame_error_(aerr);
   }
-  // ignore contents, may be used in future for flags
-  // Resize for: existing prologue + 2 size bytes + frame data
+  // Contents are extension flags (today: the resume offer); mixed into the
+  // prologue either way. Resize for: existing prologue + 2 size bytes + frame data
   size_t old_size = this->prologue_.size();
   size_t rx_size = this->rx_buf_.size();
   if (!this->prologue_.resize(old_size + 2 + rx_size)) [[unlikely]] {
@@ -289,6 +289,8 @@ APIError APINoiseFrameHelper::state_action_client_hello_() {
   return APIError::OK;
 }
 APIError APINoiseFrameHelper::state_action_server_hello_() {
+  // A verified resume offer (still in rx_buf_ from the client hello step)
+  // replaces the whole handshake; any failure falls back to the full one.
   // send server hello
   const auto &name = App.get_name();
   char mac[MAC_ADDRESS_BUFFER_SIZE];
@@ -302,7 +304,9 @@ APIError APINoiseFrameHelper::state_action_server_hello_() {
 
   // 1 (proto) + name (max ESPHOME_DEVICE_NAME_MAX_LEN) + 1 (name null)
   // + mac (MAC_ADDRESS_BUFFER_SIZE - 1) + 1 (mac null)
-  constexpr size_t max_msg_size = 1 + ESPHOME_DEVICE_NAME_MAX_LEN + 1 + MAC_ADDRESS_BUFFER_SIZE;
+  // + optional resume accept extension
+  constexpr size_t max_msg_size =
+      1 + ESPHOME_DEVICE_NAME_MAX_LEN + 1 + MAC_ADDRESS_BUFFER_SIZE + noise::RESUME_ACCEPT_SIZE;
   uint8_t msg[max_msg_size];
 
   // chosen proto
@@ -313,16 +317,32 @@ APIError APINoiseFrameHelper::state_action_server_hello_() {
   // node mac, terminated by null byte
   std::memcpy(msg + mac_offset, mac, MAC_ADDRESS_BUFFER_SIZE);
 
+  // The accept extension, if any, is written straight after the mac
+  size_t ext_len = this->ctx_.resume_cache().try_accept(
+      this->rx_buf_.data(), this->rx_buf_.size(), this->prologue_.data(), this->prologue_.size(), msg + total_size,
+      sizeof(msg) - total_size, send_cipher_, recv_cipher_);
+  bool resume = ext_len != 0;
+  total_size += ext_len;
+
   APIError aerr = write_frame_(msg, total_size);
   if (aerr != APIError::OK)
     return aerr;
 
-  // start handshake
-  aerr = init_handshake_();
-  if (aerr != APIError::OK)
-    return aerr;
-
-  state_ = State::HANDSHAKE;
+  if (resume) {
+    // A resuming client waits for this hello instead of pipelining
+    // handshake message 1, so the transport is ready now
+    this->frame_footer_size_ = noise_cipherstate_get_mac_length(this->send_cipher_);
+    HELPER_LOG("Session resumed!");
+    state_ = State::DATA;
+  } else {
+    aerr = init_handshake_();
+    if (aerr != APIError::OK)
+      return aerr;
+    state_ = State::HANDSHAKE;
+  }
+  // init_handshake_ copied the prologue into the handshake state; the resume
+  // path is done with it too
+  this->prologue_.release();
   return APIError::OK;
 }
 APIError APINoiseFrameHelper::state_action_handshake_() {
@@ -552,8 +572,6 @@ APIError APINoiseFrameHelper::init_handshake_() {
   APIError aerr = handle_noise_error_(err, LOG_STR("noise_handshake_init"), APIError::HANDSHAKESTATE_SETUP_FAILED);
   if (aerr != APIError::OK)
     return aerr;
-  // init copies the prologue into the handshakestate, so we can get rid of it now
-  prologue_.release();
   return APIError::OK;
 }
 

esphome/components/api/api_pb2.cpp

@@ -1061,6 +1061,16 @@ uint32_t NoiseEncryptionSetKeyResponse::calculate_size() const {
   size += ProtoSize::calc_bool(1, this->success);
   return size;
 }
+uint8_t *NoiseResumeTicket::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
+  uint8_t *__restrict__ pos = buffer.get_pos();
+  ProtoEncode::encode_bytes(pos PROTO_ENCODE_DEBUG_ARG, 1, this->ticket_ptr_, this->ticket_len_);
+  return pos;
+}
+uint32_t NoiseResumeTicket::calculate_size() const {
+  uint32_t size = 0;
+  size += ProtoSize::calc_length(1, this->ticket_len_);
+  return size;
+}
 #endif
 #ifdef USE_API_HOMEASSISTANT_SERVICES
 uint8_t *HomeassistantServiceMap::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {

esphome/components/api/api_pb2.h

@@ -1147,6 +1147,27 @@ class NoiseEncryptionSetKeyResponse final : public ProtoMessage {
 
  protected:
 };
+class NoiseResumeTicket final : public ProtoMessage {
+ public:
+  static constexpr uint16_t MESSAGE_TYPE = 152;
+  static constexpr uint8_t ESTIMATED_SIZE = 19;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const LogString *message_name() const override { return LOG_STR("noise_resume_ticket"); }
+#endif
+  const uint8_t *ticket_ptr_{nullptr};
+  size_t ticket_len_{0};
+  void set_ticket(const uint8_t *data, size_t len) {
+    this->ticket_ptr_ = data;
+    this->ticket_len_ = len;
+  }
+  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
+  uint32_t calculate_size() const;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const char *dump_to(DumpBuffer &out) const override;
+#endif
+
+ protected:
+};
 #endif
 #ifdef USE_API_HOMEASSISTANT_SERVICES
 class HomeassistantServiceMap final : public ProtoMessage {

esphome/components/api/api_pb2_dump.cpp

@@ -1393,6 +1393,10 @@ const char *NoiseEncryptionSetKeyResponse::dump_to(DumpBuffer &out) const {
   dump_field(out, ESPHOME_PSTR("success"), this->success);
   return out.c_str();
 }
+const char *NoiseResumeTicket::dump_to(DumpBuffer &out) const {
+  out.append_p(ESPHOME_PSTR("NoiseResumeTicket {}"));
+  return out.c_str();
+}
 #endif
 #ifdef USE_API_HOMEASSISTANT_SERVICES
 const char *HomeassistantServiceMap::dump_to(DumpBuffer &out) const {

esphome/components/noise/noise.h

@@ -6,6 +6,8 @@
 #include <cstdint>
 #include "esphome/core/log.h"
 
+#include "noise_resume.h"
+
 namespace esphome::noise {
 
 using psk_t = std::array<uint8_t, 32>;
@@ -26,12 +28,16 @@ class NoiseContext {
   void set_psk(psk_t psk) {
     this->psk_ = psk;
     this->has_psk_ = !is_all_zeros(psk);
+    // Resume tickets were minted under the old key; forget them
+    this->resume_cache_.clear();
   }
   const psk_t &get_psk() const { return this->psk_; }
   bool has_psk() const { return this->has_psk_; }
+  ResumeTicketCache &resume_cache() { return this->resume_cache_; }
 
  protected:
   psk_t psk_{};
+  ResumeTicketCache resume_cache_;
   bool has_psk_{false};
 };
 

esphome/components/noise/noise_resume.cpp

@@ -0,0 +1,127 @@
+#include "noise_resume.h"
+#ifdef USE_NOISE
+#include <cstring>
+
+#include <noise/protocol.h>
+
+#include "esphome/core/hal.h"
+#include "esphome/core/helpers.h"
+
+namespace esphome::noise {
+
+const char RESUME_LABEL_OFFER[6] PROGMEM = "offer";
+const char RESUME_LABEL_CONFIRM[8] PROGMEM = "confirm";
+const char RESUME_LABEL_KEYS[5] PROGMEM = "keys";
+bool resume_kdf(const uint8_t *secret, const char *label, size_t label_len, const uint8_t *a, size_t a_len,
+                const uint8_t *b, size_t b_len, const uint8_t *hash_in, size_t hash_in_len, uint8_t *out1,
+                size_t out1_len, uint8_t *out2) {
+  uint8_t data[RESUME_KDF_MAX_DATA];
+  uint8_t scratch[32];
+  size_t len = label_len + a_len + b_len;
+  progmem_memcpy(data, label, label_len);
+  std::memcpy(data + label_len, a, a_len);
+  std::memcpy(data + label_len + a_len, b, b_len);
+  NoiseHashState *hash = nullptr;
+  if (noise_hashstate_new_by_id(&hash, NOISE_HASH_SHA256) != NOISE_ERROR_NONE) {
+    return false;
+  }
+  int err = NOISE_ERROR_NONE;
+  if (hash_in != nullptr) {
+    err = noise_hashstate_hash_one(hash, hash_in, hash_in_len, data + len, 32);
+    len += 32;
+  }
+  if (err == NOISE_ERROR_NONE) {
+    err = noise_hashstate_hkdf(hash, secret, RESUME_SECRET_SIZE, data, len, out1, out1_len,
+                               out2 != nullptr ? out2 : scratch, 32);
+  }
+  noise_hashstate_free(hash);
+  noise_clean(data, sizeof(data));
+  noise_clean(scratch, sizeof(scratch));
+  return err == NOISE_ERROR_NONE;
+}
+
+bool ResumeTicketCache::issue(ResumeTicket &out) {
+  if (!random_bytes(reinterpret_cast<uint8_t *>(&out), sizeof(out))) {
+    return false;
+  }
+  uint8_t slot = this->next_;
+  this->next_ = static_cast<uint8_t>((slot + 1) % SLOTS);
+  this->slots_[slot] = out;
+  this->used_mask_ |= static_cast<uint8_t>(1u << slot);
+  return true;
+}
+
+size_t ResumeTicketCache::try_accept(const uint8_t *offer, size_t offer_len, const uint8_t *prologue,
+                                     size_t prologue_len, uint8_t *out_ext, size_t out_capacity,
+                                     NoiseCipherState *&send_cipher, NoiseCipherState *&recv_cipher) {
+  if (offer_len != RESUME_OFFER_SIZE || offer[0] != RESUME_OFFER_VERSION || out_capacity < RESUME_ACCEPT_SIZE) {
+    return 0;
+  }
+  const uint8_t *session_id = offer + RESUME_OFFER_SESSION_ID_OFFSET;
+  const uint8_t *client_nonce = offer + RESUME_OFFER_NONCE_OFFSET;
+  ResumeTicket *ticket = nullptr;
+  for (uint8_t i = 0; i < SLOTS; i++) {
+    if ((this->used_mask_ & (1u << i)) &&
+        std::memcmp(this->slots_[i].session_id, session_id, RESUME_SESSION_ID_SIZE) == 0) {
+      ticket = &this->slots_[i];
+      this->used_mask_ &= static_cast<uint8_t>(~(1u << i));
+      break;
+    }
+  }
+  if (ticket == nullptr) {
+    return 0;
+  }
+  uint8_t expected[RESUME_MAC_SIZE];
+  bool ok = resume_compute_offer_mac(ticket->secret, session_id, client_nonce, expected) &&
+            noise_is_equal(expected, offer + RESUME_OFFER_MAC_OFFSET, RESUME_MAC_SIZE);
+  noise_clean(expected, sizeof(expected));
+  if (!ok) {
+    // Bad MAC: keep the ticket so a forger cannot burn it
+    this->used_mask_ |= static_cast<uint8_t>(1u << static_cast<uint8_t>(ticket - this->slots_));
+    return 0;
+  }
+  // The ticket is spent from here; any later failure falls back to the full
+  // handshake and the client gets a fresh one.
+  uint8_t *server_nonce = out_ext + 1;
+  uint8_t k_c2d[32];
+  uint8_t k_d2c[32];
+  out_ext[0] = RESUME_ACCEPT_VERSION;
+  ok = random_bytes(server_nonce, RESUME_NONCE_SIZE) &&
+       resume_compute_confirm_mac(ticket->secret, client_nonce, server_nonce, out_ext + 1 + RESUME_NONCE_SIZE) &&
+       resume_derive_keys(ticket->secret, client_nonce, server_nonce, prologue, prologue_len, k_c2d, k_d2c);
+  noise_clean(ticket, sizeof(*ticket));
+  if (ok) {
+    recv_cipher = resume_make_cipher(k_c2d);
+    send_cipher = resume_make_cipher(k_d2c);
+    ok = recv_cipher != nullptr && send_cipher != nullptr;
+    if (!ok) {
+      noise_cipherstate_free(recv_cipher);
+      noise_cipherstate_free(send_cipher);
+      recv_cipher = nullptr;
+      send_cipher = nullptr;
+    }
+  }
+  noise_clean(k_c2d, sizeof(k_c2d));
+  noise_clean(k_d2c, sizeof(k_d2c));
+  return ok ? RESUME_ACCEPT_SIZE : 0;
+}
+
+void ResumeTicketCache::clear() {
+  noise_clean(this->slots_, sizeof(this->slots_));
+  this->used_mask_ = 0;
+}
+
+NoiseCipherState *resume_make_cipher(const uint8_t *key) {
+  NoiseCipherState *cipher = nullptr;
+  if (noise_cipherstate_new_by_id(&cipher, NOISE_CIPHER_CHACHAPOLY) != NOISE_ERROR_NONE) {
+    return nullptr;
+  }
+  if (noise_cipherstate_init_key(cipher, key, 32) != NOISE_ERROR_NONE) {
+    noise_cipherstate_free(cipher);
+    return nullptr;
+  }
+  return cipher;
+}
+
+}  // namespace esphome::noise
+#endif  // USE_NOISE

esphome/components/noise/noise_resume.h

@@ -0,0 +1,132 @@
+#pragma once
+#include "esphome/core/defines.h"
+#ifdef USE_NOISE
+#include <cstddef>
+#include <cstdint>
+
+// Forward declaration matching <noise/protocol/cipherstate.h>; keeps noise-c
+// headers out of everything that includes noise.h.
+extern "C" {
+typedef struct NoiseCipherState_s NoiseCipherState;  // NOLINT(modernize-use-using)
+}
+
+namespace esphome::noise {
+
+/** Session resume for the noise transports.
+ *
+ * After a full handshake the responder issues a single-use ticket over the
+ * encrypted channel. A client presents it in its next ClientHello and both
+ * sides derive the transport keys with HKDF-SHA256 alone, skipping the two
+ * curve25519 operations. Old peers ignore the extension bytes on both
+ * sides, so every mismatch degrades to a normal full handshake.
+ *
+ * HKDF is the Noise construction (noise_hashstate_hkdf). Derivations:
+ *   offer_mac   = HKDF(secret, "offer"   || session_id || client_nonce).out1[:16]
+ *   confirm_mac = HKDF(secret, "confirm" || client_nonce || server_nonce).out1[:16]
+ *   k_c2d, k_d2c = HKDF(secret, "keys" || client_nonce || server_nonce || SHA256(prologue))
+ *
+ * An offering client sends handshake message 1 only after a decline.
+ * Resumed sessions have no ephemeral DH; the ticket is wiped on use.
+ */
+
+static constexpr uint8_t RESUME_OFFER_VERSION = 0x01;
+static constexpr uint8_t RESUME_ACCEPT_VERSION = 0x01;
+static constexpr size_t RESUME_SESSION_ID_SIZE = 8;
+static constexpr size_t RESUME_NONCE_SIZE = 16;
+static constexpr size_t RESUME_MAC_SIZE = 16;
+static constexpr size_t RESUME_SECRET_SIZE = 32;
+
+// ClientHello body: version | session_id | client_nonce | offer_mac
+static constexpr size_t RESUME_OFFER_SIZE = 1 + RESUME_SESSION_ID_SIZE + RESUME_NONCE_SIZE + RESUME_MAC_SIZE;  // 41
+static constexpr size_t RESUME_OFFER_SESSION_ID_OFFSET = 1;
+static constexpr size_t RESUME_OFFER_NONCE_OFFSET = RESUME_OFFER_SESSION_ID_OFFSET + RESUME_SESSION_ID_SIZE;
+static constexpr size_t RESUME_OFFER_MAC_OFFSET = RESUME_OFFER_NONCE_OFFSET + RESUME_NONCE_SIZE;
+
+// ServerHello trailing extension: version | server_nonce | confirm_mac
+static constexpr size_t RESUME_ACCEPT_SIZE = 1 + RESUME_NONCE_SIZE + RESUME_MAC_SIZE;  // 33
+
+struct ResumeTicket {
+  uint8_t session_id[RESUME_SESSION_ID_SIZE];
+  uint8_t secret[RESUME_SECRET_SIZE];
+};
+// Sent on the wire as one blob: session_id || secret
+static_assert(sizeof(ResumeTicket) == RESUME_SESSION_ID_SIZE + RESUME_SECRET_SIZE, "ticket must be packed");
+
+/// Fixed-slot RAM cache of single-use resume tickets. Lost on reboot by
+/// design: clients fall back to a full handshake.
+class ResumeTicketCache {
+ public:
+  /// Generate a fresh ticket into out and store it, evicting the oldest
+  /// slot. Returns false (and stores nothing) if the RNG fails.
+  bool issue(ResumeTicket &out);
+  /// Accept a resume offer: verify and consume the ticket (single use; a
+  /// forged MAC never burns one), build both transport ciphers, and write
+  /// the ServerHello accept extension into out_ext. Returns the extension
+  /// length, or 0 (nothing allocated) on any miss, failure, or when
+  /// out_capacity is too small. Secrets are wiped internally.
+  size_t try_accept(const uint8_t *offer, size_t offer_len, const uint8_t *prologue, size_t prologue_len,
+                    uint8_t *out_ext, size_t out_capacity, NoiseCipherState *&send_cipher,
+                    NoiseCipherState *&recv_cipher);
+  /// Forget every ticket (PSK change).
+  void clear();
+
+  // Round robin; more clients than slots thrash and fall back to full handshakes
+  static constexpr uint8_t SLOTS = 2;
+  static_assert(SLOTS <= 8, "used_mask_ is uint8_t");
+
+ protected:
+  ResumeTicket slots_[SLOTS];
+  uint8_t used_mask_{0};
+  uint8_t next_{0};
+};
+
+/// HKDF labels, PROGMEM on ESP8266.
+extern const char RESUME_LABEL_OFFER[6];
+extern const char RESUME_LABEL_CONFIRM[8];
+extern const char RESUME_LABEL_KEYS[5];
+
+// Largest KDF input: "keys" || client_nonce || server_nonce || SHA256(prologue)
+static constexpr size_t RESUME_KDF_MAX_DATA =
+    sizeof(RESUME_LABEL_KEYS) - 1 + RESUME_NONCE_SIZE + RESUME_NONCE_SIZE + 32;
+
+/// Noise-construction HKDF-SHA256 keyed with the ticket secret over
+/// label || a || b [|| SHA256(hash_in)], at most RESUME_KDF_MAX_DATA. out2 == nullptr means MAC only.
+bool resume_kdf(const uint8_t *secret, const char *label, size_t label_len, const uint8_t *a, size_t a_len,
+                const uint8_t *b, size_t b_len, const uint8_t *hash_in, size_t hash_in_len, uint8_t *out1,
+                size_t out1_len, uint8_t *out2);
+
+/// offer_mac for the ClientHello resume offer (what a client computes and
+/// try_accept checks).
+inline bool resume_compute_offer_mac(const uint8_t *secret, const uint8_t *session_id, const uint8_t *client_nonce,
+                                     uint8_t *out_mac) {
+  static_assert(sizeof(RESUME_LABEL_OFFER) - 1 + RESUME_SESSION_ID_SIZE + RESUME_NONCE_SIZE <= RESUME_KDF_MAX_DATA,
+                "KDF buffer");
+  return resume_kdf(secret, RESUME_LABEL_OFFER, sizeof(RESUME_LABEL_OFFER) - 1, session_id, RESUME_SESSION_ID_SIZE,
+                    client_nonce, RESUME_NONCE_SIZE, nullptr, 0, out_mac, RESUME_MAC_SIZE, nullptr);
+}
+
+/// confirm_mac for the ServerHello extension.
+inline bool resume_compute_confirm_mac(const uint8_t *secret, const uint8_t *client_nonce, const uint8_t *server_nonce,
+                                       uint8_t *out_mac) {
+  static_assert(sizeof(RESUME_LABEL_CONFIRM) - 1 + RESUME_NONCE_SIZE + RESUME_NONCE_SIZE <= RESUME_KDF_MAX_DATA,
+                "KDF buffer");
+  return resume_kdf(secret, RESUME_LABEL_CONFIRM, sizeof(RESUME_LABEL_CONFIRM) - 1, client_nonce, RESUME_NONCE_SIZE,
+                    server_nonce, RESUME_NONCE_SIZE, nullptr, 0, out_mac, RESUME_MAC_SIZE, nullptr);
+}
+
+/// Derive the transport keys. k_c2d encrypts client-to-device traffic,
+/// k_d2c device-to-client.
+inline bool resume_derive_keys(const uint8_t *secret, const uint8_t *client_nonce, const uint8_t *server_nonce,
+                               const uint8_t *prologue, size_t prologue_len, uint8_t *k_c2d, uint8_t *k_d2c) {
+  static_assert(sizeof(RESUME_LABEL_KEYS) - 1 + RESUME_NONCE_SIZE + RESUME_NONCE_SIZE + 32 <= RESUME_KDF_MAX_DATA,
+                "KDF buffer");
+  return resume_kdf(secret, RESUME_LABEL_KEYS, sizeof(RESUME_LABEL_KEYS) - 1, client_nonce, RESUME_NONCE_SIZE,
+                    server_nonce, RESUME_NONCE_SIZE, prologue, prologue_len, k_c2d, 32, k_d2c);
+}
+
+/// Build a ChaChaPoly cipher state keyed with key (32 bytes); nullptr on
+/// failure. Nonce counter starts at 0, exactly like a post-split cipher.
+NoiseCipherState *resume_make_cipher(const uint8_t *key);
+
+}  // namespace esphome::noise
+#endif  // USE_NOISE

script/api_protobuf/api_protobuf.py

@@ -2531,6 +2531,11 @@ def calculate_message_max_size(desc: descriptor.DescriptorProto) -> int | None:
     return total_size
 
 
+# Contents must never reach the log: dump_to prints only the name
+SENSITIVE_MESSAGES = {"NoiseResumeTicket"}
+SENSITIVE_MESSAGES_SEEN: set[str] = set()
+
+
 def build_message_type(
     desc: descriptor.DescriptorProto,
     base_class_fields: dict[str, list[descriptor.FieldDescriptorProto]],
@@ -2808,6 +2813,10 @@ def build_message_type(
         public_content.append(prot)
     # If no fields to calculate size for or message doesn't need encoding, the default implementation in ProtoMessage will be used
 
+    if desc.name in SENSITIVE_MESSAGES:
+        dump = []
+        SENSITIVE_MESSAGES_SEEN.add(desc.name)
+
     # dump_to method declaration in header
     prot = "#ifdef HAS_PROTO_MESSAGE_DUMP\n"
     prot += "const char *dump_to(DumpBuffer &out) const override;\n"
@@ -3715,6 +3724,11 @@ static const char *const TAG = "api.service";
     except ImportError:
         pass
 
+    # A renamed message must fail the build, not silently start dumping secrets
+    missing = SENSITIVE_MESSAGES - SENSITIVE_MESSAGES_SEEN
+    if missing:
+        raise RuntimeError(f"SENSITIVE_MESSAGES not found in api.proto: {missing}")
+
 
 if __name__ == "__main__":
     sys.exit(main())

script/helpers.py

@@ -1104,10 +1104,6 @@ def get_components_per_integration_fixture() -> dict[str, set[str]]:
 
 
 _TEST_FUNC_RE = re.compile(r"async def (test_\w+)")
-# Any usage form (decorator, pytestmark assignment or list element); only
-# test_*.py files are scanned, so the marker docs elsewhere cannot false-hit
-_SHARED_YAML_USE_RE = re.compile(r"\bmark\.shared_yaml")
-_SHARED_YAML_ARG_RE = re.compile(r"\(\s*[\"'](\w+)[\"']\s*\)")
 
 
 @cache
@@ -1127,19 +1123,6 @@ def get_fixture_to_test_files() -> dict[str, frozenset[str]]:
         for func in _TEST_FUNC_RE.findall(content):
             base_name = func.replace("test_", "").partition("[")[0]
             result.setdefault(base_name, set()).add(rel_path)
-        # Shared fixtures are named by marker, not by a test function; each
-        # decorator must carry a string literal or its fixture would silently
-        # map to no tests
-        for use in _SHARED_YAML_USE_RE.finditer(content):
-            arg = _SHARED_YAML_ARG_RE.match(content, use.end())
-            if arg is None:
-                line = content.count("\n", 0, use.start()) + 1
-                raise ValueError(
-                    f"{rel_path}:{line}: shared_yaml marker must take a "
-                    "single-line string literal so CI test selection can map "
-                    "its fixture"
-                )
-            result.setdefault(arg.group(1), set()).add(rel_path)
 
     return {k: frozenset(v) for k, v in result.items()}
 

tests/components/noise/test_noise_resume.cpp

@@ -0,0 +1,224 @@
+#include <gtest/gtest.h>
+
+#include <cstring>
+
+#include <noise/protocol.h>
+
+#include "esphome/components/noise/noise.h"
+#include "esphome/components/noise/noise_resume.h"
+
+namespace esphome::noise::testing {
+
+// Known-answer vectors shared with the client implementation
+// (aioesphomeapi tests/test_noise_resume.py); the two must stay identical
+// byte for byte or resumed sessions cannot interoperate.
+static const uint8_t KAT_SECRET[RESUME_SECRET_SIZE] = {1,  2,  3,  4,  5,  6,  7,  8,  9,  10, 11, 12, 13, 14, 15, 16,
+                                                       17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32};
+static const uint8_t KAT_SESSION_ID[RESUME_SESSION_ID_SIZE] = {0xa0, 0xa1, 0xa2, 0xa3, 0xa4, 0xa5, 0xa6, 0xa7};
+static const uint8_t KAT_CLIENT_NONCE[RESUME_NONCE_SIZE] = {0x10, 0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17,
+                                                            0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f};
+static const uint8_t KAT_SERVER_NONCE[RESUME_NONCE_SIZE] = {0x30, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37,
+                                                            0x38, 0x39, 0x3a, 0x3b, 0x3c, 0x3d, 0x3e, 0x3f};
+static const uint8_t KAT_OFFER_MAC[RESUME_MAC_SIZE] = {0xa8, 0x08, 0xea, 0xdb, 0xec, 0x81, 0xa7, 0xcb,
+                                                       0xf4, 0xca, 0xaa, 0xb8, 0x0d, 0x7f, 0x9d, 0x01};
+static const uint8_t KAT_CONFIRM_MAC[RESUME_MAC_SIZE] = {0x09, 0xa3, 0x70, 0x3e, 0xc8, 0x34, 0x77, 0xe9,
+                                                         0x45, 0xe7, 0xf1, 0x61, 0x9d, 0x4f, 0x6a, 0x76};
+static const uint8_t KAT_K_C2D[32] = {0xd6, 0x01, 0xe3, 0xc1, 0x16, 0xa1, 0x64, 0x66, 0xdb, 0xc5, 0x9e,
+                                      0xdd, 0x60, 0x2a, 0x64, 0x1e, 0xbe, 0xf5, 0x11, 0x95, 0x98, 0xd2,
+                                      0xf2, 0x47, 0x1b, 0xc6, 0x8c, 0x51, 0x8f, 0xbe, 0xb7, 0x23};
+static const uint8_t KAT_K_D2C[32] = {0x7f, 0x8d, 0x57, 0x7e, 0x9f, 0xb4, 0xbb, 0xde, 0x86, 0xcd, 0xa9,
+                                      0xf4, 0x9b, 0x42, 0xe7, 0x24, 0xc8, 0x49, 0xce, 0x89, 0xd8, 0x96,
+                                      0x3f, 0x3c, 0x4b, 0x3f, 0x8f, 0x80, 0xc2, 0x56, 0xab, 0x65};
+
+/// The one place in this file that spells the offer wire layout
+static void build_offer(uint8_t *offer, const uint8_t *session_id, const uint8_t *client_nonce, const uint8_t *mac) {
+  offer[0] = RESUME_OFFER_VERSION;
+  std::memcpy(offer + RESUME_OFFER_SESSION_ID_OFFSET, session_id, RESUME_SESSION_ID_SIZE);
+  std::memcpy(offer + RESUME_OFFER_NONCE_OFFSET, client_nonce, RESUME_NONCE_SIZE);
+  std::memcpy(offer + RESUME_OFFER_MAC_OFFSET, mac, RESUME_MAC_SIZE);
+}
+
+/// "NoiseAPIInit" || be16(len) || offer, exactly as the api frame helper mixes it
+static constexpr size_t KAT_PROLOGUE_SIZE = 12 + 2 + RESUME_OFFER_SIZE;
+static void build_prologue(uint8_t *out, const uint8_t *offer) {
+  std::memcpy(out, "NoiseAPIInit", 12);  // NOLINT(bugprone-not-null-terminated-result)
+  out[12] = 0x00;
+  out[13] = RESUME_OFFER_SIZE;
+  std::memcpy(out + 14, offer, RESUME_OFFER_SIZE);
+}
+
+static void build_offer_for_ticket(uint8_t *offer, const ResumeTicket &ticket, const uint8_t *client_nonce) {
+  uint8_t mac[RESUME_MAC_SIZE];
+  ASSERT_TRUE(resume_compute_offer_mac(ticket.secret, ticket.session_id, client_nonce, mac));
+  build_offer(offer, ticket.session_id, client_nonce, mac);
+}
+
+/// Test access to the protected slots so a test can plant the KAT ticket
+struct TestCache : ResumeTicketCache {
+  void plant(const uint8_t *session_id, const uint8_t *secret) {
+    std::memcpy(this->slots_[0].session_id, session_id, RESUME_SESSION_ID_SIZE);
+    std::memcpy(this->slots_[0].secret, secret, RESUME_SECRET_SIZE);
+    this->used_mask_ |= 1u;
+  }
+};
+
+TEST(NoiseResumeKat, ConfirmMacMatchesClientImplementation) {
+  uint8_t mac[RESUME_MAC_SIZE];
+  ASSERT_TRUE(resume_compute_confirm_mac(KAT_SECRET, KAT_CLIENT_NONCE, KAT_SERVER_NONCE, mac));
+  EXPECT_EQ(std::memcmp(mac, KAT_CONFIRM_MAC, RESUME_MAC_SIZE), 0);
+}
+
+TEST(NoiseResumeKat, OfferMacMatchesClientImplementation) {
+  uint8_t mac[RESUME_MAC_SIZE];
+  ASSERT_TRUE(resume_compute_offer_mac(KAT_SECRET, KAT_SESSION_ID, KAT_CLIENT_NONCE, mac));
+  EXPECT_EQ(std::memcmp(mac, KAT_OFFER_MAC, RESUME_MAC_SIZE), 0);
+}
+
+TEST(NoiseResumeKat, KeyDerivationMatchesClientImplementation) {
+  // Prologue used by the shared vectors: "NoiseAPIInit" + be16(41) + a
+  // 41-byte offer whose MAC field is 16 bytes of 0xEE
+  uint8_t mac_filler[RESUME_MAC_SIZE];
+  std::memset(mac_filler, 0xEE, sizeof(mac_filler));
+  uint8_t offer[RESUME_OFFER_SIZE];
+  build_offer(offer, KAT_SESSION_ID, KAT_CLIENT_NONCE, mac_filler);
+  uint8_t prologue[KAT_PROLOGUE_SIZE];
+  build_prologue(prologue, offer);
+
+  uint8_t k_c2d[32], k_d2c[32];
+  ASSERT_TRUE(
+      resume_derive_keys(KAT_SECRET, KAT_CLIENT_NONCE, KAT_SERVER_NONCE, prologue, sizeof(prologue), k_c2d, k_d2c));
+  EXPECT_EQ(std::memcmp(k_c2d, KAT_K_C2D, 32), 0);
+  EXPECT_EQ(std::memcmp(k_d2c, KAT_K_D2C, 32), 0);
+}
+
+TEST(NoiseResumeCache, TryAcceptConsumesTicketOnceAndProvesPossession) {
+  TestCache cache;
+  cache.plant(KAT_SESSION_ID, KAT_SECRET);
+
+  uint8_t offer[RESUME_OFFER_SIZE];
+  build_offer(offer, KAT_SESSION_ID, KAT_CLIENT_NONCE, KAT_OFFER_MAC);
+  uint8_t prologue[KAT_PROLOGUE_SIZE];
+  build_prologue(prologue, offer);
+
+  uint8_t ext[RESUME_ACCEPT_SIZE];
+  NoiseCipherState *send = nullptr, *recv = nullptr;
+  ASSERT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send, recv),
+            RESUME_ACCEPT_SIZE);
+  ASSERT_NE(send, nullptr);
+  ASSERT_NE(recv, nullptr);
+
+  // The extension proves possession: verify like the client does
+  EXPECT_EQ(ext[0], RESUME_ACCEPT_VERSION);
+  const uint8_t *server_nonce = ext + 1;
+  uint8_t expected_confirm[RESUME_MAC_SIZE];
+  ASSERT_TRUE(resume_compute_confirm_mac(KAT_SECRET, KAT_CLIENT_NONCE, server_nonce, expected_confirm));
+  EXPECT_EQ(std::memcmp(ext + 1 + RESUME_NONCE_SIZE, expected_confirm, RESUME_MAC_SIZE), 0);
+
+  // The ciphers must interoperate with the documented key derivation
+  uint8_t k_c2d[32], k_d2c[32];
+  ASSERT_TRUE(resume_derive_keys(KAT_SECRET, KAT_CLIENT_NONCE, server_nonce, prologue, sizeof(prologue), k_c2d, k_d2c));
+  NoiseCipherState *client_send = resume_make_cipher(k_c2d);
+  ASSERT_NE(client_send, nullptr);
+  uint8_t buf[64] = "resumed";
+  NoiseBuffer nb;
+  noise_buffer_init(nb);
+  noise_buffer_set_inout(nb, buf, 7, sizeof(buf));
+  ASSERT_EQ(noise_cipherstate_encrypt(client_send, &nb), NOISE_ERROR_NONE);
+  ASSERT_EQ(noise_cipherstate_decrypt(recv, &nb), NOISE_ERROR_NONE);
+  EXPECT_EQ(std::memcmp(buf, "resumed", 7), 0);
+  noise_cipherstate_free(client_send);
+  noise_cipherstate_free(send);
+  noise_cipherstate_free(recv);
+
+  // Single use: the same offer must miss the second time
+  NoiseCipherState *send2 = nullptr, *recv2 = nullptr;
+  EXPECT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send2, recv2), 0u);
+  EXPECT_EQ(send2, nullptr);
+  EXPECT_EQ(recv2, nullptr);
+}
+
+TEST(NoiseResumeCache, BadMacOrMalformedOfferLeavesTicketIntact) {
+  TestCache cache;
+  cache.plant(KAT_SESSION_ID, KAT_SECRET);
+
+  uint8_t offer[RESUME_OFFER_SIZE];
+  uint8_t bad_mac[RESUME_MAC_SIZE];
+  std::memcpy(bad_mac, KAT_OFFER_MAC, RESUME_MAC_SIZE);
+  bad_mac[0] ^= 0x01;
+  build_offer(offer, KAT_SESSION_ID, KAT_CLIENT_NONCE, bad_mac);
+
+  uint8_t prologue[1] = {0};
+  uint8_t ext[RESUME_ACCEPT_SIZE];
+  NoiseCipherState *send = nullptr, *recv = nullptr;
+  // A forged offer must not burn the ticket
+  EXPECT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send, recv), 0u);
+  // Wrong size or version must be recognized as "no offer"
+  build_offer(offer, KAT_SESSION_ID, KAT_CLIENT_NONCE, KAT_OFFER_MAC);
+  EXPECT_EQ(cache.try_accept(offer, sizeof(offer) - 1, prologue, sizeof(prologue), ext, sizeof(ext), send, recv), 0u);
+  offer[0] = 0x7f;
+  EXPECT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send, recv), 0u);
+  offer[0] = RESUME_OFFER_VERSION;
+  // No room for the extension must also decline without burning it
+  EXPECT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext) - 1, send, recv), 0u);
+  // The genuine offer still redeems
+  EXPECT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send, recv),
+            RESUME_ACCEPT_SIZE);
+  noise_cipherstate_free(send);
+  noise_cipherstate_free(recv);
+}
+
+TEST(NoiseResumeCache, SetPskForgetsTickets) {
+  NoiseContext ctx;
+  ResumeTicket ticket;
+  ASSERT_TRUE(ctx.resume_cache().issue(ticket));
+
+  psk_t psk{};
+  psk[0] = 1;
+  ctx.set_psk(psk);
+
+  uint8_t offer[RESUME_OFFER_SIZE];
+  build_offer_for_ticket(offer, ticket, KAT_CLIENT_NONCE);
+  uint8_t prologue[KAT_PROLOGUE_SIZE];
+  build_prologue(prologue, offer);
+  uint8_t ext[RESUME_ACCEPT_SIZE];
+  NoiseCipherState *send = nullptr, *recv = nullptr;
+  EXPECT_EQ(
+      ctx.resume_cache().try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send, recv),
+      0u);
+  EXPECT_EQ(send, nullptr);
+  EXPECT_EQ(recv, nullptr);
+}
+
+TEST(NoiseResumeCache, IssueRotatesSlotsAndClearForgetsAll) {
+  ResumeTicketCache cache;
+  ResumeTicket tickets[ResumeTicketCache::SLOTS + 1];
+  for (auto &ticket : tickets) {
+    ASSERT_TRUE(cache.issue(ticket));
+  }
+  uint8_t offer[RESUME_OFFER_SIZE];
+  uint8_t prologue[1] = {0};
+  uint8_t ext[RESUME_ACCEPT_SIZE];
+
+  // The oldest ticket was evicted by the one-past-capacity issue
+  build_offer_for_ticket(offer, tickets[0], KAT_CLIENT_NONCE);
+  NoiseCipherState *send = nullptr, *recv = nullptr;
+  EXPECT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send, recv), 0u);
+  // The rest remain redeemable
+  for (int i = 1; i <= ResumeTicketCache::SLOTS; i++) {
+    build_offer_for_ticket(offer, tickets[i], KAT_CLIENT_NONCE);
+    EXPECT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send, recv),
+              RESUME_ACCEPT_SIZE);
+    noise_cipherstate_free(send);
+    noise_cipherstate_free(recv);
+    send = recv = nullptr;
+  }
+
+  // clear() forgets everything
+  ResumeTicket ticket;
+  ASSERT_TRUE(cache.issue(ticket));
+  cache.clear();
+  build_offer_for_ticket(offer, ticket, KAT_CLIENT_NONCE);
+  EXPECT_EQ(cache.try_accept(offer, sizeof(offer), prologue, sizeof(prologue), ext, sizeof(ext), send, recv), 0u);
+}
+
+}  // namespace esphome::noise::testing

tests/integration/README.md

@@ -21,13 +21,6 @@ The `yaml_config` fixture automatically loads YAML configurations based on the t
 - The fixture file must exist or the test will fail with a clear error message
 - The fixture automatically injects a dynamic port number into the API configuration
 
-Tests marked `@pytest.mark.shared_yaml("name")` load `fixtures/name.yaml` instead
-of the test-named file and compile it in a shared, hash-keyed build directory, so
-the whole group pays one full compile and each test only a relink. The marker
-argument must be a single-line string literal (CI test selection maps fixtures to
-test files by scanning for it), and marked tests must hand the `yaml_config`
-content to `run_compiled` unmodified.
-
 ### Key Fixtures
 
 - `run_compiled` - Combines write, compile, and run operations into a single context manager

tests/integration/conftest.py

@@ -4,22 +4,17 @@ from __future__ import annotations
 
 import asyncio
 from collections.abc import AsyncGenerator, Callable, Generator
-from contextlib import AbstractAsyncContextManager, asynccontextmanager, suppress
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
 import fcntl
-from functools import cache
-import hashlib
 import logging
 import os
 from pathlib import Path
 import platform
-import re
-import shutil
 import signal
 import socket
 import subprocess
 import sys
 import tempfile
-import time
 from typing import TextIO
 
 from aioesphomeapi import APIClient, APIConnectionError, LogParser, ReconnectLogic
@@ -28,13 +23,7 @@ import pytest_asyncio
 
 import esphome.config
 from esphome.core import CORE
-from esphome.helpers import (
-    get_usable_cpu_count,
-    read_file,
-    rmtree,
-    write_file,
-    write_file_if_changed,
-)
+from esphome.helpers import get_usable_cpu_count
 from esphome.platformio.toolchain import get_idedata
 
 from .const import (
@@ -67,21 +56,6 @@ import pty  # not available on Windows
 pytest.register_assert_rewrite("tests.integration.entity_utils")
 
 
-def pytest_configure(config: pytest.Config) -> None:
-    config.addinivalue_line(
-        "markers",
-        "shared_yaml(name): load fixtures/<name>.yaml and compile it in a shared, "
-        "hash-keyed incremental build directory",
-    )
-
-
-FIXTURES_DIR = Path(__file__).parent / "fixtures"
-REPO_ROOT = Path(__file__).resolve().parent.parent.parent
-
-# CI caches parts of this path; keep in sync with ci.yml integration-tests.
-INTEGRATION_TESTS_ROOT = Path.home() / ".esphome-integration-tests"
-
-
 def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
     """Get environment variables for PlatformIO with shared cache."""
     env = os.environ.copy()
@@ -104,7 +78,7 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
         )
     # Compile with THIS tree's esphome sources, not wherever the venv's editable
     # install points (which may be a different git worktree or checkout).
-    repo_root = str(REPO_ROOT)
+    repo_root = str(Path(__file__).resolve().parent.parent.parent)
     existing = env.get("PYTHONPATH")
     env["PYTHONPATH"] = f"{repo_root}{os.pathsep}{existing}" if existing else repo_root
     return env
@@ -114,7 +88,8 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
 def shared_platformio_cache() -> Generator[Path]:
     """Initialize a shared PlatformIO cache for all integration tests."""
     # Use a dedicated directory for integration tests to avoid conflicts.
-    test_cache_dir = INTEGRATION_TESTS_ROOT
+    # CI caches parts of this path; keep in sync with ci.yml integration-tests.
+    test_cache_dir = Path.home() / ".esphome-integration-tests"
     cache_dir = test_cache_dir / "platformio"
 
     # Use a lock file in the home directory to ensure only one process initializes the cache
@@ -137,9 +112,7 @@ def shared_platformio_cache() -> Generator[Path]:
                 init_dir = Path(tmpdir)
                 fixture_path = Path(__file__).parent / "fixtures" / "cache_init.yaml"
                 config_path = init_dir / "cache_init.yaml"
-                config_path.write_text(
-                    fixture_path.read_text(encoding="utf-8"), encoding="utf-8"
-                )
+                config_path.write_text(fixture_path.read_text())
 
                 # Run compilation to populate the cache
                 # We must succeed here to avoid race conditions where multiple
@@ -208,29 +181,21 @@ def unused_tcp_port(reserved_tcp_port: tuple[int, socket.socket]) -> int:
     return reserved_tcp_port[0]
 
 
-@pytest.fixture(autouse=True)
-def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> Path:
-    """Give every test its own host prefs dir; prefs are keyed only by device
-    name, which tests sharing a fixture also share."""
-    prefdir = tmp_path / "prefs"
-    monkeypatch.setenv("ESPHOME_PREFDIR", str(prefdir))
-    return prefdir
-
-
 @pytest_asyncio.fixture
 async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> str:
     """Load YAML configuration based on test name."""
-    shared_name = _shared_yaml_name(request)
-    # Base test name: test_ prefix and any parametrization stripped
-    base_name = shared_name or request.node.name.replace("test_", "").partition("[")[0]
+    # Get the test function name
+    test_name: str = request.node.name
+    # Extract the base test name (remove test_ prefix and any parametrization)
+    base_name = test_name.replace("test_", "").partition("[")[0]
 
     # Load the fixture file
-    fixture_path = FIXTURES_DIR / f"{base_name}.yaml"
+    fixture_path = Path(__file__).parent / "fixtures" / f"{base_name}.yaml"
     if not fixture_path.exists():
         raise FileNotFoundError(f"Fixture file not found: {fixture_path}")
 
     loop = asyncio.get_running_loop()
-    content = await loop.run_in_executor(None, read_file, fixture_path)
+    content = await loop.run_in_executor(None, fixture_path.read_text)
 
     # Replace the port in the config if it contains api section
     if "api:" in content:
@@ -254,13 +219,11 @@ async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> s
 
     # Replace external component path placeholder if present
     if "EXTERNAL_COMPONENT_PATH" in content:
-        external_components_path = str(FIXTURES_DIR / "external_components")
+        external_components_path = str(
+            Path(__file__).parent / "fixtures" / "external_components"
+        )
         content = content.replace("EXTERNAL_COMPONENT_PATH", external_components_path)
 
-    if shared_name is not None:
-        # _compile verifies the marked test compiles this content unmodified
-        request.node._shared_yaml_content = content
-
     return content
 
 
@@ -270,218 +233,24 @@ async def write_yaml_config(
 ) -> AsyncGenerator[ConfigWriter]:
     """Write YAML configuration to a file."""
     # Get the test name for default filename
-    base_name = request.node.name.replace("test_", "").partition("[")[0]
+    test_name = request.node.name
+    base_name = test_name.replace("test_", "").split("[")[0]
 
     async def _write_config(content: str, filename: str | None = None) -> Path:
         if filename is None:
             filename = f"{base_name}.yaml"
         config_path = integration_test_dir / filename
         loop = asyncio.get_running_loop()
-        await loop.run_in_executor(None, write_file, config_path, content)
+        await loop.run_in_executor(None, config_path.write_text, content)
         return config_path
 
     yield _write_config
 
 
-# Deliberately not CI-cached (ci.yml caches only platformio/ subpaths); stale
-# dirs for a fixture are pruned when its content hash changes.
-SHARED_BUILDS_ROOT = INTEGRATION_TESTS_ROOT / "builds"
-
-# In the dir name (not just the hash) so pruning stays inside this checkout
-_REPO_KEY = hashlib.sha256(str(REPO_ROOT).encode()).hexdigest()[:8]
-
-# Give a contended shared build lock time for a full cold compile ahead of us
-_SHARED_LOCK_TIMEOUT_S = 900
-_SHARED_LOCK_POLL_S = 0.1
-_SHARED_LOCK_REPORT_S = 30
-
-# Reclaims dirs orphaned by fixture renames or deleted checkouts
-_STALE_BUILD_MAX_AGE_S = 30 * 24 * 3600
-
-# ELF path per shared build dir; constant once compiled, so resolve it only once
-_shared_elf_paths: dict[Path, Path] = {}
-
-# Dirs this process already swept; pruning is session-scoped work
-_pruned_dirs: set[Path] = set()
-
-
-def _shared_yaml_name(request: pytest.FixtureRequest) -> str | None:
-    """Name passed to the shared_yaml marker, or None when unmarked."""
-    marker = request.node.get_closest_marker("shared_yaml")
-    if marker is None:
-        return None
-    # Exactly one \w+ positional arg: the name doubles as a build dir
-    # component, and CI test selection (script/helpers.py) parses the same shape
-    if (
-        len(marker.args) != 1
-        or marker.kwargs
-        or not re.fullmatch(r"\w+", str(marker.args[0]))
-    ):
-        raise ValueError(
-            "shared_yaml marker requires exactly one \\w+ fixture name literal"
-        )
-    return marker.args[0]
-
-
-def _shared_build_prefix(name: str) -> str:
-    return f"{name}-{_REPO_KEY}-"
-
-
-@cache
-def _shared_build_dir(name: str) -> Path:
-    """Dir keyed by checkout and fixture source, before per-test injections."""
-    key = hashlib.sha256((FIXTURES_DIR / f"{name}.yaml").read_bytes()).hexdigest()[:16]
-    return SHARED_BUILDS_ROOT / (_shared_build_prefix(name) + key)
-
-
-def _read_stamp(stamp: Path, shared_dir: Path) -> Path | None:
-    """ELF path recorded by the last completed compile, or None."""
-    try:
-        text = stamp.read_text(encoding="utf-8").strip()
-    except FileNotFoundError:
-        return None
-    except OSError as err:
-        print(f"Cannot read {stamp}: {err}")
-        return None
-    if not text:
-        print(f"Ignoring empty stamp {stamp}")
-        return None
-    built = Path(text)
-    # Never trust a stamp pointing outside its own build dir as an unlink target
-    if shared_dir.resolve() in built.resolve().parents:
-        return built
-    print(f"Ignoring stamp {stamp} pointing outside {shared_dir}")
-    return None
-
-
-def _unused_since(stale: Path, cutoff: float) -> bool:
-    """Whether a build dir looks untouched since cutoff; unknown counts as used."""
-    # Newest of the .built stamp (rewritten by every completed compile) and the
-    # dir itself (freshened by a worker claiming the dir before locking)
-    newest: float | None = None
-    for probe in (stale / ".built", stale):
-        try:
-            mtime = probe.stat().st_mtime
-        except FileNotFoundError:
-            continue
-        except NotADirectoryError:
-            return True  # a stray file where a dir should be; reclaimable
-        except OSError as err:
-            print(f"Cannot age-probe {stale}: {err}")
-            return False  # unknown never authorizes deletion
-        newest = mtime if newest is None else max(newest, mtime)
-    return newest is not None and newest < cutoff
-
-
-def _prune_stale_builds(name: str, keep: Path) -> None:
-    """Remove outdated build dirs (blocking, run in executor): this checkout's
-    other dirs for the fixture, plus anything untouched for 30 days. Tolerates
-    other workers pruning the same dirs concurrently."""
-    cutoff = time.time() - _STALE_BUILD_MAX_AGE_S
-    prefix = _shared_build_prefix(name)
-    for stale in SHARED_BUILDS_ROOT.iterdir():
-        if stale == keep:
-            continue
-        same_fixture = stale.name.startswith(prefix)
-        if not same_fixture and not _unused_since(stale, cutoff):
-            continue
-        # Creating .lock bumps the dir mtime, so remember whether the re-probe
-        # under the lock can trust it
-        lock_preexisting = (stale / ".lock").exists()
-        try:
-            lock_file = (stale / ".lock").open("w")
-        except FileNotFoundError:
-            continue  # pruned by another worker meanwhile
-        except NotADirectoryError:
-            print(f"Removing stray file {stale}")
-            stale.unlink(missing_ok=True)
-            continue
-        except OSError as err:
-            print(f"Cannot prune {stale}: {err}")
-            continue
-        with lock_file:
-            try:
-                fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
-            except BlockingIOError:
-                continue  # still in use by another run
-            # Re-probe under the lock: a worker freshens its dir before
-            # locking, so a just-claimed dir no longer looks unused. A dir
-            # whose .lock we just created cannot be held by anyone, and our
-            # own open bumped its mtime, so its pre-open probe stands
-            if (
-                lock_preexisting
-                and not same_fixture
-                and not _unused_since(stale, cutoff)
-            ):
-                continue
-            # rmtree tolerates races; a leftover partial tree only costs a
-            # rebuild, since the ELF is deleted before every compile
-            try:
-                rmtree(stale)
-            except OSError as err:
-                print(f"Failed to prune {stale}: {err}")
-
-
-async def _run_esphome_compile(
-    config_path: Path, cwd: Path, env: dict[str, str]
-) -> None:
-    """Run `esphome compile`, retrying up to 3 times on a segfault."""
-    max_retries = 3
-    for attempt in range(max_retries):
-        # Compile using subprocess, inheriting stdout/stderr to show progress
-        proc = await asyncio.create_subprocess_exec(
-            sys.executable,
-            "-m",
-            "esphome",
-            "compile",
-            str(config_path),
-            cwd=cwd,
-            stdout=None,  # Inherit stdout
-            stderr=None,  # Inherit stderr
-            stdin=asyncio.subprocess.DEVNULL,
-            # Start in a new process group to isolate signal handling
-            start_new_session=True,
-            env=env,
-            close_fds=False,
-        )
-        await proc.wait()
-
-        if proc.returncode == 0:
-            break
-        if proc.returncode == -11 and attempt < max_retries - 1:
-            # Segfault (-11 = SIGSEGV), retry
-            print(
-                f"Compilation segfaulted (attempt {attempt + 1}/{max_retries}), retrying..."
-            )
-            await asyncio.sleep(1)  # Brief pause before retry
-            continue
-        raise RuntimeError(
-            f"Failed to compile {config_path}, return code: {proc.returncode}. "
-            f"Run with 'pytest -s' to see compilation output."
-        )
-
-
-def _resolve_compiled_binary(config_path: Path) -> Path:
-    """Load the config to learn the compiled ELF path (blocking, run in executor)."""
-    CORE.reset()  # Reset CORE state between test runs
-    CORE.config_path = config_path
-    config = esphome.config.read_config(
-        {"command": "compile", "config": str(config_path)}
-    )
-    if config is None:
-        raise RuntimeError(f"Failed to read config from {config_path}")
-    idedata = get_idedata(config)
-    binary_path = Path(idedata.firmware_elf_path)
-    if not binary_path.exists():
-        raise RuntimeError(f"Compiled binary not found at {binary_path}")
-    return binary_path
-
-
 @pytest_asyncio.fixture
 async def compile_esphome(
     integration_test_dir: Path,
     shared_platformio_cache: Path,
-    request: pytest.FixtureRequest,
 ) -> AsyncGenerator[CompileFunction]:
     """Compile an ESPHome configuration and return the binary path."""
 
@@ -489,96 +258,66 @@ async def compile_esphome(
         # Use the shared PlatformIO cache for faster compilation
         # This avoids re-downloading dependencies for each test
         env = _get_platformio_env(shared_platformio_cache)
+
+        # Retry compilation up to 3 times if we get a segfault
+        max_retries = 3
+        for attempt in range(max_retries):
+            # Compile using subprocess, inheriting stdout/stderr to show progress
+            proc = await asyncio.create_subprocess_exec(
+                sys.executable,
+                "-m",
+                "esphome",
+                "compile",
+                str(config_path),
+                cwd=integration_test_dir,
+                stdout=None,  # Inherit stdout
+                stderr=None,  # Inherit stderr
+                stdin=asyncio.subprocess.DEVNULL,
+                # Start in a new process group to isolate signal handling
+                start_new_session=True,
+                env=env,
+                close_fds=False,
+            )
+            await proc.wait()
+
+            if proc.returncode == 0:
+                # Success!
+                break
+            if proc.returncode == -11 and attempt < max_retries - 1:
+                # Segfault (-11 = SIGSEGV), retry
+                print(
+                    f"Compilation segfaulted (attempt {attempt + 1}/{max_retries}), retrying..."
+                )
+                await asyncio.sleep(1)  # Brief pause before retry
+                continue
+            # Other error or final retry
+            raise RuntimeError(
+                f"Failed to compile {config_path}, return code: {proc.returncode}. "
+                f"Run with 'pytest -s' to see compilation output."
+            )
+
+        # Load the config to get idedata (blocking call, must use executor)
         loop = asyncio.get_running_loop()
 
-        name = _shared_yaml_name(request)
-        if name is None:
-            await _run_esphome_compile(config_path, integration_test_dir, env)
-            return await loop.run_in_executor(
-                None, _resolve_compiled_binary, config_path
+        def _read_config_and_get_binary():
+            CORE.reset()  # Reset CORE state between test runs
+            CORE.config_path = config_path
+            config = esphome.config.read_config(
+                {"command": "compile", "config": str(config_path)}
             )
+            if config is None:
+                raise RuntimeError(f"Failed to read config from {config_path}")
 
-        # Shared fixture: build in a hash-keyed dir so tests sharing a config
-        # pay one full compile and later only a main.cpp (port) rebuild + relink
-        shared_dir = _shared_build_dir(name)
-        shared_dir.mkdir(parents=True, exist_ok=True)
-        # Freshen the dir before locking so a concurrent age sweep, which
-        # re-probes under the lock, never reaps a dir a worker just claimed;
-        # if a peer reaped it already, the guarded lock open recreates it
-        with suppress(FileNotFoundError):
-            os.utime(shared_dir)
-        if shared_dir not in _pruned_dirs:
-            _pruned_dirs.add(shared_dir)
-            await loop.run_in_executor(None, _prune_stale_builds, name, shared_dir)
-        shared_config = shared_dir / f"{name}.yaml"
-        private_binary = integration_test_dir / f"{name}.elf"
-        content = await loop.run_in_executor(None, read_file, config_path)
-        if content != getattr(request.node, "_shared_yaml_content", None):
-            # The dir is keyed by the fixture source; a mutated config would be
-            # cached under a hash that does not describe it
-            raise RuntimeError(
-                "shared_yaml tests must compile the yaml_config content unmodified"
-            )
-        # flock serializes concurrent xdist workers; closing the fd releases it.
-        # Hand-rolled rather than filelock.FileLock: non-blocking retries keep
-        # the wait cancellable, while a blocking acquire in an executor thread
-        # would survive test cancellation holding the fd
-        try:
-            lock_file = (shared_dir / ".lock").open("w")
-        except FileNotFoundError:
-            # A peer run pruning divergent hashes reaped the dir between our
-            # mkdir and this open; recreate it and pay a full rebuild
-            shared_dir.mkdir(parents=True, exist_ok=True)
-            lock_file = (shared_dir / ".lock").open("w")
-        with lock_file:
-            start = time.monotonic()
-            last_report = start
-            while True:
-                try:
-                    fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
-                    break
-                except BlockingIOError:
-                    now = time.monotonic()
-                    if now - start > _SHARED_LOCK_TIMEOUT_S:
-                        raise RuntimeError(
-                            f"Timed out waiting for the {shared_dir} lock"
-                        ) from None
-                    if now - last_report >= _SHARED_LOCK_REPORT_S:
-                        last_report = now
-                        print(
-                            f"Waited {now - start:.0f}s for another worker's "
-                            f"build of {shared_dir.name}"
-                        )
-                    await asyncio.sleep(_SHARED_LOCK_POLL_S)
-            # .built carries the ELF path of the last completed compile, so
-            # later workers skip the config re-read in _resolve_compiled_binary
-            stamp = shared_dir / ".built"
-            if (built := _shared_elf_paths.get(shared_dir)) is None:
-                built = await loop.run_in_executor(None, _read_stamp, stamp, shared_dir)
-            # Delete the ELF before compiling: whatever exists afterwards is
-            # this compile's output, so no staleness check is ever needed.
-            # With no usable stamp, sweep any leftover at the known layout
-            if built is not None:
-                built.unlink(missing_ok=True)
-            else:
-                # Layout-agnostic: ESPHOME_BUILD_PATH can move the build tree
-                for leftover in shared_dir.rglob("program"):
-                    if leftover.is_file():
-                        leftover.unlink()
-            await loop.run_in_executor(
-                None, write_file_if_changed, shared_config, content
-            )
-            await _run_esphome_compile(shared_config, shared_dir, env)
-            if built is None or not built.exists():
-                built = await loop.run_in_executor(
-                    None, _resolve_compiled_binary, shared_config
-                )
-            _shared_elf_paths[shared_dir] = built
-            await loop.run_in_executor(None, write_file, stamp, str(built))
-            # Copy out before unlocking: another worker may relink firmware.elf
-            # while this test is still running its private copy
-            await loop.run_in_executor(None, shutil.copy2, built, private_binary)
-        return private_binary
+            # Get the compiled binary path
+            idedata = get_idedata(config)
+            return Path(idedata.firmware_elf_path)
+
+        binary_path = await loop.run_in_executor(None, _read_config_and_get_binary)
+
+        if not binary_path.exists():
+            raise RuntimeError(f"Compiled binary not found at {binary_path}")
+
+        return binary_path
 
     yield _compile
 

tests/integration/fixtures/api_noise_resume.yaml

@@ -0,0 +1,9 @@
+esphome:
+  name: host-noise-resume
+host:
+api:
+  encryption:
+    key: N4Yle5YirwZhPiHHsdZLdOA73ndj/84veVaLhTvxCuU=
+# VERY_VERBOSE so the frame helper logs "Session resumed!"
+logger:
+  level: VERY_VERBOSE

tests/integration/fixtures/sensor_filters_batch_window.yaml

@@ -0,0 +1,58 @@
+esphome:
+  name: test-batch-window-filters
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: DEBUG
+
+# Template sensor that we'll use to publish values
+sensor:
+  - platform: template
+    name: "Source Sensor"
+    id: source_sensor
+    accuracy_decimals: 2
+
+  # Batch window filters (window_size == send_every) - use streaming filters
+  - platform: copy
+    source_id: source_sensor
+    name: "Min Sensor"
+    id: min_sensor
+    filters:
+      - min:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Max Sensor"
+    id: max_sensor
+    filters:
+      - max:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+  - platform: copy
+    source_id: source_sensor
+    name: "Moving Avg Sensor"
+    id: moving_avg_sensor
+    filters:
+      - sliding_window_moving_average:
+          window_size: 5
+          send_every: 5
+          send_first_at: 1
+
+# Button to trigger publishing test values
+button:
+  - platform: template
+    name: "Publish Values Button"
+    id: publish_button
+    on_press:
+      - lambda: |-
+          // Publish 10 values: 1.0, 2.0, ..., 10.0
+          for (int i = 1; i <= 10; i++) {
+            id(source_sensor).publish_state(float(i));
+          }

tests/integration/fixtures/uart_mock_modbus_client_read_write.yaml

@@ -0,0 +1,111 @@
+esphome:
+  name: uart-mock-modbus-cli-rw
+
+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
+
+# Two virtual buses looped back to each other: the client's transmissions reach the server and the
+# server's replies reach the client. auto_start so forwarding is active before the button fires.
+uart_mock:
+  - id: virtual_uart_server
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_client
+              data: !lambda return data;
+  - id: virtual_uart_client
+    baud_rate: 9600
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+
+globals:
+  - id: stored_1
+    type: uint16_t
+    initial_value: "0"
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_client
+    id: virtual_modbus_client
+    role: client
+    turnaround_time: 10ms
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    registers:
+      # Writable + readable register: the read publishes what it returns, so the test can confirm the
+      # write half of the 0x17 ran before the read half (Modbus 6.17).
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: |-
+          id(srv_read_1).publish_state(id(stored_1));
+          return id(stored_1);
+        write_lambda: |-
+          id(stored_1) = x;
+          id(srv_write_1).publish_state(x);
+          return true;
+      # Read-only register, returned together with 0x01 by the 2-register read half.
+      - address: 0x02
+        value_type: U_WORD
+        read_lambda: return 0x00AA;
+
+sensor:
+  # Server-side observations.
+  - platform: template
+    name: "srv_write_1"
+    id: srv_write_1
+  - platform: template
+    name: "srv_read_1"
+    id: srv_read_1
+  # Client-side read-back: the values the client's on_response received.
+  - platform: template
+    name: "client_read_0"
+    id: client_read_0
+  - platform: template
+    name: "client_read_1"
+    id: client_read_1
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      # FC 0x17: write reg 0x0001 = 0x1234, then read regs 0x0001..0x0002 back in the same transaction.
+      - modbus_client.read_write_multiple_registers:
+          address: 0x01
+          read_address: 0x0001
+          read_count: 2
+          write_address: 0x0001
+          values: [0x1234]
+          on_response:
+            then:
+              - lambda: |-
+                  // values is the read-back block: reg 0x0001 (must be the just-written 0x1234) and reg 0x0002.
+                  if (values.size() >= 2) {
+                    id(client_read_0).publish_state(values[0]);
+                    id(client_read_1).publish_state(values[1]);
+                  }

tests/integration/fixtures/uart_mock_modbus_custom_pdu.yaml

@@ -0,0 +1,88 @@
+esphome:
+  name: uart-mock-modbus-custom-pdu
+
+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;
+
+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: 0x01
+        value_type: U_WORD
+        read_lambda: return 259;
+
+sensor:
+  # Plain read to confirm the controller <-> server link is up.
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "plain_read"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+  # Custom PDU: read holding register 0x0001, count 1. The PDU is
+  # {function code, address hi, address lo, count hi, count lo}; the device
+  # address and CRC are added by the hub. The lambda parses the response payload
+  # (the register value, big-endian).
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "custom_read"
+    custom_pdu: [0x03, 0x00, 0x01, 0x00, 0x01]
+    lambda: |-
+      if (data.size() < 2) return {};
+      return (float) ((data[0] << 8) | data[1]);
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_deprecated_write_buffer.yaml

@@ -0,0 +1,106 @@
+esphome:
+  name: uart-mock-modbus-dep-buffer
+
+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: reg10
+    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: 0x10
+        value_type: U_WORD
+        read_lambda: return id(reg10);
+        write_lambda: |-
+          id(reg10) = x;
+          return true;
+
+# A number whose write_lambda uses the DEPRECATED buffer parameter (fills `payload` with a legacy raw
+# frame as words: device address + function code + data) instead of the new item->write_* API. The write
+# must still land with its legacy semantics, and the one-time deprecation warning must fire only once per
+# entity no matter how many writes happen.
+number:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "buf_number"
+    id: buf_number
+    address: 0x10
+    register_type: holding
+    value_type: U_WORD
+    min_value: 0
+    max_value: 1000
+    step: 1
+    write_lambda: |-
+      // Legacy raw frame as words: [addr 0x01 | fc 0x06], register 0x0010, value.
+      payload.push_back(0x0106);
+      payload.push_back(0x0010);
+      payload.push_back((uint16_t) x);
+      return {};
+
+# Reports the server-side register so the test can observe that the deprecated buffer write landed.
+sensor:
+  - platform: template
+    name: "written_value"
+    id: written_value
+    update_interval: 0.5s
+    lambda: "return id(reg10);"
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    # The test drives the writes via number_command; the mock is autostart.

tests/integration/fixtures/uart_mock_modbus_lambda_invert.yaml

@@ -0,0 +1,95 @@
+esphome:
+  name: uart-mock-modbus-lambda-invert
+
+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: reg40
+    type: uint16_t
+    initial_value: "5"
+
+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: 0x40
+        value_type: U_WORD
+        read_lambda: return id(reg40);
+        write_lambda: id(reg40) = x; return true;
+
+# An active-low holding switch: the write_lambda inverts the wire value, but the entity must still
+# report the REQUESTED state. assumed_state keeps the register unpolled, so the published state comes
+# only from write_state() - turning ON writes 0x0000 yet the switch shows ON.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "invert_switch"
+    register_type: holding
+    address: 0x40
+    assumed_state: true
+    write_lambda: |-
+      return !x;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_40"
+    address: 0x40
+    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)

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)

tests/integration/fixtures/uart_mock_modbus_loopback.yaml

@@ -1,233 +0,0 @@
-esphome:
-  name: uart-mock-modbus-loopback
-
-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
-
-# Shared loopback fixture (see the shared_yaml markers in the test file);
-# register spaces are disjoint so each test only observes its own entities.
-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: reg10
-    type: uint16_t
-    initial_value: "100"
-  - id: reg11
-    type: uint16_t
-    initial_value: "200"
-  - id: reg12
-    type: uint16_t
-    initial_value: "300"
-  - id: reg13
-    type: uint16_t
-    initial_value: "0xABCD"
-  - id: reg30
-    type: uint16_t
-    initial_value: "0"
-  - id: reg40
-    type: uint16_t
-    initial_value: "5"
-  - id: reg50
-    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
-    registers:
-      - address: 0x01
-        value_type: U_WORD
-        read_lambda: return 259;
-      - address: 0x10
-        value_type: U_WORD
-        read_lambda: return id(reg10);
-        write_lambda: id(reg10) = x; return true;
-      - address: 0x11
-        value_type: U_WORD
-        read_lambda: return id(reg11);
-        write_lambda: id(reg11) = x; return true;
-      - address: 0x12
-        value_type: U_WORD
-        read_lambda: return id(reg12);
-        write_lambda: id(reg12) = x; return true;
-      - address: 0x13
-        value_type: U_WORD
-        read_lambda: return id(reg13);
-      - address: 0x30
-        value_type: U_WORD
-        read_lambda: return id(reg30);
-        write_lambda: id(reg30) = x; return true;
-      - address: 0x40
-        value_type: U_WORD
-        read_lambda: return id(reg40);
-        write_lambda: id(reg40) = x; return true;
-      - address: 0x50
-        value_type: U_WORD
-        read_lambda: return id(reg50);
-        write_lambda: id(reg50) = x; return true;
-
-# Byte-based offset: 2 bytes -> register 0x11 (the old code folded it in as a
-# register count, hitting 0x12). assumed_state keeps the switch write-only.
-switch:
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "offset_switch"
-    register_type: holding
-    address: 0x10
-    offset: 2
-    assumed_state: true
-  # Reading switch, byte offset 6 -> register 0x13; the pre-fix resolution (0x16)
-  # would draw ILLEGAL_DATA_ADDRESS and never publish.
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "read_offset_switch"
-    register_type: holding
-    address: 0x10
-    offset: 6
-    bitmask: 0x1
-  # Coil switch whose write_lambda dispatches a holding-register write via `item`;
-  # returning an empty optional suppresses the default coil write.
-  - 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 {};
-  # Active-low: the write_lambda inverts the wire value but the entity must still
-  # report the requested state (assumed_state keeps the register unpolled).
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "invert_switch"
-    register_type: holding
-    address: 0x40
-    assumed_state: true
-    write_lambda: |-
-      return !x;
-
-# Uses the deprecated buffer parameter (legacy raw frame as words); the write
-# must land and the deprecation warning must fire only once per entity.
-number:
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "buf_number"
-    id: buf_number
-    address: 0x50
-    register_type: holding
-    value_type: U_WORD
-    min_value: 0
-    max_value: 1000
-    step: 1
-    write_lambda: |-
-      // Legacy raw frame as words: [addr 0x01 | fc 0x06], register 0x0050, value.
-      payload.push_back(0x0106);
-      payload.push_back(0x0050);
-      payload.push_back((uint16_t) x);
-      return {};
-
-sensor:
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "plain_read"
-    address: 0x01
-    register_type: holding
-    value_type: U_WORD
-  # Custom PDU: read holding register 0x0001; device address and CRC are added
-  # by the hub. The lambda parses the big-endian register value.
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "custom_read"
-    custom_pdu: [0x03, 0x00, 0x01, 0x00, 0x01]
-    lambda: |-
-      if (data.size() < 2) return {};
-      return (float) ((data[0] << 8) | data[1]);
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_10"
-    address: 0x10
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_11"
-    address: 0x11
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_12"
-    address: 0x12
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_30"
-    address: 0x30
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_1
-    name: "reg_40"
-    address: 0x40
-    register_type: holding
-    value_type: U_WORD
-  # Reports the server-side register so the test can observe that the deprecated buffer write landed.
-  - platform: template
-    name: "written_value"
-    id: written_value
-    update_interval: 0.5s
-    lambda: "return id(reg50);"
-
-button:
-  - platform: template
-    name: "Start Scenario"
-    id: start_scenario_btn
-    # Nothing to start (mock is autostart); tests drive entities directly

tests/integration/fixtures/uart_mock_modbus_register_offset.yaml

@@ -0,0 +1,138 @@
+esphome:
+  name: uart-mock-modbus-reg-offset
+
+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: reg10
+    type: uint16_t
+    initial_value: "100"
+  - id: reg11
+    type: uint16_t
+    initial_value: "200"
+  - id: reg12
+    type: uint16_t
+    initial_value: "300"
+  - id: reg13
+    type: uint16_t
+    initial_value: "0xABCD"
+
+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: 0x10
+        value_type: U_WORD
+        read_lambda: return id(reg10);
+        write_lambda: id(reg10) = x; return true;
+      - address: 0x11
+        value_type: U_WORD
+        read_lambda: return id(reg11);
+        write_lambda: id(reg11) = x; return true;
+      - address: 0x12
+        value_type: U_WORD
+        read_lambda: return id(reg12);
+        write_lambda: id(reg12) = x; return true;
+      - address: 0x13
+        value_type: U_WORD
+        read_lambda: return id(reg13);
+        write_lambda: id(reg13) = x; return true;
+
+# A holding-register switch at 0x10 with a 2-BYTE offset. offset is byte-based, so the write must target
+# register 0x10 + 2/2 = 0x11. The old (pre-fix) behavior folded offset into the address as a register
+# count, hitting 0x12 instead. assumed_state keeps the switch write-only so it does not read any register.
+switch:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "offset_switch"
+    register_type: holding
+    address: 0x10
+    offset: 2
+    assumed_state: true
+  # A holding-register switch that READS its state. Byte offset 6 -> register 0x10 + 6/2 = 0x13. Post-fix
+  # the switch itself resolves to 0x13 (the even byte offset folds into the address as whole registers) and
+  # joins the 0x10..0x13 range, so no separate 0x13 sensor is needed. Pre-fix the whole byte offset folds
+  # into the address (0x16), where the server answers ILLEGAL_DATA_ADDRESS and the switch never publishes.
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "read_offset_switch"
+    register_type: holding
+    address: 0x10
+    offset: 6
+    bitmask: 0x1
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_10"
+    address: 0x10
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_11"
+    address: 0x11
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_12"
+    address: 0x12
+    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)

tests/integration/fixtures/uart_mock_modbus_server.yaml

@@ -0,0 +1,124 @@
+esphome:
+  name: uart-mock-modbus-server-test
+
+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_dev
+    baud_rate: 9600
+    rx_full_threshold: 120
+    rx_timeout: 2
+    auto_start: false
+    debug:
+    injections:
+      - delay: 100ms
+        inject_rx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] #  Read holding register 3 on device 1 (basic_read)
+      - delay: 100ms
+        #  Read holding register 7 on device 2
+        #  Reply from device 2
+        #  Read holding register 5 on device 1 (read_after_peer_response)
+        inject_rx:
+          [
+            0x02,
+            0x03,
+            0x00,
+            0x07,
+            0x00,
+            0x01,
+            0x35,
+            0xF8,
+            0x02,
+            0x03,
+            0x02,
+            0x00,
+            0xF0,
+            0xFC,
+            0x00,
+            0x01,
+            0x03,
+            0x00,
+            0x05,
+            0x00,
+            0x01,
+            0x94,
+            0x0B,
+          ]
+      - delay: 100ms
+        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] #  Read holding register 7 on device 2, with no response
+      - delay: 100ms
+        #  Read holding register 7 on device 2, with no response
+        #  Read holding register A on device 1 (read_after_peer_timeout)
+        inject_rx:
+          [
+            0x02,
+            0x03,
+            0x00,
+            0x07,
+            0x00,
+            0x01,
+            0x35,
+            0xF8,
+            0x01,
+            0x03,
+            0x00,
+            0x0A,
+            0x00,
+            0x01,
+            0xA4,
+            0x08,
+          ]
+
+modbus:
+  uart_id: virtual_uart_dev
+  role: server
+
+modbus_server:
+  - address: 1
+    registers:
+      - address: 0x03
+        value_type: U_WORD
+        read_lambda: |-
+          id(basic_read).publish_state(1);
+          return 1;
+      - address: 0x05
+        value_type: U_WORD
+        read_lambda: |-
+          id(read_after_peer_response).publish_state(1);
+          return 1;
+      - address: 0x0A
+        value_type: U_WORD
+        read_lambda: |-
+          id(read_after_peer_timeout).publish_state(1);
+          return 1;
+
+sensor:
+  - platform: template
+    name: "basic_read"
+    id: basic_read
+  - platform: template
+    name: "read_after_peer_response"
+    id: read_after_peer_response
+  - platform: template
+    name: "read_after_peer_timeout"
+    id: read_after_peer_timeout
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: "id(virtual_uart_dev).start_scenario();"

tests/integration/fixtures/uart_mock_modbus_server_controller.yaml

@@ -1,5 +1,5 @@
 esphome:
-  name: uart-mock-modbus-mesh
+  name: uart-mock-modbus-server-contro
 
 host:
 api:
@@ -17,14 +17,13 @@ uart:
   baud_rate: 115200
   port: /dev/null
 
-# Shared 3-bus mesh (see the shared_yaml markers): addr 1 = typed read-only
-# registers, addr 5 = the read/write 0x17 target, addr 2/3 on the second
-# server hub. auto_start everywhere: the controller polls at boot, so the
-# forwarding must already be live or early requests generate warnings.
-# Every test presses Start Scenario, so all merged actions fire in every test.
 uart_mock:
   - id: virtual_uart_server
     baud_rate: 9600
+    # auto_start must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
     auto_start: true
     debug:
     on_tx:
@@ -32,68 +31,35 @@ uart_mock:
           - uart_mock.inject_rx:
               id: virtual_uart_controller
               data: !lambda return data;
-          - uart_mock.inject_rx:
-              id: virtual_uart_server_2
-              data: !lambda return data;
-  - id: virtual_uart_server_2
-    baud_rate: 9600
-    auto_start: true
-    debug:
-    on_tx:
-      - then:
-          - uart_mock.inject_rx:
-              id: virtual_uart_server
-              data: !lambda return data;
-          - uart_mock.inject_rx:
-              id: virtual_uart_controller
-              data: !lambda return data;
   - id: virtual_uart_controller
     baud_rate: 9600
-    auto_start: true
+    auto_start: true # See comment on virtual_uart_server above
     debug:
     on_tx:
       - then:
           - uart_mock.inject_rx:
               id: virtual_uart_server
               data: !lambda return data;
-          - uart_mock.inject_rx:
-              id: virtual_uart_server_2
-              data: !lambda return data;
-
-globals:
-  - id: stored_1
-    type: uint16_t
-    initial_value: "0"
 
 modbus:
   - uart_id: virtual_uart_server
     id: virtual_modbus_server
     role: server
-  - uart_id: virtual_uart_server_2
-    id: virtual_modbus_server_2
-    role: server
   - uart_id: virtual_uart_controller
-    id: virtual_modbus_client
+    id: virtual_modbus_controller
     role: client
     turnaround_time: 10ms
 
 modbus_controller:
   - address: 1
-    modbus_id: virtual_modbus_client
+    modbus_id: virtual_modbus_controller
     id: modbus_controller_1
     update_interval: 1s
-  - address: 2
-    modbus_id: virtual_modbus_client
-    id: modbus_controller_2
-    update_interval: 1s
-  - address: 3
-    modbus_id: virtual_modbus_client
-    id: modbus_controller_3
-    update_interval: 1s
 
 modbus_server:
   - address: 1
     modbus_id: virtual_modbus_server
+    id: modbus_server_1
     registers:
       - address: 0x01
         value_type: U_WORD
@@ -137,34 +103,6 @@ modbus_server:
       - address: 0x28
         value_type: FP32_R
         read_lambda: return 3.14;
-  - address: 5
-    modbus_id: virtual_modbus_server
-    registers:
-      # Writable + readable register: srv_write_1 plus the client's read-back
-      # confirm the write half of the 0x17 ran before the read half (Modbus 6.17).
-      - address: 0x01
-        value_type: U_WORD
-        read_lambda: return id(stored_1);
-        write_lambda: |-
-          id(stored_1) = x;
-          id(srv_write_1).publish_state(x);
-          return true;
-      # Read-only register, returned together with 0x01 by the 2-register read half.
-      - address: 0x02
-        value_type: U_WORD
-        read_lambda: return 0x00AA;
-  - address: 2
-    modbus_id: virtual_modbus_server_2
-    registers:
-      - address: 0x01
-        value_type: U_WORD
-        read_lambda: return 919;
-  - address: 3
-    modbus_id: virtual_modbus_server_2
-    registers:
-      - address: 0x01
-        value_type: U_WORD
-        read_lambda: return 929;
 
 sensor:
   - platform: modbus_controller
@@ -257,46 +195,9 @@ sensor:
     address: 0x28
     register_type: holding
     value_type: FP32_R
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_2
-    name: "multi_reg_a"
-    address: 0x01
-    register_type: holding
-    value_type: U_WORD
-  - platform: modbus_controller
-    modbus_controller_id: modbus_controller_3
-    name: "multi_reg_b"
-    address: 0x01
-    register_type: holding
-    value_type: U_WORD
-  # client_read_write observations, server- and client-side.
-  - platform: template
-    name: "srv_write_1"
-    id: srv_write_1
-  - platform: template
-    name: "client_read_0"
-    id: client_read_0
-  - platform: template
-    name: "client_read_1"
-    id: client_read_1
 
 button:
   - platform: template
     name: "Start Scenario"
     id: start_scenario_btn
-    on_press:
-      # FC 0x17: write reg 0x0001 = 0x1234, then read regs 0x0001..0x0002 back in the same transaction.
-      - modbus_client.read_write_multiple_registers:
-          address: 5
-          read_address: 0x0001
-          read_count: 2
-          write_address: 0x0001
-          values: [0x1234]
-          on_response:
-            then:
-              - lambda: |-
-                  // values is the read-back block: reg 0x0001 (must be the just-written 0x1234) and reg 0x0002.
-                  if (values.size() >= 2) {
-                    id(client_read_0).publish_state(values[0]);
-                    id(client_read_1).publish_state(values[1]);
-                  }
+    # This test does not have anything to start (mock is autostart)

tests/integration/fixtures/uart_mock_modbus_server_controller_multiple.yaml

@@ -0,0 +1,116 @@
+esphome:
+  name: uart-mock-modbus-server-mult
+
+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 must be true for loopback fixtures: the modbus controller
+    # polls on its update_interval immediately at boot, so the uart_mock
+    # forwarding must already be active or early requests are lost and
+    # generate modbus warnings.
+    auto_start: true
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_server_2
+              data: !lambda return data;
+  - id: virtual_uart_server_2
+    baud_rate: 9600
+    auto_start: true # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_controller
+              data: !lambda return data;
+  - id: virtual_uart_controller
+    baud_rate: 9600
+    auto_start: true # See comment on virtual_uart_server above
+    debug:
+    on_tx:
+      - then:
+          - uart_mock.inject_rx:
+              id: virtual_uart_server
+              data: !lambda return data;
+          - uart_mock.inject_rx:
+              id: virtual_uart_server_2
+              data: !lambda return data;
+
+modbus:
+  - uart_id: virtual_uart_server
+    id: virtual_modbus_server
+    role: server
+  - uart_id: virtual_uart_server_2
+    id: virtual_modbus_server_2
+    role: server
+  - uart_id: virtual_uart_controller
+    id: virtual_modbus_client
+    role: client
+    turnaround_time: 10ms
+
+modbus_controller:
+  - address: 1
+    modbus_id: virtual_modbus_client
+    update_interval: 1s
+    id: modbus_controller_1
+  - address: 2
+    modbus_id: virtual_modbus_client
+    update_interval: 1s
+    id: modbus_controller_2
+
+modbus_server:
+  - address: 1
+    modbus_id: virtual_modbus_server
+    registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 919;
+  - address: 2
+    modbus_id: virtual_modbus_server_2
+    registers:
+      - address: 0x01
+        value_type: U_WORD
+        read_lambda: return 929;
+
+sensor:
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_1
+    name: "reg_u_word"
+    address: 0x01
+    register_type: holding
+    value_type: U_WORD
+  - platform: modbus_controller
+    modbus_controller_id: modbus_controller_2
+    name: "reg_u_word_2"
+    address: 0x01
+    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)

tests/integration/fixtures/uart_mock_modbus_server_read_write.yaml

@@ -1,5 +1,5 @@
 esphome:
-  name: uart-mock-modbus-srv-injected
+  name: uart-mock-modbus-srv-rw
 
 host:
 api:
@@ -17,8 +17,6 @@ uart:
   baud_rate: 115200
   port: /dev/null
 
-# Shared server-role fixture (see the shared_yaml markers in the test file);
-# the injections concatenate and each test waits only on its own sensors.
 uart_mock:
   - id: virtual_uart_dev
     baud_rate: 9600
@@ -27,31 +25,18 @@ uart_mock:
     auto_start: false
     debug:
     injections:
-      - delay: 100ms
-        inject_rx: [0x01, 0x03, 0x00, 0x03, 0x00, 0x01, 0x74, 0x0A] #  Read holding register 3 on device 1 (basic_read)
-      - delay: 100ms
-        #  Read holding register 7 on device 2, its reply, then read holding
-        #  register 5 on device 1 (read_after_peer_response)
-        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8,
-                    0x02, 0x03, 0x02, 0x00, 0xF0, 0xFC,
-                    0x00, 0x01, 0x03, 0x00, 0x05, 0x00, 0x01, 0x94, 0x0B]
-      - delay: 100ms
-        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8] #  Read holding register 7 on device 2, with no response
-      - delay: 100ms
-        #  Read holding register 7 on device 2 with no response, then read
-        #  holding register A on device 1 (read_after_peer_timeout)
-        inject_rx: [0x02, 0x03, 0x00, 0x07, 0x00, 0x01, 0x35, 0xF8,
-                    0x01, 0x03, 0x00, 0x0A, 0x00, 0x01, 0xA4, 0x08]
-      # FC 0x17 on device 1: write reg 0x0001 = 0x1234 then read 0x0001..0x0002;
-      # per Modbus 6.17 the write runs first, so 0x0001 must read back 0x1234.
+      # FC 0x17 Read/Write Multiple Registers on device 1:
+      #   write reg 0x0001 = 0x1234 (qty 1), then read regs 0x0001..0x0002 (qty 2).
+      # Per Modbus 6.17 the write is performed before the read, so reg 0x0001 must
+      # read back the just-written 0x1234 in the same request.
       - delay: 100ms
         inject_rx:
           [0x01, 0x17, 0x00, 0x01, 0x00, 0x02, 0x00, 0x01, 0x00, 0x01, 0x02, 0x12, 0x34, 0x49, 0xD8]
-      # FC 0x17: write reg 0x0006 = 0x5678 (qty 1), then read reg 0x0006 (qty 1) -
+      # FC 0x17: write reg 0x0003 = 0x5678 (qty 1), then read reg 0x0003 (qty 1) -
       # a write and read targeting a different register block.
       - delay: 100ms
         inject_rx:
-          [0x01, 0x17, 0x00, 0x06, 0x00, 0x01, 0x00, 0x06, 0x00, 0x01, 0x02, 0x56, 0x78, 0x8B, 0x55]
+          [0x01, 0x17, 0x00, 0x03, 0x00, 0x01, 0x00, 0x03, 0x00, 0x01, 0x02, 0x56, 0x78, 0x9B, 0x10]
 
 globals:
   - id: stored_1
@@ -85,18 +70,8 @@ modbus_server:
         read_lambda: |-
           id(rw_read_2).publish_state(0x00AA);
           return 0x00AA;
+      # Second writable + readable register, targeted by the second request.
       - address: 0x03
-        value_type: U_WORD
-        read_lambda: |-
-          id(basic_read).publish_state(1);
-          return 1;
-      - address: 0x05
-        value_type: U_WORD
-        read_lambda: |-
-          id(read_after_peer_response).publish_state(1);
-          return 1;
-      # Second writable + readable register, targeted by the second FC 0x17 request.
-      - address: 0x06
         value_type: U_WORD
         read_lambda: |-
           id(rw_read_3).publish_state(id(stored_3));
@@ -105,22 +80,8 @@ modbus_server:
           id(stored_3) = x;
           id(rw_write_3).publish_state(x);
           return true;
-      - address: 0x0A
-        value_type: U_WORD
-        read_lambda: |-
-          id(read_after_peer_timeout).publish_state(1);
-          return 1;
 
 sensor:
-  - platform: template
-    name: "basic_read"
-    id: basic_read
-  - platform: template
-    name: "read_after_peer_response"
-    id: read_after_peer_response
-  - platform: template
-    name: "read_after_peer_timeout"
-    id: read_after_peer_timeout
   - platform: template
     name: "rw_write_1"
     id: rw_write_1

tests/integration/host_prefs.py

@@ -1,7 +1,7 @@
 """Helpers for manipulating the host platform's preferences file.
 
 ESPHome's host platform stores preferences in
-``$ESPHOME_PREFDIR/<app_name>.prefs`` using a simple binary layout that
+``~/.esphome/prefs/<app_name>.prefs`` using a simple binary layout that
 mirrors ``HostPreferences::sync()``:
 ``[uint32_t key][uint8_t len][uint8_t data[len]]`` per entry.
 
@@ -11,21 +11,13 @@ boot (e.g. forcing safe mode) or to clear stale state between runs.
 
 from __future__ import annotations
 
-import os
 from pathlib import Path
 import struct
 
 
 def host_prefs_path(device_name: str) -> Path:
-    """Return the on-disk prefs file path for a host-platform device.
-
-    Requires ESPHOME_PREFDIR, which the autouse isolated_preferences fixture
-    sets; refusing the ~/.esphome/prefs fallback keeps tests off real user
-    data if the fixture is ever bypassed."""
-    prefdir = os.environ.get("ESPHOME_PREFDIR")
-    if not prefdir:
-        raise RuntimeError("ESPHOME_PREFDIR is not set; refusing the real prefs dir")
-    return Path(prefdir) / f"{device_name}.prefs"
+    """Return the on-disk prefs file path for a host-platform device."""
+    return Path.home() / ".esphome" / "prefs" / f"{device_name}.prefs"
 
 
 def clear_host_prefs(device_name: str) -> None:

tests/integration/test_api_noise_resume.py

@@ -0,0 +1,58 @@
+"""Integration test for noise session resume."""
+
+from __future__ import annotations
+
+import asyncio
+
+import aioesphomeapi.core
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+NOISE_KEY = "N4Yle5YirwZhPiHHsdZLdOA73ndj/84veVaLhTvxCuU="
+
+
+@pytest.mark.asyncio
+async def test_api_noise_resume(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """A reconnect with the ticket from the first connection resumes the session."""
+    if not hasattr(aioesphomeapi.core, "ResumeAPIError"):
+        pytest.skip("aioesphomeapi without noise session resume")
+
+    resumed = asyncio.Event()
+    resumed_count = 0
+
+    def on_line(line: str) -> None:
+        nonlocal resumed_count
+        if "Session resumed" in line:
+            resumed_count += 1
+            resumed.set()
+
+    async with (
+        run_compiled(yaml_config, line_callback=on_line),
+        api_client_connected(noise_psk=NOISE_KEY) as client,
+    ):
+        # First connection: full handshake, the device issues a ticket
+        info = await client.device_info()
+        assert info.name == "host-noise-resume"
+        assert resumed_count == 0
+
+        # Same client reconnects and offers the ticket
+        await client.disconnect()
+        await client.connect(login=True)
+        info = await client.device_info()
+        assert info.name == "host-noise-resume"
+        await asyncio.wait_for(resumed.wait(), timeout=10.0)
+        assert resumed_count == 1
+        resumed.clear()
+
+        # The resumed session issued a fresh ticket, so it resumes again
+        await client.disconnect()
+        await client.connect(login=True)
+        info = await client.device_info()
+        assert info.name == "host-noise-resume"
+        await asyncio.wait_for(resumed.wait(), timeout=10.0)
+        assert resumed_count == 2

tests/integration/test_api_zero_psk_provisioning.py

@@ -24,6 +24,12 @@ NEW_KEY = base64.b64encode(b"n" * 32)
 KEY_ACTIVATION_DELAY = 0.5
 
 
+@pytest.fixture(autouse=True)
+def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
+    """Keep host preferences per-test so every run starts unprovisioned."""
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
+
+
 @pytest.mark.asyncio
 async def test_api_zero_psk_provisioning(
     yaml_config: str,

tests/integration/test_host_preferences_suspend_resume.py

@@ -41,6 +41,15 @@ async def _poll_until_exists(path: Path) -> None:
         await asyncio.sleep(0.05)
 
 
+@pytest.fixture(autouse=True)
+def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> Path:
+    """Keep host preferences per-test so this test never touches the real
+    ~/.esphome/prefs and never races other tests over ESPHOME_PREFDIR."""
+    prefdir = tmp_path / "prefs"
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(prefdir))
+    return prefdir / f"{DEVICE_NAME}.prefs"
+
+
 @pytest.mark.asyncio
 async def test_host_preferences_suspend_resume(
     yaml_config: str,
@@ -49,7 +58,7 @@ async def test_host_preferences_suspend_resume(
     isolated_preferences: Path,
 ) -> None:
     """Test that a running syncer flushes, a suspended one doesn't, and resume restores flushing."""
-    pref_file = isolated_preferences / f"{DEVICE_NAME}.prefs"
+    pref_file = isolated_preferences
 
     loop = asyncio.get_running_loop()
     saved_in_memory = loop.create_future()

tests/integration/test_light_initial_state.py

@@ -11,6 +11,14 @@ from .state_utils import InitialStateHelper, require_entity
 from .types import APIClientConnectedFactory, RunCompiledFunction
 
 
+@pytest.fixture(autouse=True)
+def isolated_preferences(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
+    """Keep host preferences per-test so RESTORE_AND_ON never loads a stale value left
+    behind by a previous run (host preferences otherwise persist to ~/.esphome/prefs,
+    keyed only by device name)."""
+    monkeypatch.setenv("ESPHOME_PREFDIR", str(tmp_path / "prefs"))
+
+
 @pytest.mark.asyncio
 async def test_light_initial_state(
     yaml_config: str,

tests/integration/test_uart_mock_modbus.py

@@ -173,7 +173,6 @@ async def test_uart_mock_modbus_no_threshold(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_server_injected")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server(
     yaml_config: str,
@@ -204,7 +203,6 @@ async def test_uart_mock_modbus_server(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_server_injected")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server_read_write(
     yaml_config: str,
@@ -233,8 +231,8 @@ async def test_uart_mock_modbus_server_read_write(
             "rw_write_1": 4660,  # 0x1234 written to reg 0x0001
             "rw_read_1": 4660,  # reg 0x0001 reads back the just-written value
             "rw_read_2": 170,  # 0x00AA read from reg 0x0002 in the same request
-            "rw_write_3": 22136,  # 0x5678 written to reg 0x0006
-            "rw_read_3": 22136,  # reg 0x0006 reads back the just-written value
+            "rw_write_3": 22136,  # 0x5678 written to reg 0x0003
+            "rw_read_3": 22136,  # reg 0x0003 reads back the just-written value
         }
     )
 
@@ -243,8 +241,7 @@ async def test_uart_mock_modbus_server_read_write(
         api_client_connected() as client,
     ):
         await tracker.setup_and_start_scenario(client)
-        # The FC 0x17 injections fire last, behind four earlier 100ms delays
-        await tracker.await_all(futures, timeout=4.0)
+        await tracker.await_all(futures)
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
@@ -299,7 +296,6 @@ async def test_uart_mock_modbus_server_read_write_invalid(
     )
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server_controller(
     yaml_config: str,
@@ -489,7 +485,6 @@ async def test_uart_mock_modbus_server_controller_bits(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_server_controller_multiple(
     yaml_config: str,
@@ -500,7 +495,7 @@ async def test_uart_mock_modbus_server_controller_multiple(
 
     line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
 
-    expected_values = {"multi_reg_a": 919, "multi_reg_b": 929}
+    expected_values = {"reg_u_word": 919, "reg_u_word_2": 929}
     tracker = SensorTracker(list(expected_values.keys()))
     futures = tracker.expect_all(expected_values)
 
@@ -711,7 +706,6 @@ async def test_uart_mock_modbus_shared_address(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_custom_pdu(
     yaml_config: str,
@@ -938,7 +932,6 @@ async def test_uart_mock_modbus_broadcast_write(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_mesh")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_client_read_write(
     yaml_config: str,
@@ -954,7 +947,9 @@ async def test_uart_mock_modbus_client_read_write(
     """
     line_callback, error_log_lines, warning_log_lines = _make_modbus_line_callback()
 
-    tracker = SensorTracker(["srv_write_1", "client_read_0", "client_read_1"])
+    tracker = SensorTracker(
+        ["srv_write_1", "srv_read_1", "client_read_0", "client_read_1"]
+    )
     futures = tracker.expect_all(
         {
             "srv_write_1": 4660,  # server wrote 0x1234 to reg 0x0001
@@ -972,7 +967,6 @@ async def test_uart_mock_modbus_client_read_write(
         _assert_no_modbus_errors(error_log_lines, warning_log_lines)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_register_offset(
     yaml_config: str,
@@ -1028,7 +1022,6 @@ async def test_uart_mock_modbus_register_offset(
         )
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_lambda_write(
     yaml_config: str,
@@ -1065,7 +1058,6 @@ async def test_uart_mock_modbus_lambda_write(
         await tracker.await_change(wrote_30, "reg_30", timeout=4.0)
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_lambda_invert(
     yaml_config: str,
@@ -1121,7 +1113,6 @@ async def test_uart_mock_modbus_lambda_invert(
         )
 
 
-@pytest.mark.shared_yaml("uart_mock_modbus_loopback")
 @pytest.mark.asyncio
 async def test_uart_mock_modbus_deprecated_write_buffer(
     yaml_config: str,

tests/script/test_helpers.py

@@ -2122,34 +2122,6 @@ def test_get_cpp_changed_components_independent_of_cwd(
     ) == ["time"]
 
 
-def test_fixture_map_includes_shared_yaml_markers() -> None:
-    """Fixtures named only by shared_yaml markers must map to their test file."""
-    helpers.get_fixture_to_test_files.cache_clear()
-    mapping = helpers.get_fixture_to_test_files()
-    for fixture in (
-        "uart_mock_modbus_loopback",
-        "uart_mock_modbus_mesh",
-        "uart_mock_modbus_server_injected",
-    ):
-        assert mapping[fixture] == frozenset(
-            {"tests/integration/test_uart_mock_modbus.py"}
-        )
-
-
-def test_no_orphan_integration_fixtures() -> None:
-    """Every fixture must reach CI test selection; an orphan selects nothing."""
-    helpers.get_fixture_to_test_files.cache_clear()
-    mapping = helpers.get_fixture_to_test_files()
-    fixtures_dir = (Path(__file__).parent.parent / "integration" / "fixtures").resolve()
-    fixtures = list(fixtures_dir.glob("*.yaml"))
-    assert fixtures, f"no fixtures found under {fixtures_dir}"
-    # cache_init is covered via INTEGRATION_TESTS_TRIGGER_FILES instead
-    orphans = [
-        f.stem for f in fixtures if f.stem != "cache_init" and f.stem not in mapping
-    ]
-    assert not orphans, f"fixtures invisible to CI test selection: {orphans}"
-
-
 def test_lpt_partition_balances_skewed_weights() -> None:
     """Heavy items spread across groups instead of clustering."""
     items = [f"i{n}" for n in range(6)]