[api] Mark the last two raw varint writers nodiscard and make StateWaiter failures visible

A predicate that raises now fails its wait instead of dying inside the state callback,
and a timeout names the predicate it was waiting for.

esphome/components/api/proto.h

@@ -328,7 +328,7 @@ class ProtoEncode {
     return encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, value);
   }
   /// Encode a varint that is expected to be 1-2 bytes (e.g. zigzag RSSI, small lengths).
-  static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
   encode_varint_raw_short(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint32_t value) {
     if (value < VARINT_MAX_1_BYTE) [[likely]] {
       PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
@@ -357,7 +357,7 @@ class ProtoEncode {
   /// fast path -- any non-zero bit in the top 6 of 48 -- emits exactly 7 bytes
   /// with no per-byte branch. Falls back to the general loop otherwise.
   /// Caller must guarantee value fits in 48 bits (checked in debug builds).
-  static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
   encode_varint_raw_48bit(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint64_t value) {
 #ifdef ESPHOME_DEBUG_API
     assert(value < (1ULL << (MAC_ADDRESS_SIZE * 8)) && "encode_varint_raw_48bit: value exceeds 48 bits");

tests/integration/state_utils.py

@@ -66,18 +66,30 @@ class StateWaiter:
         ] = []
 
     def on_state(self, state: EntityState) -> None:
-        for predicate, future in list(self._waiters):
-            if not future.done() and predicate(state):
+        for predicate, future in self._waiters:
+            if future.done():
+                continue
+            try:
+                matched = predicate(state)
+            except Exception as exc:  # noqa: BLE001  the wait re-raises it, the callback must not die
+                future.set_exception(exc)
+                continue
+            if matched:
                 future.set_result(state)
 
     async def expect(
         self, predicate: Callable[[EntityState], bool], timeout: float = 5.0
     ) -> EntityState:
-        """Wait for the next state matching ``predicate``."""
+        """Wait for the next state matching ``predicate``; states seen before this call do not count."""
         entry = (predicate, asyncio.get_running_loop().create_future())
         self._waiters.append(entry)
         try:
-            return await asyncio.wait_for(entry[1], timeout)
+            async with asyncio.timeout(timeout):
+                return await entry[1]
+        except TimeoutError:
+            raise TimeoutError(
+                f"no state matched {predicate} within {timeout}s"
+            ) from None
         finally:
             self._waiters.remove(entry)