[core] Add bit_at, the read side of assign_bits

Co-authored-by: J. Nick Koston <nick@home-assistant.io>

.agents/skills/hardware-test/SKILL.md

@@ -115,7 +115,7 @@ Per-topic scripts, each described in the matching `reference/recipes/<component>
 (host driver, Zephyr driver probe), `rtsp_server/` (protocol client, raw rate reader, measurement set,
 API image grab), `cse7766/` (API observer with ping), `zephyr_coredump/` (reader flow, decode with the
 Zephyr tools), `libretiny_prefs/` (rtc preference harness header), `libretiny_crash/` (BK72xx crash probe header and
-capture decoder), `api_outgoing/` (wake cycle and Wi-Fi drop timing readers), `ld2450/` (zone number driver), `dfrobot_sen0395/` (API driver and case runner), `homeassistant_light/` (direct bulb driver, held-report case driver), `usb_audio/` (URL player), `speaker_media_player/` (API player driver, pipeline task case runner), `web_server/` (SSE client, request mix, SSE and HTTPD case runners), `mqtt/` (broker config with TLS, client-certificate and WebSocket listeners, test CA with server and client certificates, transport case runner, flash-and-drive runner, CLI TLS check without a board), `udp/` (datagram recorder, `udp.write` payload checker and case driver), `entity_icons/` (icon and device class readback over the API, MQTT discovery, web_server and dump_config, with a comparer).
+capture decoder), `api_outgoing/` (wake cycle and Wi-Fi drop timing readers), `ld2450/` (zone number driver, loopback harness driver with number and binary sensor states and TX frame summarizer), `api/` (raw plaintext native API client for user defined actions), `dfrobot_sen0395/` (API driver and case runner), `homeassistant_light/` (direct bulb driver, held-report case driver), `usb_audio/` (URL player), `speaker_media_player/` (API player driver, pipeline task case runner), `web_server/` (SSE client, request mix, SSE and HTTPD case runners), `mqtt/` (broker config with TLS, client-certificate and WebSocket listeners, test CA with server and client certificates, transport case runner, flash-and-drive runner, CLI TLS check without a board), `udp/` (datagram recorder, `udp.write` payload checker and case driver), `entity_icons/` (icon and device class readback over the API, MQTT discovery, web_server and dump_config, with a comparer).
 
 ## Templates in `$SKILL/templates`
 

.agents/skills/hardware-test/reference/boards.md

@@ -7,6 +7,9 @@ Ports are not listed on purpose: they follow the USB hub port and change on ever
 
 - Opening the port usually resets the board through DTR, but not always; `serial_capture.py --reset`
   pulses RTS to capture from boot. 74880 baud shows the ROM boot messages (`rst cause`, `boot mode`).
+- UART0 can be looped back onto itself (`USC0(0) |= (1 << UCLBE)` in an `on_boot` lambda) so a UART
+  component is testable without its device: frames written with `uart.write` arrive on RX. Log over the
+  API with `logger: baud_rate: 0`, since the serial port carries the component's bytes.
 - `esphome logs` on some CH340s emits thousands of duplicated garbage lines right after the RTS reset;
   a plain pyserial capture at 115200 is reliable.
 - A failed esptool write can wedge the adapter (`termios` error 22, or an open that hangs) until replug.

.agents/skills/hardware-test/reference/recipes/README.md

@@ -4,12 +4,13 @@ Each flow names the pull requests it was used on, so their result tables show wh
 
 One file per component, named after it:
 
-- [`api.md`](api.md): noise handshake and OTA encryption on the ESP8266, outgoing connection wake timing on deep sleep and after a Wi-Fi drop
+- [`api.md`](api.md): noise handshake and OTA encryption on the ESP8266, outgoing connection wake timing on deep sleep and after a Wi-Fi drop, user defined action listing and execution with a raw plaintext client
 - [`captive_portal.md`](captive_portal.md): fallback AP and /wifisave hammer
 - [`cse7766.md`](cse7766.md): Sonoff S31 parser check over the API
 - [`espnow.md`](espnow.md): ESP8266 broadcast receive
 - [`hdmi_cec.md`](hdmi_cec.md): two-board wire
-- [`ld2450.md`](ld2450.md): polygon zones, zone number round trip through the radar's readback
+- [`ld2450.md`](ld2450.md): polygon zones, zone number round trip through the radar's readback, ESP8266 loopback harness without a radar
+- [`ld2412.md`](ld2412.md): output pin level through the basic configuration command, held back writes
 - [`libretiny.md`](libretiny.md): rtc preference emulation, BK72xx crash report on the next boot
 - [`mqtt.md`](mqtt.md): broker bench, QoS, reconnect, crash log over MQTT, WebSocket transports with a local and a public broker, TLS trust sources (own CA, client certificate, certificate bundle, no verification) with the CLI check
 - [`nrf52.md`](nrf52.md): dongle board support
@@ -20,7 +21,7 @@ One file per component, named after it:
 - [`scheduler.md`](scheduler.md): loop wake A/B
 - [`udp.md`](udp.md): datagrams, multicast, sockets, `udp.write` payload shapes
 - [`usb_cdc_acm.md`](usb_cdc_acm.md): nRF52840 dongle host driver
-- [`wifi.md`](wifi.md): reboot_timeout with a fallback AP, DHCP lease cache, Easy Connect (DPP)
+- [`wifi.md`](wifi.md): reboot_timeout with a fallback AP, DHCP lease cache, Easy Connect (DPP), WISP bridge with an ethernet upstream
 - [`zephyr.md`](zephyr.md): preferences on the nRF52840 dongle
 - [`zephyr_coredump.md`](zephyr_coredump.md): crash and decode on nRF52 boards
 - [`zwave_proxy.md`](zwave_proxy.md): frame checksums on a live proxy

.agents/skills/hardware-test/reference/recipes/api.md

@@ -28,3 +28,15 @@ connected first, and the loop-time maxima. For the loop cost of polling, repeat
 so after a drop every dial is closed and HA's incoming reconnect wins 11 to 33 s later; the measurable effect is the
 first dial landing in the loop the network flag flips.
 Results: https://github.com/esphome/esphome/pull/21085#issuecomment-6254742791
+
+## User defined action listing and execution without a client library (#18882)
+
+`scripts/api/rawactions.py HOST list` speaks the plaintext native API over a raw socket and decodes every
+`ListEntitiesServicesResponse` by field number (argument fields 1 name, 2 type, 3 description, 4 example, 5 optional,
+6 default_value), so new wire fields show up without waiting for the client library. `rawactions.py HOST exec ACTION
+k=v ...` sends `ExecuteServiceRequest` with `sint32` ints (field 5, not the legacy int32 field 2), fixed32 floats,
+length delimited strings and `int[]` as repeated sint32 (field 7); `ALLOW_MISSING=1` leaves arguments out on purpose
+(the device then drops the call). `templates/api-actions-dut.yaml` (D1 mini, three actions, optional variables with
+defaults) and `templates/api-actions-strings-dut.yaml` (required strings) log what arrived. String arguments are
+`StringRef`s that are not null terminated: copy them with `snprintf("%.*s", size, c_str())` into a buffer before
+logging, and do not pass `%.*s` to `logger.log`, whose validator counts it as one pattern.

.agents/skills/hardware-test/reference/recipes/ld2412.md

@@ -0,0 +1,13 @@
+# ld2412
+
+## Output pin level and the basic configuration command (#21353)
+
+`templates/ld2412-select-only-dut.yaml` (ESP32-S3, the module on GPIO43 and GPIO44 at 115200, only the output pin
+level select, the module's OUT pin on GPIO38 as a binary sensor plus the presence sensor) with `uart: debug:
+direction: BOTH`. Drive the select with `scripts/ld2450/lddrive.py HOST KEY LOGFILE sel:NAME=OPTION` and read the TX
+dump for command `02` carrying the level byte, then the OUT pin binary sensor against presence: with level high the
+pin follows presence, with level low it is inverted. A module that never answers the boot query (no `<<<` lines)
+only shows the held back path: `Command 02 held back until the module reports its settings`, a query `12`, and after
+2 s `Command 02 not sent, the module did not report its settings` with the select republished to the last known
+level. With the full set of numbers configured a select change also needs every number to have a state, so wait
+for the boot query answer before judging.

.agents/skills/hardware-test/reference/recipes/ld2450.md

@@ -13,3 +13,30 @@ selects over the API and prints every LD2450 state update. After each set the hu
 the radar and republishes all twelve coordinates from the answer, so set each coordinate of each zone to a distinct
 value and check the readbacks; no target is needed. Readbacks can arrive late over Wi-Fi, so match them by content.
 Results: https://github.com/esphome/esphome/pull/21146#issuecomment-6278026922
+
+## ESP8266 loopback harness, no radar needed (#21349)
+
+`templates/ld2450-loopback.yaml` (package; substitutions `name` and `baud`) runs the hub on UART0 (GPIO1 and GPIO3)
+with `logger: baud_rate: 0` and sets the UART0 loopback bit in `on_boot` (`USC0(0) |= (1 << UCLBE)`), so everything
+the hub or a `uart.write` action sends comes back on RX. Template buttons inject periodic frames (all four direction
+strings, targets 2 and 3), query zone acks (zone type 0, 1 and 2 for the select readback) and call the name based
+setters with a valid and a bogus name. `uart: debug: direction: TX` delimited on the command footer dumps every
+command the hub sends. Drive it with `scripts/ld2450/lddrive.py HOST KEY LOGFILE press:NAME sel:NAME=OPTION wait:S`
+(logs over the API at VERBOSE, 2 s between steps because the API log stream drops lines while `send_command_`
+blocks), then `txsum.py LOGFILE` lists the TX frames per step; diff the summaries of the two builds. The hub reads its
+own commands back and logs `Invalid status` for each, which is harmless. The boot readback of the baud rate select
+comes from the UART setting, so a second build at `baud: "9600"` checks the flash table lookup.
+Results: https://github.com/esphome/esphome/pull/21349#issuecomment-6332320026
+
+## Frame splitter collisions (#21357)
+
+The loopback template also carries the zone 1 numbers and seven collision buttons: a periodic frame with the
+command footer inside target 1, zone acks with `55 CC` and with `04 03 02 01` inside the coordinates, a periodic
+frame cut after 12 bytes followed by a zone ack, a frame missing its last byte followed by a complete one, a bogus
+255 byte length then a frame, and garbage then a frame. Judge by the state updates plus the `Handling Periodic` and
+`Handling Ack` lines. Through a real radar the set zone ack is sometimes corrupted by the radar itself (its tail
+replaced by the periodic frame it was sending, for example `04 03 03 03 00 ... 55 CC`), so count readbacks over at
+least 24 selects per build and read the raw bytes with `uart: debug: direction: RX` (`!extend` the package's uart
+id) before blaming the parser; the API log stream drops lines under that dump volume, so some misses stay
+unclassified.
+Results: https://github.com/esphome/esphome/pull/21357#issuecomment-6333244886

.agents/skills/hardware-test/reference/recipes/wifi.md

@@ -23,3 +23,17 @@ Only an Android phone configures: Settings, network details, Add device, Share W
 QR scanner makes the phone an enrollee). Render the URI with the `qrcode` module and open the PNG.
 `esp32: framework: log_level: DEBUG` plus `sdkconfig_options: CONFIG_WPA_DEBUG_PRINT: y` prints the
 supplicant state machine.
+
+## WISP bridge with an ethernet upstream (#7147)
+
+`templates/wisp-ethernet-dut.yaml` (ESP32-S3 with a W5500, `wifi: ap:` with `ap_mode: WISP` and no station) and
+`templates/wisp-client.yaml` (D1 mini joining the bridge, `http_request` to a web server on the wired LAN and to
+`http://example.com/` every 15 s, `time: sntp`). Secrets: `wisp_psk` and `lan_http` (serve a text file on the host
+with `python -m http.server 8787 --bind 0.0.0.0` from a scratch directory). Judge by the client's `GET status 200`
+lines, the web server log showing the device's ethernet address as the source (NAT), the client's DHCP info (DNS1 is
+the LAN resolver) and `SNTP synced`. The device logs `Starting AP` and `DHCP DNS server: <resolver>` on its USB
+console; flashing over the air reboots it for repeat boots. Before the AP address and DHCP server moved ahead of the
+AP start (the fix in that PR), the first AP start failed on every boot with `esp_netif_dhcps_option failed: 20484`
+(DHCP server already started by the AP start event) and `Failed to enable NAT`, and the bridge only came up on the
+retry 90 s later; a client that cannot find the SSID for the first minute on an older build is that. With the fix
+the bridge is up 6 s after boot. Any change to the ESP-IDF AP start should also rerun the captive portal hammer.

.agents/skills/hardware-test/scripts/api/rawactions.py

@@ -0,0 +1,211 @@
+"""Plaintext native API client that lists user actions with their raw argument fields and executes one.
+
+usage: rawactions.py HOST list
+       rawactions.py HOST exec ACTION key=value [...]   (every argument must be given; types from the listing)
+Fields of ListEntitiesServicesArgument: 1 name, 2 type, 3 description, 4 example, 5 optional, 6 default_value.
+"""
+
+import os
+import socket
+import struct
+import sys
+
+HOST = sys.argv[1]
+CMD = sys.argv[2]
+ARGS = sys.argv[3:]
+TYPE_NAMES = {
+    0: "bool",
+    1: "int",
+    2: "float",
+    3: "string",
+    4: "bool[]",
+    5: "int[]",
+    6: "float[]",
+    7: "string[]",
+}
+
+
+def varint(v: int) -> bytes:
+    out = bytearray()
+    while True:
+        b = v & 0x7F
+        v >>= 7
+        if v:
+            out.append(b | 0x80)
+        else:
+            out.append(b)
+            return bytes(out)
+
+
+def svarint(v: int) -> bytes:
+    return varint((v << 1) ^ (v >> 63))
+
+
+def decode_varint(buf: bytes, pos: int) -> tuple[int, int]:
+    value = shift = 0
+    while True:
+        b = buf[pos]
+        pos += 1
+        value |= (b & 0x7F) << shift
+        if not b & 0x80:
+            return value, pos
+        shift += 7
+
+
+def fields(payload: bytes) -> list[tuple[int, int, object]]:
+    out = []
+    pos = 0
+    while pos < len(payload):
+        key, pos = decode_varint(payload, pos)
+        fnum, wt = key >> 3, key & 7
+        if wt == 0:
+            v, pos = decode_varint(payload, pos)
+        elif wt == 2:
+            ln, pos = decode_varint(payload, pos)
+            v = payload[pos : pos + ln]
+            pos += ln
+        elif wt == 5:
+            v = payload[pos : pos + 4]  # fixed32, caller unpacks
+            pos += 4
+        else:
+            raise ValueError(f"wire type {wt}")
+        out.append((fnum, wt, v))
+    return out
+
+
+def frame(mtype: int, payload: bytes) -> bytes:
+    return b"\x00" + varint(len(payload)) + varint(mtype) + payload
+
+
+def field_bytes(fnum: int, data: bytes) -> bytes:
+    return varint((fnum << 3) | 2) + varint(len(data)) + data
+
+
+def field_varint(fnum: int, v: int) -> bytes:
+    return varint((fnum << 3) | 0) + varint(v)
+
+
+def recv_frames(
+    sock: socket.socket, until: set[int], limit: float = 5.0
+) -> list[tuple[int, bytes]]:
+    sock.settimeout(limit)
+    buf = bytearray()
+    frames = []
+    while True:
+        try:
+            chunk = sock.recv(4096)
+        except TimeoutError:
+            return frames
+        if not chunk:
+            return frames
+        buf += chunk
+        while True:
+            if len(buf) < 3:
+                break
+            if buf[0] != 0:
+                raise ValueError("not plaintext")
+            try:
+                ln, p = decode_varint(buf, 1)
+                mt, p = decode_varint(buf, p)
+            except IndexError:
+                break
+            if len(buf) < p + ln:
+                break
+            frames.append((mt, bytes(buf[p : p + ln])))
+            del buf[: p + ln]
+            if mt in until:
+                return frames
+
+
+def main() -> None:
+    sock = socket.create_connection((HOST, 6053), timeout=5)
+    sock.sendall(
+        frame(
+            1, field_bytes(1, b"rawactions") + field_varint(2, 1) + field_varint(3, 10)
+        )
+    )  # HelloRequest
+    recv_frames(sock, {2})
+    sock.sendall(frame(3, b""))  # ConnectRequest, no password
+    recv_frames(sock, {4})
+    sock.sendall(frame(11, b""))  # ListEntitiesRequest
+    frames = recv_frames(sock, {19})
+    services = {}
+    for mt, payload in frames:
+        if mt != 41:
+            continue
+        name = key = None
+        args = []
+        for fnum, _, v in fields(payload):
+            if fnum == 1:
+                name = v.decode()
+            elif fnum == 2:
+                key = struct.unpack("<I", v)[0]
+            elif fnum == 3:
+                a = {
+                    "name": None,
+                    "type": 0,
+                    "description": None,
+                    "example": None,
+                    "optional": None,
+                    "default": None,
+                }
+                for af, _, av in fields(v):
+                    if af == 1:
+                        a["name"] = av.decode()
+                    elif af == 2:
+                        a["type"] = av
+                    elif af == 3:
+                        a["description"] = av.decode()
+                    elif af == 4:
+                        a["example"] = av.decode()
+                    elif af == 5:
+                        a["optional"] = bool(av)
+                    elif af == 6:
+                        a["default"] = av.decode()
+                    else:
+                        a[f"field{af}"] = av
+                args.append(a)
+        services[name] = (key, args)
+    if CMD == "list":
+        for name, (key, args) in services.items():
+            print(f"action {name} key={key}")
+            for a in args:
+                extra = {
+                    k: v
+                    for k, v in a.items()
+                    if k not in ("name", "type") and v is not None
+                }
+                print(f"  {a['name']}: {TYPE_NAMES.get(a['type'], a['type'])} {extra}")
+        return
+    name = ARGS[0]
+    key, args = services[name]
+    payload = varint((1 << 3) | 5) + struct.pack("<I", key)
+    given = dict(kv.split("=", 1) for kv in ARGS[1:])
+    for a in args:
+        if a["name"] not in given:
+            if os.environ.get("ALLOW_MISSING"):
+                continue  # leave the argument out of the request on purpose
+            sys.exit(f"missing {a['name']}")
+        raw = given[a["name"]]
+        t = a["type"]
+        if t == 0:
+            av = field_varint(1, 1 if raw.lower() == "true" else 0)
+        elif t == 1:
+            av = varint((5 << 3) | 0) + svarint(int(raw))  # sint32 int_
+        elif t == 2:
+            av = varint((3 << 3) | 5) + struct.pack("<f", float(raw))
+        elif t == 3:
+            av = field_bytes(4, raw.encode())
+        elif t == 5:
+            av = b"".join(
+                varint((7 << 3) | 0) + svarint(int(x)) for x in raw.split(",") if x
+            )  # repeated sint32 int_array
+        else:
+            sys.exit(f"type {t} not supported here")
+        payload += field_bytes(2, av)
+    sock.sendall(frame(42, payload))  # ExecuteServiceRequest
+    print(f"sent {name} with {given}")
+    sock.close()
+
+
+main()

.agents/skills/hardware-test/scripts/ld2450/lddrive.py

@@ -0,0 +1,123 @@
+"""Drive an LD24xx harness over the API: press buttons, set selects and numbers, print state updates, save the log.
+
+usage: lddrive.py HOST KEY LOGFILE OP [OP ...]   (KEY of - means a plaintext API)
+  OP: list | press:NAME | sel:NAME=OPTION | set:NAME=VALUE | wait:SECONDS
+Steps are 2 s apart: the API log stream drops lines while the hub blocks in send_command_.
+"""
+
+import asyncio
+from pathlib import Path
+import sys
+import time
+
+from aioesphomeapi import (
+    APIClient,
+    BinarySensorState,
+    ButtonInfo,
+    LogLevel,
+    NumberInfo,
+    NumberState,
+    SelectInfo,
+    SelectState,
+    SensorInfo,
+    SensorState,
+    TextSensorInfo,
+    TextSensorState,
+)
+
+HOST, KEY, LOGFILE = sys.argv[1], sys.argv[2], sys.argv[3]
+OPS = sys.argv[4:]
+T0 = time.time()
+STEP_GAP_S = 2.0
+
+
+def stamp() -> str:
+    return f"{time.time() - T0:7.3f}"
+
+
+async def run(c: APIClient, log) -> None:
+    info = await c.device_info()
+    print(
+        f"device {info.name} esphome {info.esphome_version} compiled {info.compilation_time}"
+    )
+    ents, _ = await c.list_entities_services()
+    by_key = {e.key: e for e in ents}
+    buttons = {e.name: e for e in ents if isinstance(e, ButtonInfo)}
+    selects = {e.name: e for e in ents if isinstance(e, SelectInfo)}
+    numbers = {e.name: e for e in ents if isinstance(e, NumberInfo)}
+
+    def on_state(s) -> None:
+        e = by_key.get(s.key)
+        if e is None:
+            return
+        missing = " (missing)" if getattr(s, "missing_state", False) else ""
+        if isinstance(s, TextSensorState):
+            print(f"{stamp()} TXT {e.name} = {s.state!r}{missing}", flush=True)
+        elif isinstance(s, BinarySensorState):
+            print(f"{stamp()} BIN {e.name} = {s.state}{missing}", flush=True)
+        elif isinstance(s, NumberState):
+            print(f"{stamp()} NUM {e.name} = {s.state:.0f}{missing}", flush=True)
+        elif isinstance(s, SelectState):
+            print(f"{stamp()} SEL {e.name} = {s.state!r}{missing}", flush=True)
+        elif (
+            isinstance(s, SensorState)
+            and isinstance(e, SensorInfo)
+            and "Count" in e.name
+        ):
+            print(f"{stamp()} CNT {e.name} = {s.state:.0f}", flush=True)
+
+    def on_log(msg) -> None:
+        log.write(f"{stamp()} {msg.message.decode(errors='replace')}\n")
+
+    c.subscribe_states(on_state)
+    c.subscribe_logs(on_log, log_level=LogLevel.LOG_LEVEL_VERBOSE)
+    await asyncio.sleep(1.5)
+    for op in OPS:
+        if op == "list":
+            for n in sorted(buttons):
+                print("button:", n)
+            for n in sorted(selects):
+                print("select:", n, selects[n].options)
+            for e in ents:
+                if isinstance(e, TextSensorInfo):
+                    print("text_sensor:", e.name)
+            continue
+        kind, _, rest = op.partition(":")
+        if kind == "wait":
+            await asyncio.sleep(float(rest))
+            continue
+        if kind == "press":
+            if rest not in buttons:
+                sys.exit(f"no button named {rest!r}; have {sorted(buttons)}")
+            print(f"{stamp()} >>> press {rest}", flush=True)
+            log.write(f"{stamp()} >>> press {rest}\n")
+            c.button_command(buttons[rest].key)
+        elif kind == "set":
+            name, _, value = rest.partition("=")
+            if name not in numbers:
+                sys.exit(f"no number named {name!r}; have {sorted(numbers)}")
+            print(f"{stamp()} >>> set {name} = {value}", flush=True)
+            log.write(f"{stamp()} >>> set {name} = {value}\n")
+            c.number_command(numbers[name].key, float(value))
+        elif kind == "sel":
+            name, _, value = rest.partition("=")
+            if name not in selects:
+                sys.exit(f"no select named {name!r}; have {sorted(selects)}")
+            print(f"{stamp()} >>> select {name} = {value}", flush=True)
+            log.write(f"{stamp()} >>> select {name} = {value}\n")
+            c.select_command(selects[name].key, value)
+        else:
+            sys.exit(f"unknown op {op!r}")
+        await asyncio.sleep(STEP_GAP_S)
+    await asyncio.sleep(1)
+
+
+async def main() -> None:
+    c = APIClient(HOST, 6053, None, noise_psk=None if KEY == "-" else KEY)
+    await c.connect(login=True)
+    with Path(LOGFILE).open("w", encoding="utf-8") as log:
+        await run(c, log)
+    await c.disconnect()
+
+
+asyncio.run(main())

.agents/skills/hardware-test/scripts/ld2450/txsum.py

@@ -0,0 +1,33 @@
+"""Summarize an lddrive.py log: for each step, the LD2450 TX frames, commands and publishes.
+
+usage: txsum.py LOGFILE   (diff the output of two builds)
+"""
+
+from pathlib import Path
+import re
+import sys
+
+ANSI = re.compile(r"\x1b\[[0-9;]*m")
+out: list[str] = []
+with Path(sys.argv[1]).open(encoding="utf-8", errors="replace") as log:
+    for raw in log:
+        line = ANSI.sub("", raw.rstrip("\n"))
+        m = re.match(r"\s*[\d.]+ (.*)", line)
+        if not m:
+            continue
+        text = m.group(1)
+        if text.startswith(">>>"):
+            out.append(f"\n# {text[4:]}")
+        elif "[uart_debug" in text:
+            out.append(f"  TX {text.split('>>> ')[1]}")
+        elif "[ld2450" in text and (
+            "Sending COMMAND" in text
+            or "zone type" in text.lower()
+            or "Query zone" in text
+        ):
+            out.append(f"  LOG {text.split(']: ', 1)[1]}")
+        elif "[select:" in text and ">>" in text:
+            out.append(f"  SEL {text.split(']: ', 1)[1]}")
+        elif "[text_sensor:098]" in text:
+            out.append(f"  TXT {text.split(']: ', 1)[1]}")
+print("\n".join(out))

.agents/skills/hardware-test/templates/api-actions-dut.yaml

@@ -0,0 +1,65 @@
+esphome:
+  name: ${name}
+
+esp8266:
+  board: d1_mini
+
+wifi:
+  ssid: !secret wifi_ssid
+  password: !secret wifi_password
+  power_save_mode: none
+
+logger:
+  level: DEBUG
+
+api:
+  reboot_timeout: 0s
+  actions:
+    - action: play_buzzer
+      description: Play a song or build one from parts
+      variables:
+        song_str:
+          type: string
+          description: Full RTTTL string
+          example: "scale:d=4,o=5,b=100:c,d,e"
+          required: false
+        notes:
+          type: string
+          required: false
+        octave:
+          type: int
+          default: 5
+        bpm:
+          type: int
+          default: 100
+        tempo:
+          type: float
+          default: 1.5
+        loud:
+          type: bool
+          default: true
+        label:
+          type: string
+          default: ""
+      then:
+        - lambda: |-
+            char s[40], n[40], l[40];
+            snprintf(s, sizeof(s), "%.*s", (int) song_str.size(), song_str.c_str());
+            snprintf(n, sizeof(n), "%.*s", (int) notes.size(), notes.c_str());
+            snprintf(l, sizeof(l), "%.*s", (int) label.size(), label.c_str());
+            ESP_LOGD("main", "play_buzzer song=%s(%d) notes=%s(%d) octave=%d bpm=%d tempo=%.1f loud=%d label=%s(%d)", s,
+                     (int) song_str.size(), n, (int) notes.size(), octave, bpm, tempo, loud, l, (int) label.size());
+    - action: plain_action
+      variables:
+        value: int
+      then:
+        - logger.log:
+            format: "plain_action value=%d"
+            args: ["value"]
+    - action: list_action
+      variables:
+        values: int[]
+      then:
+        - logger.log:
+            format: "list_action n=%d"
+            args: ["values.size()"]

.agents/skills/hardware-test/templates/api-actions-strings-dut.yaml

@@ -0,0 +1,28 @@
+esphome:
+  name: ${name}
+
+esp8266:
+  board: d1_mini
+
+wifi:
+  ssid: !secret wifi_ssid
+  password: !secret wifi_password
+  power_save_mode: none
+
+logger:
+  level: DEBUG
+
+api:
+  reboot_timeout: 0s
+  actions:
+    - action: str_action
+      variables:
+        song_str: string
+        notes: string
+        octave: int
+      then:
+        - lambda: |-
+            char s[40], n[40];
+            snprintf(s, sizeof(s), "%.*s", (int) song_str.size(), song_str.c_str());
+            snprintf(n, sizeof(n), "%.*s", (int) notes.size(), notes.c_str());
+            ESP_LOGD("main", "str_action song=%s(%d) notes=%s(%d) octave=%d", s, (int) song_str.size(), n, (int) notes.size(), octave);

.agents/skills/hardware-test/templates/ld2412-select-only-dut.yaml

@@ -0,0 +1,57 @@
+substitutions:
+  name: ld2412sel
+
+esphome:
+  name: ${name}
+
+esp32:
+  variant: esp32s3
+  board: esp32-s3-devkitc-1
+  flash_size: 8MB
+  framework:
+    type: esp-idf
+
+logger:
+  level: VERBOSE
+
+api:
+  reboot_timeout: 0s
+
+ota:
+  - platform: esphome
+
+wifi:
+  ssid: !secret wifi_ssid
+  password: !secret wifi_password
+  power_save_mode: none
+
+uart:
+  - id: uart_ld2412
+    tx_pin: GPIO43
+    rx_pin: GPIO44
+    baud_rate: 115200
+    debug:
+      direction: BOTH
+      dummy_receiver: false
+      after:
+        timeout: 10ms
+        bytes: 120
+
+ld2412:
+  id: ld2412_radar
+  uart_id: uart_ld2412
+
+select:
+  - platform: ld2412
+    ld2412_id: ld2412_radar
+    out_pin_level:
+      name: "LD2412 Hardware output pin level"
+
+binary_sensor:
+  - platform: gpio
+    name: "LD2412 Out Pin"
+    pin: GPIO38
+  - platform: ld2412
+    ld2412_id: ld2412_radar
+    has_target:
+      name: "LD2412 Presence"

.agents/skills/hardware-test/templates/ld2450-loopback.yaml

@@ -0,0 +1,258 @@
+esphome:
+  name: ${name}
+  on_boot:
+    priority: -100
+    then:
+      - lambda: |-
+          // UART0 TX looped back into its own RX so injected frames reach the hub
+          USC0(0) |= (1 << UCLBE);
+          ESP_LOGI("harness", "UART0 loopback on");
+
+esp8266:
+  board: d1_mini
+
+wifi:
+  ssid: !secret wifi_ssid
+  password: !secret wifi_password
+
+api:
+  reboot_timeout: 0s
+
+logger:
+  baud_rate: 0
+  level: VERBOSE
+
+uart:
+  id: radar_uart
+  tx_pin: GPIO1
+  rx_pin: GPIO3
+  baud_rate: ${baud}
+  debug:
+    direction: TX
+    dummy_receiver: false
+    after:
+      delimiter: [0x04, 0x03, 0x02, 0x01]
+
+ld2450:
+  id: radar
+  uart_id: radar_uart
+
+select:
+  - platform: ld2450
+    baud_rate:
+      name: Baud Rate
+    zone_type:
+      name: Zone Type
+
+text_sensor:
+  - platform: ld2450
+    version:
+      name: Radar Version
+    target_1:
+      direction:
+        name: T1 Direction
+    target_2:
+      direction:
+        name: T2 Direction
+    target_3:
+      direction:
+        name: T3 Direction
+
+sensor:
+  - platform: ld2450
+    target_count:
+      name: Target Count
+
+button:
+  # periodic frames: header, 3 x (x, y, speed, resolution), footer
+  - platform: template
+    name: Frame Approach
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0xE8, 0x83, 0x14, 0x00, 0x64, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55, 0xCC]
+  - platform: template
+    name: Frame Away
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0xE8, 0x83, 0x14, 0x80, 0x64, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55, 0xCC]
+  - platform: template
+    name: Frame Stationary
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0xE8, 0x83, 0x00, 0x00, 0x64, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55, 0xCC]
+  - platform: template
+    name: Frame None
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55, 0xCC]
+  - platform: template
+    name: Frame T2 T3
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0xF4, 0x81, 0x20, 0x83, 0x1E, 0x00, 0x64, 0x00,
+                 0x2C, 0x01, 0xB0, 0x84, 0x0A, 0x80, 0x64, 0x00,
+                 0x55, 0xCC]
+  # query zone acks: header, length, command C1, status 01, 00 00, zone type, 00, 24 coordinate bytes, footer
+  - platform: template
+    name: Zone Ack Disabled
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xFD, 0xFC, 0xFB, 0xFA, 0x1E, 0x00, 0xC1, 0x01, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x04, 0x03, 0x02, 0x01]
+  - platform: template
+    name: Zone Ack Detection
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xFD, 0xFC, 0xFB, 0xFA, 0x1E, 0x00, 0xC1, 0x01, 0x00, 0x00, 0x01, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x04, 0x03, 0x02, 0x01]
+  - platform: template
+    name: Zone Ack Filter
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xFD, 0xFC, 0xFB, 0xFA, 0x1E, 0x00, 0xC1, 0x01, 0x00, 0x00, 0x02, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x04, 0x03, 0x02, 0x01]
+  # name based setters, the path lambdas use
+  - platform: template
+    name: Set Zone Filter By Name
+    on_press:
+      - lambda: id(radar).set_zone_type("Filter");
+  - platform: template
+    name: Set Zone Bogus By Name
+    on_press:
+      - lambda: id(radar).set_zone_type("Bogus");
+  - platform: template
+    name: Set Baud 57600 By Name
+    on_press:
+      - lambda: id(radar).set_baud_rate("57600");
+  - platform: template
+    name: Set Baud Bogus By Name
+    on_press:
+      - lambda: id(radar).set_baud_rate("Bogus");
+  # target 1 at x 772, y 258: bytes 04 03 02 01 are the command footer
+  - platform: template
+    name: Frame CmdFooter In Payload
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xAA, 0xFF, 0x03, 0x00,
+                 0x04, 0x03, 0x02, 0x01, 0x00, 0x00, 0x64, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55, 0xCC]
+  # zone ack, type Detection, zone 1 x1 = 0xCC55: bytes 55 CC are the periodic footer
+  - platform: template
+    name: Zone Ack DataFooter In Payload
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xFD, 0xFC, 0xFB, 0xFA, 0x1E, 0x00, 0xC1, 0x01, 0x00, 0x00, 0x01, 0x00,
+                 0x55, 0xCC, 0xC8, 0x00, 0xDC, 0x05, 0xB8, 0x0B, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x04, 0x03, 0x02, 0x01]
+  # zone ack, type Filter, zone 1 x1 = 0x0304 and y1 = 0x0102: bytes 04 03 02 01 are the command footer
+  - platform: template
+    name: Zone Ack CmdFooter In Payload
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xFD, 0xFC, 0xFB, 0xFA, 0x1E, 0x00, 0xC1, 0x01, 0x00, 0x00, 0x02, 0x00,
+                 0x04, 0x03, 0x02, 0x01, 0xDC, 0x05, 0xB8, 0x0B, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x04, 0x03, 0x02, 0x01]
+  # a periodic frame cut after 12 bytes, then a complete zone ack (type Disabled, zeros)
+  - platform: template
+    name: Cut Frame Then Zone Ack
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xAA, 0xFF, 0x03, 0x00, 0x00, 0x00, 0xE8, 0x83, 0x14, 0x00, 0x64, 0x00,
+                 0xFD, 0xFC, 0xFB, 0xFA, 0x1E, 0x00, 0xC1, 0x01, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x04, 0x03, 0x02, 0x01]
+  # a periodic frame missing its last byte, then a complete one (target 1 moving away)
+  - platform: template
+    name: Cut Frame Then Frame
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0xE8, 0x83, 0x14, 0x00, 0x64, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55,
+                 0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0xE8, 0x83, 0x14, 0x80, 0x64, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55, 0xCC]
+  # a command header claiming a 255 byte payload, then a complete periodic frame (target 1 stationary)
+  - platform: template
+    name: Bogus Length Then Frame
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0xFD, 0xFC, 0xFB, 0xFA, 0xFF, 0x00,
+                 0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0xE8, 0x83, 0x00, 0x00, 0x64, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55, 0xCC]
+  # five garbage bytes, then a complete periodic frame (target 1 approaching)
+  - platform: template
+    name: Garbage Then Frame
+    on_press:
+      - uart.write:
+          id: radar_uart
+          data: [0x01, 0x02, 0x03, 0x42, 0x99,
+                 0xAA, 0xFF, 0x03, 0x00,
+                 0x00, 0x00, 0xE8, 0x83, 0x14, 0x00, 0x64, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+                 0x55, 0xCC]
+
+number:
+  - platform: ld2450
+    presence_timeout:
+      name: Presence Timeout
+    zone_1:
+      x1:
+        name: Z1 X1
+      y1:
+        name: Z1 Y1
+      x2:
+        name: Z1 X2
+      y2:
+        name: Z1 Y2

.agents/skills/hardware-test/templates/wisp-client.yaml

@@ -0,0 +1,57 @@
+substitutions:
+  name: wispcli
+  ap_ssid: wisp-bridge
+
+esphome:
+  name: ${name}
+
+esp8266:
+  board: d1_mini
+
+logger:
+  level: DEBUG
+
+wifi:
+  ssid: ${ap_ssid}
+  password: !secret wisp_psk
+  power_save_mode: none
+  reboot_timeout: 0s
+
+http_request:
+  esp8266_disable_ssl_support: true
+  timeout: 8s
+
+time:
+  - platform: sntp
+    id: sntp_time
+    on_time_sync:
+      - logger.log:
+          format: "SNTP synced"
+          level: INFO
+
+interval:
+  - interval: 15s
+    then:
+      - http_request.get:
+          url: !secret lan_http
+          capture_response: true
+          on_response:
+            - logger.log:
+                format: "LAN GET status %d body %s"
+                args: ["response->status_code", "body.c_str()"]
+                level: INFO
+          on_error:
+            - logger.log:
+                format: "LAN GET failed"
+                level: WARN
+      - http_request.get:
+          url: http://example.com/
+          on_response:
+            - logger.log:
+                format: "WAN GET status %d"
+                args: ["response->status_code"]
+                level: INFO
+          on_error:
+            - logger.log:
+                format: "WAN GET failed"
+                level: WARN

.agents/skills/hardware-test/templates/wisp-ethernet-dut.yaml

@@ -0,0 +1,40 @@
+substitutions:
+  name: wispeth
+  ap_ssid: wisp-bridge
+
+esphome:
+  name: ${name}
+
+esp32:
+  board: esp32-s3-devkitc-1
+  variant: esp32s3
+  flash_size: 16MB
+  framework:
+    type: esp-idf
+
+logger:
+  hardware_uart: USB_SERIAL_JTAG
+  level: VERBOSE
+
+api:
+  encryption:
+    key: !secret api_key
+  reboot_timeout: 0s
+
+ota:
+  - platform: esphome
+
+ethernet:
+  type: W5500
+  clk_pin: GPIO13
+  mosi_pin: GPIO11
+  miso_pin: GPIO12
+  cs_pin: GPIO14
+  interrupt_pin: GPIO10
+  reset_pin: GPIO9
+
+wifi:
+  ap:
+    ssid: ${ap_ssid}
+    password: !secret wisp_psk
+    ap_mode: WISP

AGENTS.md

@@ -212,9 +212,10 @@ file does, and it is the authority when they disagree. The most useful starting
         logged with `get_log_str` and `LOG_STR_ARG`, copied out with `ESPHOME_strncpy_P` before `publish_state`.
         A `dump_config` helper that picks a name returns `const LogString *` from `LOG_STR(...)`.
     *   **CRCs and bit tricks** come from `esphome/core/helpers.h` (`crc8`, `crc16`, `xor_bytes`, `sum16_bytes`,
-        `reverse_bits`, `encode_uint16`, `encode_value_le`, `sign_extend<BITS>`, `bcd_to_uint8`, `assign_bits`); pick
-        the helper's arguments instead of writing a loop. 7-byte BCD RTC calendars use `time::decode_bcd_calendar<LAYOUT>`
-        and `encode_bcd_calendar<LAYOUT>` with `EPSON_BCD_CALENDAR`, `DS_BCD_CALENDAR` or `PCF_BCD_CALENDAR`.
+        `reverse_bits`, `encode_uint16`, `encode_value_le`, `sign_extend<BITS>`, `bcd_to_uint8`, `assign_bits`,
+        `bit_at`); pick the helper's arguments instead of writing a loop. 7-byte BCD RTC calendars use
+        `time::decode_bcd_calendar<LAYOUT>` and `encode_bcd_calendar<LAYOUT>` with `EPSON_BCD_CALENDAR`,
+        `DS_BCD_CALENDAR` or `PCF_BCD_CALENDAR`.
     *   **Bus helpers:** UART drains with `read_available(on_byte)` and `discard_available()`, text lines with
         `LineReader<N>`; I2C has `update_register`, `write_registers` (PROGMEM `I2CRegValue` table), `check_chip_id`
         and `read_bytes_16_le`. `if (!this->status_comm_result(ok)) return;` sets or clears the warning.

esphome/automation.py

@@ -24,6 +24,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, ID, EsphomeError, HexInt, Lambda
 from esphome.cpp_generator import (
+    CallExpression,
     Expression,
     LambdaExpression,
     MockObj,
@@ -446,9 +447,17 @@ async def _apply_parent(config: ConfigType, id_key: str = CONF_ID) -> str:
 
 
 def _apply_lambda_args(args: TemplateArgsType) -> TemplateArgsType:
-    # The generated function's parameters; a std::string arg is never copied.
+    # The generated function's parameters: a reference keeps its spelling, so a mutable
+    # one stays mutable; a value is taken by const reference, so a std::string is never copied.
     return [
-        (cg.RawExpression(f"const std::remove_cvref_t<{cg.safe_exp(t)}> &"), arg)
+        (
+            (
+                t
+                if str(cg.safe_exp(t)).rstrip().endswith("&")
+                else cg.RawExpression(f"const std::remove_cvref_t<{cg.safe_exp(t)}> &")
+            ),
+            arg,
+        )
         for t, arg in args
     ]
 
@@ -578,7 +587,7 @@ def apply_action_builder(
             if isinstance(spec, ApplyBytes):
                 statements.extend(
                     await _render_bytes(
-                        spec, config, receiver, parent, lambda_args, table_name
+                        spec, config, receiver, parent, args, table_name
                     )
                 )
                 continue
@@ -608,10 +617,14 @@ async def _render_bytes(
     config: ConfigType,
     receiver: str,
     parent: str,
-    lambda_args: TemplateArgsType,
+    args: TemplateArgsType,
     table_name: str,
 ) -> list[str]:
-    """The statements for one ``ApplyBytes``: what ``TemplatableBytes::visit`` does, inlined."""
+    """The statements for one ``ApplyBytes``: what ``TemplatableBytes::visit`` does, inlined.
+
+    The payload lambda keeps the automation's own argument types, as ``TemplatableBytes``
+    gave it, and is always called, so a body that modifies an argument still compiles.
+    """
     value = _config_lookup(config, spec.conf_key)
     if value is None:
         return []
@@ -622,21 +635,23 @@ async def _render_bytes(
 
     if isinstance(value, Lambda):
         inner = await cg.process_lambda(
-            value, lambda_args, return_type=cg.std_vector.template(cg.uint8)
+            value, args, return_type=cg.std_vector.template(cg.uint8)
         )
+        # Never reduced into the outer function, whose own parameters are const
+        call = CallExpression(inner, *[cg.MockObj(name) for _, name in args])
         return [
             "{",
-            f"  const std::vector<uint8_t> esphome__bytes = ({call_lambda(inner)});",
+            f"  const std::vector<uint8_t> esphome__bytes = {call};",
             f"  {send(spec.target, 'esphome__bytes.data(), esphome__bytes.size()')}",
             "}",
         ]
     table, size = _static_bytes(table_name, value)
     # Qualified so a trigger argument cannot shadow a reused table; an empty payload is nullptr
-    args = f"::{table}, {size}" if size else f"{table}, {size}"
+    table_args = f"::{table}, {size}" if size else f"{table}, {size}"
     if CORE.is_esp8266 and spec.progmem_target is not None:
-        return [send(spec.progmem_target, args)]
+        return [send(spec.progmem_target, table_args)]
     if not CORE.is_esp8266:
-        return [send(spec.target, args)]
+        return [send(spec.target, table_args)]
     # A PROGMEM table must be read with aligned loads on ESP8266, so copy it out first;
     # the length is known here, so the copy is an exact array, on the heap past ``stack``.
     # An empty payload still gets a one byte array: the raw lwIP socket rejects a null buffer.
@@ -649,7 +664,7 @@ async def _render_bytes(
         )
     statements = ["{", f"  {buffer}"]
     if size:
-        statements.append(f"  esphome::progmem_memcpy({data}, {args});")
+        statements.append(f"  esphome::progmem_memcpy({data}, {table_args});")
     return [*statements, f"  {send(spec.target, f'{data}, {size}')}", "}"]
 
 

esphome/components/cc1101/__init__.py

@@ -295,11 +295,6 @@ async def to_code(config):
         )
 
 
-# Actions
-SendPacketAction = ns.class_(
-    "SendPacketAction", automation.Action, cg.Parented.template(CC1101Component)
-)
-
 CC1101_ACTION_SCHEMA = cv.Schema(
     maybe_simple_id({cv.GenerateID(CONF_ID): cv.use_id(CC1101Component)})
 )
@@ -338,23 +333,11 @@ SEND_PACKET_ACTION_SCHEMA = cv.maybe_simple_value(
 )
 
 
-@automation.register_action(
+automation.register_apply_action(
     "cc1101.send_packet",
-    SendPacketAction,
     SEND_PACKET_ACTION_SCHEMA,
-    synchronous=True,
+    automation.ApplyBytes(CONF_DATA, "transmit_packet({})"),
 )
-async def send_packet_action_to_code(config, action_id, template_arg, args):
-    var = cg.new_Pvariable(action_id, template_arg)
-    await cg.register_parented(var, config[CONF_ID])
-    await automation.templatable_bytes(
-        config[CONF_DATA],
-        args,
-        var.set_data_template,
-        var.set_data_static,
-        "cc1101_data",
-    )
-    return var
 
 
 _CONFIG_VALIDATORS = {opt.schema: validator for opt, validator in CONFIG_MAP.items()}

esphome/components/cc1101/cc1101.h

@@ -132,13 +132,4 @@ class CC1101Component final : public Component,
   bool enter_tx_();
 };
 
-// Action Wrappers
-template<typename... Ts> class SendPacketAction final : public Action<Ts...>, public Parented<CC1101Component> {
-  TEMPLATABLE_BYTES(data)
-
-  void play(const Ts &...x) override {
-    this->data_.visit([this](const uint8_t *data, size_t len) { this->parent_->transmit_packet(data, len); }, x...);
-  }
-};
-
 }  // namespace esphome::cc1101

esphome/components/esphome/ota/ota_esphome_udp.cpp

@@ -222,10 +222,13 @@ void ESPHomeOTAComponent::udp_open_(UdpLink &link) {
   if (!random_bytes(token, sizeof(token)))
     return;
 #ifdef ESPHOME_OTA_UDP_LWIP
+#if USE_NETWORK_IPV4
   uint32_t ipv4;
   if (socket::sockaddr_to_ipv4(reinterpret_cast<const struct sockaddr *>(&peer), &ipv4)) {
     ip_addr_set_ip4_u32(&link.peer_ip, ipv4);
-  } else {
+  } else
+#endif
+  {
 #if USE_NETWORK_IPV6 && LWIP_IPV6
     inet6_addr_to_ip6addr(ip_2_ip6(&link.peer_ip), &reinterpret_cast<const sockaddr_in6 *>(&peer)->sin6_addr);
     IP_SET_TYPE_VAL(link.peer_ip, IPADDR_TYPE_V6);

esphome/components/factory_reset/button/__init__.py

@@ -1,29 +1,20 @@
-import esphome.codegen as cg
 from esphome.components import button
-import esphome.config_validation as cv
 from esphome.const import (
-    CONF_ID,
     DEVICE_CLASS_RESTART,
     ENTITY_CATEGORY_CONFIG,
     ICON_RESTART_ALERT,
 )
 from esphome.types import ConfigType
 
-from .. import factory_reset_ns
-
-FactoryResetButton = factory_reset_ns.class_(
-    "FactoryResetButton", button.Button, cg.Component
-)
-
 CONFIG_SCHEMA = button.button_schema(
-    FactoryResetButton,
+    button.ApplyButton,
     device_class=DEVICE_CLASS_RESTART,
     entity_category=ENTITY_CATEGORY_CONFIG,
     icon=ICON_RESTART_ALERT,
-).extend(cv.COMPONENT_SCHEMA)
+)
 
 
 async def to_code(config: ConfigType) -> None:
-    var = cg.new_Pvariable(config[CONF_ID])
-    await cg.register_component(var, config)
-    await button.register_button(var, config)
+    await button.sub_buttons(config).create(
+        config, None, apply="factory_reset::factory_reset_device()"
+    )

esphome/components/factory_reset/button/factory_reset_button.cpp

@@ -13,24 +13,20 @@ namespace esphome::factory_reset {
 
 ESPHOME_LOG_TAG(TAG, "factory_reset.button");
 
-void FactoryResetButton::dump_config() { LOG_BUTTON("", "Factory Reset Button", this); }
-void FactoryResetButton::press_action() {
+static void reset_and_reboot() {
+  global_preferences->reset();
+  App.safe_reboot();
+}
+
+void factory_reset_device() {
   ESP_LOGI(TAG, "Resetting");
   // Let MQTT settle a bit
   delay(100);  // NOLINT
 #ifdef USE_OPENTHREAD
-  openthread::global_openthread_component->on_factory_reset(FactoryResetButton::factory_reset_callback);
+  openthread::global_openthread_component->on_factory_reset(reset_and_reboot);
 #else
-  global_preferences->reset();
-  App.safe_reboot();
+  reset_and_reboot();
 #endif
 }
 
-#ifdef USE_OPENTHREAD
-void FactoryResetButton::factory_reset_callback() {
-  global_preferences->reset();
-  App.safe_reboot();
-}
-#endif
-
 }  // namespace esphome::factory_reset

esphome/components/factory_reset/button/factory_reset_button.h

@@ -1,21 +1,8 @@
 #pragma once
 
-#include "esphome/core/defines.h"
-
-#include "esphome/components/button/button.h"
-#include "esphome/core/component.h"
-
 namespace esphome::factory_reset {
 
-class FactoryResetButton final : public button::Button, public Component {
- public:
-  void dump_config() override;
-#ifdef USE_OPENTHREAD
-  static void factory_reset_callback();
-#endif
-
- protected:
-  void press_action() override;
-};
+/// Run by the factory reset button.
+void factory_reset_device();
 
 }  // namespace esphome::factory_reset

esphome/components/restart/button/__init__.py

@@ -1,26 +1,16 @@
-import esphome.codegen as cg
 from esphome.components import button
-import esphome.config_validation as cv
-from esphome.const import (
-    CONF_ID,
-    DEVICE_CLASS_RESTART,
-    ENTITY_CATEGORY_CONFIG,
-    ICON_RESTART,
-)
+from esphome.const import DEVICE_CLASS_RESTART, ENTITY_CATEGORY_CONFIG, ICON_RESTART
 from esphome.types import ConfigType
 
-restart_ns = cg.esphome_ns.namespace("restart")
-RestartButton = restart_ns.class_("RestartButton", button.Button, cg.Component)
-
 CONFIG_SCHEMA = button.button_schema(
-    RestartButton,
+    button.ApplyButton,
     icon=ICON_RESTART,
     device_class=DEVICE_CLASS_RESTART,
     entity_category=ENTITY_CATEGORY_CONFIG,
-).extend(cv.COMPONENT_SCHEMA)
+)
 
 
 async def to_code(config: ConfigType) -> None:
-    var = cg.new_Pvariable(config[CONF_ID])
-    await cg.register_component(var, config)
-    await button.register_button(var, config)
+    await button.sub_buttons(config).create(
+        config, None, apply="restart::restart_device()"
+    )

esphome/components/restart/button/restart_button.cpp

@@ -7,12 +7,11 @@ namespace esphome::restart {
 
 ESPHOME_LOG_TAG(TAG, "restart.button");
 
-void RestartButton::press_action() {
+void restart_device() {
   ESP_LOGI(TAG, "Restarting device");
   // Let MQTT settle a bit
   delay(100);  // NOLINT
   App.safe_reboot();
 }
-void RestartButton::dump_config() { LOG_BUTTON("", "Restart Button", this); }
 
 }  // namespace esphome::restart

esphome/components/restart/button/restart_button.h

@@ -1,19 +1,8 @@
 #pragma once
 
-#include "esphome/components/button/button.h"
-#include "esphome/core/component.h"
-
 namespace esphome::restart {
 
-class RestartButton final : public button::Button, public Component {
- public:
-  // User provided, not "= default": `new(p) RestartButton()` would zero-fill .bss that is already zero.
-  RestartButton() {}
-
-  void dump_config() override;
-
- protected:
-  void press_action() override;
-};
+/// Run by the restart button.
+void restart_device();
 
 }  // namespace esphome::restart

esphome/components/safe_mode/button/__init__.py

@@ -1,4 +1,3 @@
-import esphome.codegen as cg
 from esphome.components import button
 import esphome.config_validation as cv
 from esphome.const import (
@@ -9,26 +8,18 @@ from esphome.const import (
 )
 from esphome.types import ConfigType
 
-from .. import SafeModeComponent, safe_mode_ns
+from .. import SafeModeComponent
 
 DEPENDENCIES = ["safe_mode"]
 
-SafeModeButton = safe_mode_ns.class_("SafeModeButton", button.Button, cg.Component)
-
-CONFIG_SCHEMA = (
-    button.button_schema(
-        SafeModeButton,
-        device_class=DEVICE_CLASS_RESTART,
-        entity_category=ENTITY_CATEGORY_CONFIG,
-        icon=ICON_RESTART_ALERT,
-    )
-    .extend({cv.GenerateID(CONF_SAFE_MODE): cv.use_id(SafeModeComponent)})
-    .extend(cv.COMPONENT_SCHEMA)
-)
+CONFIG_SCHEMA = button.button_schema(
+    button.ApplyButton,
+    device_class=DEVICE_CLASS_RESTART,
+    entity_category=ENTITY_CATEGORY_CONFIG,
+    icon=ICON_RESTART_ALERT,
+).extend({cv.GenerateID(CONF_SAFE_MODE): cv.use_id(SafeModeComponent)})
 
 
 async def to_code(config: ConfigType) -> None:
-    var = await button.new_button(config)
-    await cg.register_component(var, config)
-
-    await cg.add_variable(var.set_safe_mode, config[CONF_SAFE_MODE])
+    buttons = button.sub_buttons(config, parent=config[CONF_SAFE_MODE])
+    await buttons.create(config, None, apply="restart_in_safe_mode()")

esphome/components/safe_mode/button/safe_mode_button.cpp

@@ -1,21 +0,0 @@
-#include "safe_mode_button.h"
-#include "esphome/core/hal.h"
-#include "esphome/core/log.h"
-#include "esphome/core/application.h"
-
-namespace esphome::safe_mode {
-
-ESPHOME_LOG_TAG(TAG, "safe_mode.button");
-
-void SafeModeButton::press_action() {
-  ESP_LOGI(TAG, "Restarting in safe mode");
-  this->safe_mode_component_->set_safe_mode_pending(true);
-
-  // Let MQTT settle a bit
-  delay(100);  // NOLINT
-  App.safe_reboot();
-}
-
-void SafeModeButton::dump_config() { LOG_BUTTON("", "Safe Mode Button", this); }
-
-}  // namespace esphome::safe_mode

esphome/components/safe_mode/button/safe_mode_button.h

@@ -1,19 +0,0 @@
-#pragma once
-
-#include "esphome/components/button/button.h"
-#include "esphome/components/safe_mode/safe_mode.h"
-#include "esphome/core/component.h"
-
-namespace esphome::safe_mode {
-
-class SafeModeButton final : public button::Button, public Component {
- public:
-  void dump_config() override;
-  void set_safe_mode(SafeModeComponent *safe_mode_component) { this->safe_mode_component_ = safe_mode_component; }
-
- protected:
-  SafeModeComponent *safe_mode_component_;
-  void press_action() override;
-};
-
-}  // namespace esphome::safe_mode

esphome/components/safe_mode/safe_mode.cpp

@@ -174,6 +174,14 @@ bool SafeModeComponent::get_safe_mode_pending() {
   return this->read_rtc_() == SafeModeComponent::ENTER_SAFE_MODE_MAGIC;
 }
 
+void SafeModeComponent::restart_in_safe_mode() {
+  ESP_LOGI(TAG, "Restarting in safe mode");
+  this->set_safe_mode_pending(true);
+  // Let MQTT settle a bit
+  delay(100);  // NOLINT
+  App.safe_reboot();
+}
+
 bool SafeModeComponent::should_enter_safe_mode(uint8_t num_attempts, uint32_t enable_time, uint32_t boot_is_good_after,
                                                bool in_flash) {
   this->safe_mode_start_time_ = millis();

esphome/components/safe_mode/safe_mode.h

@@ -25,6 +25,8 @@ class SafeModeComponent final : public Component {
   /// Set to true if the next startup will enter safe mode
   void set_safe_mode_pending(const bool &pending);
   bool get_safe_mode_pending();
+  /// Set safe mode pending and reboot; run by the safe mode button.
+  void restart_in_safe_mode();
 
   void dump_config() override;
   float get_setup_priority() const override;

esphome/components/shutdown/button/__init__.py

@@ -1,18 +1,13 @@
-import esphome.codegen as cg
 from esphome.components import button
-import esphome.config_validation as cv
-from esphome.const import CONF_ID, ENTITY_CATEGORY_CONFIG, ICON_POWER
+from esphome.const import ENTITY_CATEGORY_CONFIG, ICON_POWER
 from esphome.types import ConfigType
 
-shutdown_ns = cg.esphome_ns.namespace("shutdown")
-ShutdownButton = shutdown_ns.class_("ShutdownButton", button.Button, cg.Component)
-
 CONFIG_SCHEMA = button.button_schema(
-    ShutdownButton, entity_category=ENTITY_CATEGORY_CONFIG, icon=ICON_POWER
-).extend(cv.COMPONENT_SCHEMA)
+    button.ApplyButton, entity_category=ENTITY_CATEGORY_CONFIG, icon=ICON_POWER
+)
 
 
 async def to_code(config: ConfigType) -> None:
-    var = cg.new_Pvariable(config[CONF_ID])
-    await cg.register_component(var, config)
-    await button.register_button(var, config)
+    await button.sub_buttons(config).create(
+        config, None, apply="shutdown::shutdown_device()"
+    )

esphome/components/shutdown/button/shutdown_button.cpp

@@ -14,8 +14,7 @@ namespace esphome::shutdown {
 
 ESPHOME_LOG_TAG(TAG, "shutdown.button");
 
-void ShutdownButton::dump_config() { LOG_BUTTON("", "Shutdown Button", this); }
-void ShutdownButton::press_action() {
+void shutdown_device() {
   ESP_LOGI(TAG, "Shutting down");
   // Let MQTT settle a bit
   delay(100);  // NOLINT

esphome/components/shutdown/button/shutdown_button.h

@@ -1,16 +1,8 @@
 #pragma once
 
-#include "esphome/core/component.h"
-#include "esphome/components/button/button.h"
-
 namespace esphome::shutdown {
 
-class ShutdownButton final : public button::Button, public Component {
- public:
-  void dump_config() override;
-
- protected:
-  void press_action() override;
-};
+/// Run by the shutdown button.
+void shutdown_device();
 
 }  // namespace esphome::shutdown

esphome/components/spi/__init__.py

@@ -17,6 +17,7 @@ from esphome.components.esp32 import (
     VARIANT_ESP32S2,
     VARIANT_ESP32S3,
     VARIANT_ESP32S31,
+    idf_version,
     only_on_variant,
 )
 from esphome.components.psram import DOMAIN as PSRAM_DOMAIN
@@ -142,13 +143,19 @@ TYPE_CLASS = {
 }
 
 
+def _require_psram_dma_idf(value: Any) -> Any:
+    if idf_version() < cv.Version(5, 5, 3):
+        raise cv.Invalid("psram_dma requires ESP-IDF 5.5.3 or newer")
+    return value
+
+
 def _validate_psram_dma(value: Any) -> bool:
     value = cv.boolean(value)
     if not value:
         return value
     return cv.All(
         cv.only_on_esp32,
-        cv.only_with_framework("esp-idf"),
+        _require_psram_dma_idf,
         only_on_variant(
             supported=[
                 VARIANT_ESP32C5,
@@ -159,7 +166,6 @@ def _validate_psram_dma(value: Any) -> bool:
             ],
             msg_prefix="PSRAM DMA",
         ),
-        cv.require_framework_version(esp_idf=cv.Version(5, 5, 3)),
         cv.requires_component(PSRAM_DOMAIN),
     )(value)
 

esphome/components/sx126x/__init__.py

@@ -11,10 +11,9 @@ from esphome.components.const import (
     CONF_RST_PIN,
 )
 import esphome.config_validation as cv
-from esphome.const import CONF_BUSY_PIN, CONF_DATA, CONF_FREQUENCY, CONF_ID
-from esphome.core import ID, TimePeriod
-from esphome.cpp_generator import MockObj
-from esphome.types import ConfigType, TemplateArgsType
+from esphome.const import CONF_BUSY_PIN, CONF_DATA, CONF_FREQUENCY
+from esphome.core import TimePeriod
+from esphome.types import ConfigType
 
 DOMAIN = "sx126x"
 MULTI_CONF = True
@@ -133,10 +132,6 @@ SHAPING = {
     "NONE": SX126xPulseShape.NO_FILTER,
 }
 
-SendPacketAction = sx126x_ns.class_(
-    "SendPacketAction", automation.Action, cg.Parented.template(SX126x)
-)
-
 
 def validate_raw_data(value: Any) -> bytes | list[int]:
     if isinstance(value, str):
@@ -324,25 +319,8 @@ SEND_PACKET_ACTION_SCHEMA = cv.maybe_simple_value(
 )
 
 
-@automation.register_action(
+automation.register_apply_action(
     "sx126x.send_packet",
-    SendPacketAction,
     SEND_PACKET_ACTION_SCHEMA,
-    synchronous=True,
+    automation.ApplyBytes(CONF_DATA, "transmit_packet({})", stack=MAX_PACKET_SIZE),
 )
-async def send_packet_action_to_code(
-    config: ConfigType,
-    action_id: ID,
-    template_arg: cg.TemplateArguments,
-    args: TemplateArgsType,
-) -> MockObj:
-    var = cg.new_Pvariable(action_id, template_arg)
-    await cg.register_parented(var, config[CONF_ID])
-    await automation.templatable_bytes(
-        config[CONF_DATA],
-        args,
-        var.set_data_template,
-        var.set_data_static,
-        "sx126x_data",
-    )
-    return var

esphome/components/sx126x/automation.h

@@ -1,18 +0,0 @@
-#pragma once
-
-#include "esphome/core/component.h"
-#include "esphome/core/automation.h"
-#include "esphome/components/sx126x/sx126x.h"
-
-namespace esphome::sx126x {
-
-template<typename... Ts> class SendPacketAction final : public Action<Ts...>, public Parented<SX126x> {
-  TEMPLATABLE_BYTES(data)
-
-  void play(const Ts &...x) override {
-    this->data_.template visit<SX126X_MAX_PACKET_SIZE>(
-        [this](const uint8_t *data, size_t len) { this->parent_->transmit_packet(data, len); }, x...);
-  }
-};
-
-}  // namespace esphome::sx126x

esphome/components/sx127x/__init__.py

@@ -11,9 +11,7 @@ from esphome.components.const import (
     CONF_RST_PIN,
 )
 import esphome.config_validation as cv
-from esphome.const import CONF_DATA, CONF_FREQUENCY, CONF_ID
-from esphome.core import ID
-from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.const import CONF_DATA, CONF_FREQUENCY
 from esphome.types import ConfigType
 
 DOMAIN = "sx127x"
@@ -117,10 +115,6 @@ SHAPING = {
     "NONE": SX127xPaRamp.SHAPING_NONE,
 }
 
-SendPacketAction = sx127x_ns.class_(
-    "SendPacketAction", automation.Action, cg.Parented.template(SX127x)
-)
-
 
 def validate_raw_data(value: Any) -> bytes | list[int]:
     if isinstance(value, str):
@@ -291,25 +285,8 @@ SEND_PACKET_ACTION_SCHEMA = cv.maybe_simple_value(
 )
 
 
-@automation.register_action(
+automation.register_apply_action(
     "sx127x.send_packet",
-    SendPacketAction,
     SEND_PACKET_ACTION_SCHEMA,
-    synchronous=True,
+    automation.ApplyBytes(CONF_DATA, "transmit_packet({})", stack=MAX_PACKET_SIZE),
 )
-async def send_packet_action_to_code(
-    config: ConfigType,
-    action_id: ID,
-    template_arg: cg.TemplateArguments,
-    args: TemplateArgsType,
-) -> MockObj:
-    var = cg.new_Pvariable(action_id, template_arg)
-    await cg.register_parented(var, config[CONF_ID])
-    await automation.templatable_bytes(
-        config[CONF_DATA],
-        args,
-        var.set_data_template,
-        var.set_data_static,
-        "sx127x_data",
-    )
-    return var

esphome/components/sx127x/automation.h

@@ -1,19 +0,0 @@
-#pragma once
-
-#include "esphome/core/component.h"
-#include "esphome/core/automation.h"
-#include "esphome/components/sx127x/sx127x.h"
-
-namespace esphome::sx127x {
-
-template<typename... Ts> class SendPacketAction final : public Action<Ts...>, public Parented<SX127x> {
-  TEMPLATABLE_BYTES(data)
-
- public:
-  void play(const Ts &...x) override {
-    this->data_.template visit<SX127X_MAX_PACKET_SIZE>(
-        [this](const uint8_t *data, size_t len) { this->parent_->transmit_packet(data, len); }, x...);
-  }
-};
-
-}  // namespace esphome::sx127x

esphome/components/uart/__init__.py

@@ -121,7 +121,6 @@ HOST_BAUD_RATES = [
 ]
 
 UARTDevice = uart_ns.class_("UARTDevice")
-UARTWriteAction = uart_ns.class_("UARTWriteAction", automation.Action)
 UARTDebugger = uart_ns.class_("UARTDebugger", cg.Component, automation.Action)
 UARTDummyReceiver = uart_ns.class_("UARTDummyReceiver", cg.Component)
 MULTI_CONF = True
@@ -593,9 +592,8 @@ def new_uart_component(
     return cg.new_bus_component(register_uart_device, config, *args)
 
 
-@automation.register_action(
+automation.register_apply_action(
     "uart.write",
-    UARTWriteAction,
     cv.maybe_simple_value(
         {
             cv.GenerateID(): cv.use_id(UARTComponent),
@@ -603,15 +601,11 @@ def new_uart_component(
         },
         key=CONF_DATA,
     ),
-    synchronous=True,
+    # A constant stays in flash: write_array_progmem copies it out in chunks on ESP8266
+    automation.ApplyBytes(
+        CONF_DATA, "write_array({})", progmem_target="write_array_progmem({})"
+    ),
 )
-async def uart_write_to_code(config, action_id, template_arg, args):
-    var = cg.new_Pvariable(action_id, template_arg)
-    await cg.register_parented(var, config[CONF_ID])
-    await automation.templatable_bytes(
-        config[CONF_DATA], args, var.set_data_template, var.set_data_static, "uart_data"
-    )
-    return var
 
 
 @coroutine_with_priority(CoroPriority.FINAL)

esphome/components/uart/automation.h

@@ -1,22 +0,0 @@
-#pragma once
-
-#include "uart.h"
-#include "esphome/core/automation.h"
-
-#include <vector>
-
-namespace esphome::uart {
-
-template<typename... Ts> class UARTWriteAction final : public Action<Ts...>, public Parented<UARTComponent> {
-  TEMPLATABLE_BYTES(data)
-
-  void play(const Ts &...x) override {
-    if (this->data_.is_static()) {
-      this->parent_->write_array_progmem(this->data_.data(), this->data_.size());
-    } else {
-      this->parent_->write_array(this->data_.value(x...));
-    }
-  }
-};
-
-}  // namespace esphome::uart

esphome/components/udp/__init__.py

@@ -29,7 +29,6 @@ DOMAIN = "udp"
 MULTI_CONF = True
 udp_ns = cg.esphome_ns.namespace("udp")
 UDPComponent = udp_ns.class_("UDPComponent", cg.Component)
-UDPWriteAction = udp_ns.class_("UDPWriteAction", automation.Action)
 trigger_argname = "data"
 # Listener callback type (non-owning span from UDP component)
 listener_args = cg.std_span.template(cg.uint8.operator("const"))
@@ -240,17 +239,23 @@ def validate_raw_data(value: Any) -> bytes | list[int]:
     )
 
 
-@automation.register_action(
+_UDP_WRITE_SCHEMA = cv.maybe_simple_value(
+    {
+        cv.GenerateID(): cv.use_id(UDPComponent),
+        cv.Required(CONF_DATA): cv.templatable(validate_raw_data),
+    },
+    key=CONF_DATA,
+)
+_udp_write_builder = automation.apply_action_builder(
     "udp.write",
-    UDPWriteAction,
-    cv.maybe_simple_value(
-        {
-            cv.GenerateID(): cv.use_id(UDPComponent),
-            cv.Required(CONF_DATA): cv.templatable(validate_raw_data),
-        },
-        key=CONF_DATA,
-    ),
-    synchronous=True,
+    _UDP_WRITE_SCHEMA,
+    # Constant payloads up to this size are sent from the stack on ESP8266
+    automation.ApplyBytes(CONF_DATA, "send_packet({})", stack=128),
+)
+
+
+@automation.register_action(
+    "udp.write", automation.ApplyAction, _UDP_WRITE_SCHEMA, synchronous=True
 )
 async def udp_write_to_code(
     config: ConfigType,
@@ -258,11 +263,6 @@ async def udp_write_to_code(
     template_arg: cg.TemplateArguments,
     args: TemplateArgsType,
 ) -> MockObj:
-    var = cg.new_Pvariable(action_id, template_arg)
     udp_var = await cg.get_variable(config[CONF_ID])
-    await cg.register_parented(var, udp_var)
     cg.add(udp_var.set_should_broadcast())
-    await automation.templatable_bytes(
-        config[CONF_DATA], args, var.set_data_template, var.set_data_static, "udp_data"
-    )
-    return var
+    return await _udp_write_builder(config, action_id, template_arg, args)

esphome/components/udp/automation.h

@@ -1,24 +0,0 @@
-#pragma once
-
-#include "udp_component.h"
-#ifdef USE_NETWORK
-#include "esphome/core/automation.h"
-
-namespace esphome::udp {
-
-/// Constant payloads up to this size are sent from the stack on ESP8266.
-static constexpr size_t STACK_PAYLOAD_SIZE = 128;
-
-template<typename... Ts> class UDPWriteAction final : public Action<Ts...>, public Parented<UDPComponent> {
-  TEMPLATABLE_BYTES(data)
-
-  void play(const Ts &...x) override {
-    // One sendto per packet, so a failed allocation never sends a partial packet
-    this->data_.template visit<STACK_PAYLOAD_SIZE>(
-        [this](const uint8_t *data, size_t len) { this->parent_->send_packet(data, len); }, x...);
-  }
-};
-
-}  // namespace esphome::udp
-
-#endif

esphome/components/voice_assistant/__init__.py

@@ -1,12 +1,6 @@
 from esphome import automation, core
 import esphome.codegen as cg
-from esphome.components import (
-    media_player,
-    micro_wake_word,
-    microphone,
-    network,
-    speaker,
-)
+from esphome.components import media_player, micro_wake_word, microphone, speaker
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_ID,
@@ -175,7 +169,6 @@ CONFIG_SCHEMA = cv.All(
         }
     ).extend(cv.COMPONENT_SCHEMA),
     cv.key_requires((CONF_ON_TTS_STREAM_START, CONF_ON_TTS_STREAM_END), CONF_SPEAKER),
-    network.require_ipv4("voice_assistant"),
 )
 
 FINAL_VALIDATE_SCHEMA = cv.All(

esphome/components/voice_assistant/voice_assistant.cpp

@@ -82,7 +82,11 @@ void VoiceAssistant::setup() {
 float VoiceAssistant::get_setup_priority() const { return setup_priority::AFTER_CONNECTION; }
 
 bool VoiceAssistant::start_udp_socket_() {
+#if USE_NETWORK_IPV4
   this->socket_ = socket::socket(AF_INET, SOCK_DGRAM, IPPROTO_IP);
+#else
+  this->socket_ = socket::socket_ip(SOCK_DGRAM, IPPROTO_IP);
+#endif
   if (this->socket_ == nullptr) {
     ESP_LOGE(TAG, "Could not create socket");
     this->mark_failed();
@@ -663,17 +667,20 @@ void VoiceAssistant::start_streaming(struct sockaddr_storage *addr, uint16_t por
   }
 
   memcpy(&this->dest_addr_, addr, sizeof(this->dest_addr_));
-  if (this->dest_addr_.ss_family == AF_INET) {
-    ((struct sockaddr_in *) &this->dest_addr_)->sin_port = htons(port);
-  }
-#if LWIP_IPV6
-  else if (this->dest_addr_.ss_family == AF_INET6) {
-    ((struct sockaddr_in6 *) &this->dest_addr_)->sin6_port = htons(port);
-  }
+  switch (this->dest_addr_.ss_family) {
+#if USE_NETWORK_IPV4
+    case AF_INET:
+      ((struct sockaddr_in *) &this->dest_addr_)->sin_port = htons(port);
+      break;
 #endif
-  else {
-    ESP_LOGW(TAG, "Unknown address family: %d", this->dest_addr_.ss_family);
-    return;
+#if LWIP_IPV6
+    case AF_INET6:
+      ((struct sockaddr_in6 *) &this->dest_addr_)->sin6_port = htons(port);
+      break;
+#endif
+    default:
+      ESP_LOGW(TAG, "Unknown address family: %d", this->dest_addr_.ss_family);
+      return;
   }
 
   // Only primary microphone channel over UDP

esphome/components/wifi/__init__.py

@@ -224,8 +224,8 @@ WIFI_AP_MODE_WISP = "WISP"
 def _validate_ap_mode(value: str) -> str:
     value = cv.one_of(WIFI_AP_MODE_FALLBACK, WIFI_AP_MODE_WISP, upper=True)(value)
     if value == WIFI_AP_MODE_WISP:
-        # WISP mode routes through NAT, which only ESP-IDF provides
-        cv.only_with_framework(Framework.ESP_IDF)(value)
+        # The bridge is part of the ESP32 backend, on both of its frameworks
+        cv.only_on_esp32(value)
     return value
 
 

esphome/core/entity_helpers.py

@@ -533,7 +533,8 @@ class SubEntities:
         """Create the entity from ``conf``; ``setter`` may be None when the hub keeps no pointer.
 
         With ``apply`` the entity is the domain's ``Apply*`` class and its control runs
-        ``parent->apply``, with ``publish`` ("before"/"after") placing the domain's publish.
+        ``parent->apply``, or ``apply`` as a namespaced free statement when no parent is bound, with
+        ``publish`` ("before"/"after") placing the domain's publish.
         """
         entity_id: ID = conf[CONF_ID]
         is_apply_class = self.apply_entity is not None and str(entity_id.type) == str(
@@ -545,8 +546,13 @@ class SubEntities:
             raise ValueError(
                 f"{entity_id}: apply= needs the domain's Apply class, not {entity_id.type}"
             )
-        if apply is not None and self.parent is None:
-            raise ValueError(f"{entity_id}: apply= needs a parent")
+        # A parentless statement is a namespaced free function, so a hub platform that forgot
+        # parent= fails here instead of in the generated C++
+        if apply is not None and self.parent is None and "::" not in apply:
+            raise ValueError(
+                f"{entity_id}: apply= without a parent must call a namespaced free function "
+                f"(ns::fn()); bind parent= for a hub method"
+            )
         if publish is not None and (apply is None or self.apply_entity.publish is None):
             raise ValueError(f"{entity_id}: publish= needs apply= on a stateful domain")
         if publish not in (None, "before", "after"):
@@ -571,7 +577,7 @@ class SubEntities:
         if isinstance(parent, ID):
             parent = await get_variable(parent)
         apply_entity = self.apply_entity
-        body = [f"{parent_ref(parent)}->{apply};"]
+        body = [f"{apply};" if parent is None else f"{parent_ref(parent)}->{apply};"]
         if publish == "before":
             body.insert(0, apply_entity.publish)
         elif publish == "after":

esphome/core/helpers.h

@@ -1077,6 +1077,9 @@ template<typename T> ESPHOME_ALWAYS_INLINE constexpr void assign_bits(T &value,
   }
 }
 
+/// Whether bit \p bit (0 is the LSB) of \p value is set; the read side of assign_bits().
+template<typename T> ESPHOME_ALWAYS_INLINE constexpr bool bit_at(T value, unsigned bit) { return (value >> bit) & 1; }
+
 /// Reverse the order of 8 bits.
 inline uint8_t reverse_bits(uint8_t x) {
   x = ((x & 0xAA) >> 1) | ((x & 0x55) << 1);

tests/component_tests/cc1101/test_cc1101_payload_tables.py

@@ -14,7 +14,7 @@ def test_constant_payloads_share_progmem_tables(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Equal send_packet payloads share one table; lambdas stay templates."""
+    """Equal send_packet payloads share one table; a lambda is sent from its vector."""
     main_cpp = generate_main(component_config_path("payload_tables.yaml"))
 
     tables = re.findall(
@@ -23,8 +23,10 @@ def test_constant_payloads_share_progmem_tables(
     assert [v for _, v in tables] == [
         "{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08}"
     ]
-    assert main_cpp.count(f"set_data_static({tables[0][0]}, 8);") == 2
-    assert "set_data_template(" in main_cpp
+    assert main_cpp.count(f"->transmit_packet(::{tables[0][0]}, 8);") == 2
+    assert (
+        "->transmit_packet(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    )
 
 
 @pytest.mark.parametrize("size", [64, 65])

tests/component_tests/spi/test_psram_dma.py

@@ -5,6 +5,7 @@ import pytest
 from esphome import config_validation as cv
 from esphome.components.esp32 import (
     KEY_BOARD,
+    KEY_IDF_VERSION,
     KEY_VARIANT,
     VARIANT_ESP32,
     VARIANT_ESP32S3,
@@ -35,11 +36,17 @@ def _stage(
     platform_framework: PlatformFramework,
     variant: str,
     version: cv.Version,
+    idf: cv.Version | None = None,
 ) -> None:
+    """``idf`` is the ESP-IDF underneath an Arduino ``version``; an ESP-IDF build is its own."""
     set_core_config(
         platform_framework,
         core_data={KEY_FRAMEWORK_VERSION: version},
-        platform_data={KEY_BOARD: "test-board", KEY_VARIANT: variant},
+        platform_data={
+            KEY_BOARD: "test-board",
+            KEY_VARIANT: variant,
+            KEY_IDF_VERSION: idf or version,
+        },
     )
     CORE.loaded_integrations.add("psram")
 
@@ -74,17 +81,46 @@ def test_psram_dma_accepts_esp32s31(
     assert config[CONF_PSRAM_DMA] is True
 
 
-def test_psram_dma_rejects_arduino(
+def test_psram_dma_accepts_arduino_on_idf_5_5_3(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """The ESP-IDF underneath an Arduino build is what counts; 3.3.7 is the first on 5.5.3."""
+    _stage(
+        set_core_config,
+        PlatformFramework.ESP32_ARDUINO,
+        VARIANT_ESP32S3,
+        cv.Version(3, 3, 7),
+        idf=cv.Version(5, 5, 3),
+    )
+
+    config = _schema()({CONF_PSRAM_DMA: True})
+
+    assert config[CONF_PSRAM_DMA] is True
+
+
+def test_psram_dma_rejects_other_platforms(
+    set_core_config: SetCoreConfigCallable,
+) -> None:
+    """No ESP32 state to read a version from: the platform check comes first."""
+    set_core_config(PlatformFramework.ESP8266_ARDUINO)
+    CORE.loaded_integrations.add("psram")
+
+    with pytest.raises(cv.Invalid, match="only available on"):
+        _schema()({CONF_PSRAM_DMA: True})
+
+
+def test_psram_dma_rejects_older_arduino(
     set_core_config: SetCoreConfigCallable,
 ) -> None:
     _stage(
         set_core_config,
         PlatformFramework.ESP32_ARDUINO,
         VARIANT_ESP32S3,
-        cv.Version(5, 5, 3),
+        cv.Version(3, 3, 6),
+        idf=cv.Version(5, 5, 2),
     )
 
-    with pytest.raises(cv.Invalid, match="only available with framework"):
+    with pytest.raises(cv.Invalid, match="requires ESP-IDF 5.5.3"):
         _schema()({CONF_PSRAM_DMA: True})
 
 
@@ -127,7 +163,7 @@ def test_psram_dma_rejects_older_idf(
         cv.Version(5, 5, 2),
     )
 
-    with pytest.raises(cv.Invalid, match="requires at least framework version 5.5.3"):
+    with pytest.raises(cv.Invalid, match="requires ESP-IDF 5.5.3"):
         _schema()({CONF_PSRAM_DMA: True})
 
 

tests/component_tests/sx126x/test_packet_tables.py

@@ -14,7 +14,7 @@ def test_constant_packets_share_progmem_tables(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Equal constant packets share one table; lambdas stay templates."""
+    """Equal constant packets share one table; a lambda is sent from its vector."""
     main_cpp = generate_main(component_config_path("packet_tables.yaml"))
 
     tables = dict(
@@ -26,8 +26,10 @@ def test_constant_packets_share_progmem_tables(
         ["{0xC5, 0x51, 0x78, 0x82}", "{0x68, 0x69}"]
     )
     shared = next(k for k, v in tables.items() if v == "{0xC5, 0x51, 0x78, 0x82}")
-    assert main_cpp.count(f"set_data_static({shared}, 4);") == 2
-    assert "set_data_template(" in main_cpp
+    assert main_cpp.count(f"->transmit_packet(::{shared}, 4);") == 2
+    assert (
+        "->transmit_packet(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    )
     assert "set_sx1261" not in main_cpp
 
 

tests/component_tests/sx127x/test_packet_tables.py

@@ -14,7 +14,7 @@ def test_constant_packets_share_progmem_tables(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Equal constant packets share one table; lambdas stay templates."""
+    """Equal constant packets share one table; a lambda is sent from its vector."""
     main_cpp = generate_main(component_config_path("packet_tables.yaml"))
 
     tables = dict(
@@ -26,8 +26,10 @@ def test_constant_packets_share_progmem_tables(
         ["{0xC5, 0x51, 0x78, 0x82}", "{0x68, 0x69}"]
     )
     shared = next(k for k, v in tables.items() if v == "{0xC5, 0x51, 0x78, 0x82}")
-    assert main_cpp.count(f"set_data_static({shared}, 4);") == 2
-    assert "set_data_template(" in main_cpp
+    assert main_cpp.count(f"->transmit_packet(::{shared}, 4);") == 2
+    assert (
+        "->transmit_packet(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    )
 
 
 def test_packet_data_length_limit() -> None:

tests/component_tests/uart/config/payload_tables_esp8266.yaml

@@ -0,0 +1,34 @@
+---
+esphome:
+  name: test
+  on_boot:
+    then:
+      - uart.write: [0x01, 0x02, 0x03]
+      - uart.write: !lambda return {0x09};
+      - uart.write: ""
+
+esp8266:
+  board: d1_mini
+
+uart:
+  tx_pin: GPIO1
+  rx_pin: GPIO3
+  baud_rate: 9600
+
+switch:
+  - platform: uart
+    id: shared_switch
+    name: Shared Switch
+    data: [0x01, 0x02, 0x03]
+  - platform: uart
+    id: on_off_switch
+    name: On Off Switch
+    data:
+      turn_on: "ON"
+      turn_off: [0x04]
+
+button:
+  - platform: uart
+    id: shared_button
+    name: Shared Button
+    data: [0x01, 0x02, 0x03]

tests/component_tests/uart/test_payload_tables.py

@@ -23,6 +23,19 @@ def test_constant_payloads_share_progmem_tables(
     )
     assert f"shared_switch->set_data_on({shared}, 3);" in main_cpp
     assert f"shared_button->set_data({shared}, 3);" in main_cpp
-    assert f"set_data_static({shared}, 3);" in main_cpp
-    assert "set_data_static(nullptr, 0);" in main_cpp
-    assert "set_data_template(" in main_cpp
+    assert f"->write_array(::{shared}, 3);" in main_cpp
+    assert "->write_array(nullptr, 0);" in main_cpp
+    assert "->write_array(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+
+
+def test_esp8266_write_action_reads_flash_directly(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """On ESP8266 the write action hands the table to write_array_progmem instead of copying it."""
+    main_cpp = generate_main(component_config_path("payload_tables_esp8266.yaml"))
+
+    assert "->write_array_progmem(::uart_data, 3);" in main_cpp
+    assert "->write_array_progmem(nullptr, 0);" in main_cpp
+    assert "->write_array(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    assert "progmem_memcpy" not in main_cpp

tests/component_tests/udp/test_udp_write.py

@@ -8,10 +8,17 @@ def test_write_payloads_share_a_progmem_table(
     generate_main: Callable[[str | Path], str],
     component_config_path: Callable[[str], Path],
 ) -> None:
-    """Identical constant payloads share one flash table; an empty one needs none."""
+    """Identical constant payloads share one flash table and are copied out of it on
+    ESP8266 before sending; an empty one needs none; a lambda is sent from its vector."""
     main_cpp = generate_main(component_config_path("udp_write.yaml"))
 
-    assert main_cpp.count("udp_data[] PROGMEM = {0xAA, 0x55, 0x01, 0x02};") == 1
-    assert main_cpp.count("->set_data_static(udp_data, 4);") == 2
-    assert "->set_data_static(nullptr, 0);" in main_cpp
-    assert "->set_data_template(" in main_cpp
+    table = "esphome__udp_write__bytes"
+    assert main_cpp.count(f"{table}[] PROGMEM = {{0xAA, 0x55, 0x01, 0x02}};") == 1
+    assert (
+        main_cpp.count(f"esphome::progmem_memcpy(esphome__bytes, ::{table}, 4);") == 2
+    )
+    assert main_cpp.count("->send_packet(esphome__bytes, 4);") == 2
+    assert "uint8_t esphome__bytes[1];" in main_cpp
+    assert "->send_packet(esphome__bytes, 0);" in main_cpp
+    assert "->send_packet(esphome__bytes.data(), esphome__bytes.size());" in main_cpp
+    assert main_cpp.count("->set_should_broadcast();") == 4

tests/components/core/test_bit_at.cpp

@@ -0,0 +1,23 @@
+#include <gtest/gtest.h>
+
+#include "esphome/core/helpers.h"
+
+namespace esphome::core::testing {
+
+static_assert(bit_at(uint8_t{0b1010'0001}, 0));
+static_assert(!bit_at(uint8_t{0b1010'0001}, 1));
+static_assert(bit_at(uint8_t{0b1010'0001}, 7));
+static_assert(bit_at(uint64_t{1} << 63, 63));
+static_assert(!bit_at(uint64_t{1} << 63, 62));
+static_assert(bit_at(int32_t{-1}, 31));
+static_assert(!bit_at(int32_t{0x7FFF'FFFF}, 31));
+
+TEST(BitAt, IsTheReadSideOfAssignBits) {
+  uint16_t reg = 0;
+  assign_bits(reg, 1 << 9, true);
+  EXPECT_TRUE(bit_at(reg, 9));
+  assign_bits(reg, 1 << 9, false);
+  EXPECT_FALSE(bit_at(reg, 9));
+}
+
+}  // namespace esphome::core::testing

tests/components/openthread/test-voice-assistant.esp32-c6-idf.yaml

@@ -0,0 +1,36 @@
+# voice_assistant over Thread without the IPv4 stack
+packages:
+  i2s_audio: !include ../../test_build_components/common/i2s_audio/esp32-c6-idf.yaml
+
+openthread:
+  channel: 13
+  network_name: OpenThread-8f28
+  network_key: 0xdfd34f0f05cad978ec4e32b0413038ff
+  pan_id: 0x8f28
+  ext_pan_id: 0xd63e8e3e495ebbc3
+  pskc: 0xc23a76e98f1a6483639b1ac1271e2e27
+  mesh_local_prefix: fd53:145f:ed22:ad81::/64
+
+network:
+  enable_ipv6: true
+  enable_ipv4: false
+
+api:
+
+microphone:
+  - platform: i2s_audio
+    id: va_mic
+    i2s_audio_id: i2s_audio_bus
+    i2s_din_pin: GPIO7
+    adc_type: external
+
+speaker:
+  - platform: i2s_audio
+    id: va_speaker
+    i2s_audio_id: i2s_audio_bus
+    dac_type: external
+    i2s_dout_pin: GPIO10
+
+voice_assistant:
+  microphone: va_mic
+  speaker: va_speaker

tests/components/openthread/test.esp32-c6-idf.yaml

@@ -11,3 +11,6 @@ openthread:
   output_power: 1dBm
   core_options:
     OPENTHREAD_CONFIG_MAC_DEFAULT_MAX_FRAME_RETRIES_DIRECT: "5"
+
+ota:
+  - platform: esphome

tests/test_build_components/common/i2s_audio/esp32-c6-idf.yaml

@@ -0,0 +1,14 @@
+# Common I2S audio bus configuration for ESP32-C6 IDF tests
+# Provides a shared i2s_audio bus that speaker/microphone components can use
+# Each consumer must give its speaker/microphone a unique data pin
+
+substitutions:
+  i2s_bclk_pin: GPIO4
+  i2s_lrclk_pin: GPIO5
+  i2s_mclk_pin: GPIO6
+
+i2s_audio:
+  - id: i2s_audio_bus
+    i2s_bclk_pin: ${i2s_bclk_pin}
+    i2s_lrclk_pin: ${i2s_lrclk_pin}
+    i2s_mclk_pin: ${i2s_mclk_pin}

tests/unit_tests/components/test_wifi.py

@@ -236,7 +236,7 @@ def wisp_core(monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest) -
         CORE.data,
         KEY_CORE,
         {
-            KEY_TARGET_PLATFORM: Platform.ESP32,
+            KEY_TARGET_PLATFORM: params.get("platform", Platform.ESP32),
             KEY_TARGET_FRAMEWORK: params.get("framework", "esp-idf"),
             KEY_FRAMEWORK_VERSION: params.get("idf", cv.Version(5, 4, 0)),
         },
@@ -312,10 +312,14 @@ def test_second_dns_server_requires_idf_6(wisp_core: None, valid: bool) -> None:
             wifi_network_ap(ap)
 
 
-@pytest.mark.parametrize("wisp_core", [{"framework": "arduino"}], indirect=True)
-def test_wisp_requires_esp_idf(wisp_core: None) -> None:
-    """NAT is only available with the ESP-IDF framework."""
-    with pytest.raises(cv.Invalid, match="only available with framework"):
+@pytest.mark.parametrize(
+    "wisp_core",
+    [{"platform": Platform.ESP8266, "framework": "arduino"}],
+    indirect=True,
+)
+def test_wisp_requires_esp32(wisp_core: None) -> None:
+    """The bridge is part of the ESP32 backend; other platforms have no NAT."""
+    with pytest.raises(cv.Invalid, match="only available on .*ESP32"):
         wifi_network_ap(_wisp_ap())
 
 

tests/unit_tests/core/test_entity_helpers.py

@@ -1536,13 +1536,87 @@ async def test_sub_entities_apply_errors(
 
 
 @pytest.mark.asyncio
-async def test_sub_entities_apply_needs_parent(
+async def test_sub_entities_apply_without_parent_needs_namespaced_call(
+    setup_test_environment: list[str],
+    apply_functions: list[tuple[str, str, str]],
+) -> None:
+    """A hub method with no parent bound fails in Python, not as an undeclared name in C++."""
+    calls: list[tuple[Any, ...]] = []
+    conf = _apply_conf("k_id", _BUTTON_APPLY.cls)
+    sub = _apply_sub(calls, {"k": conf}, _BUTTON_APPLY, parent=None)
+    with pytest.raises(ValueError, match="namespaced free function"):
+        await sub("k", None, apply="restart()")
+    assert calls == []
+    assert apply_functions == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("apply_entity", "type_", "publish", "params", "body"),
+    [
+        (
+            _BUTTON_APPLY,
+            _BUTTON_APPLY.cls,
+            None,
+            "button::Button * entity",
+            "restart::restart_device();",
+        ),
+        (
+            _SWITCH_APPLY,
+            _APPLY_SWITCH_CLS,
+            "after",
+            "switch_::Switch * entity, bool state",
+            "restart::restart_device();\nentity->publish_state(state);",
+        ),
+    ],
+)
+async def test_sub_entities_apply_without_parent(
+    monkeypatch: pytest.MonkeyPatch,
+    setup_test_environment: list[str],
+    apply_functions: list[tuple[str, str, str]],
+    apply_entity: entity_helpers.ApplyEntity,
+    type_: MockObjClass,
+    publish: str | None,
+    params: str,
+    body: str,
+) -> None:
+    """Without a parent the statement runs as written, with no parent lookup or deref."""
+
+    async def fail(*args: Any) -> None:
+        raise AssertionError("must not be called")
+
+    monkeypatch.setattr(entity_helpers.cg, "register_parented", fail)
+    monkeypatch.setattr(entity_helpers, "get_variable", fail)
+    calls: list[tuple[Any, ...]] = []
+    sub = _apply_sub(calls, {}, apply_entity, parent=None)
+    var = await sub.create(
+        _apply_conf("rst", type_),
+        None,
+        apply="restart::restart_device()",
+        publish=publish,
+    )
+    assert str(var) == "rst"
+    assert calls == [("rst", "<esphome__rst__fn>")]
+    assert apply_functions == [("esphome__rst__fn", params, body)]
+    assert setup_test_environment == []
+
+
+@pytest.mark.asyncio
+async def test_sub_entities_apply_without_parent_shares_equal_functions(
     setup_test_environment: list[str], apply_functions: list[tuple[str, str, str]]
 ) -> None:
+    """Parentless entities with the same statement share one function."""
     calls: list[tuple[Any, ...]] = []
-    sub = _apply_sub(calls, {"k": _apply_conf("k_id")}, parent=None)
-    with pytest.raises(ValueError, match="needs a parent"):
-        await sub("k", None, apply="x()")
+    sub = _apply_sub(calls, {}, _BUTTON_APPLY, parent=None)
+    for name in ("a_btn", "b_btn"):
+        await sub.create(_apply_conf(name, _BUTTON_APPLY.cls), None, apply="x::go()")
+    await sub.create(_apply_conf("c_btn", _BUTTON_APPLY.cls), None, apply="x::stop()")
+    assert calls == [
+        ("a_btn", "<esphome__a_btn__fn>"),
+        ("b_btn", "<esphome__a_btn__fn>"),
+        ("c_btn", "<esphome__c_btn__fn>"),
+    ]
+    assert [body for _, _, body in apply_functions] == ["x::go();", "x::stop();"]
 
 
 @pytest.mark.parametrize(

tests/unit_tests/test_automation.py

@@ -1319,25 +1319,52 @@ async def test_apply_string_ref_literal_stays_in_flash_on_esp8266(
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("platform", ["esp32", "esp8266"])
+@pytest.mark.parametrize(
+    "body", ["x *= 2;\nreturn {0x01, x};", "return std::vector<uint8_t>{++x};"]
+)
 async def test_apply_bytes_lambda_is_a_vector_in_a_block(
-    registries: tuple[Registry, Registry], mock_cg: MockCodegen, platform: str
+    registries: tuple[Registry, Registry],
+    mock_cg: MockCodegen,
+    platform: str,
+    body: str,
 ) -> None:
-    """A lambda payload is called once into a scoped vector and passed as data and size."""
+    """A lambda payload is called once into a scoped vector and passed as data and size;
+    it takes the trigger args by their own types and is never reduced, so it may modify them."""
     fields = (ApplyBytes("data", "send({})"),)
-    config = {"data": Lambda("return {0x01, x};")}
+    config = {"data": Lambda(body)}
     await _run_apply_action(
         registries, fields, config, args=[(cg.uint8, "x")], platform=platform
     )
     text = _apply_definition(mock_cg)
-    # Parenthesized so a reduced `return a, b;` stays one initializer
-    assert "const std::vector<uint8_t> esphome__bytes = (" in text
-    assert "-> std::vector<uint8_t>" in text
+    assert text.startswith(
+        "static void esphome__obj_1__fn(const std::remove_cvref_t<uint8_t> & x) {"
+    )
+    assert (
+        "const std::vector<uint8_t> esphome__bytes = [](uint8_t x) -> std::vector<uint8_t> {"
+        in text
+    )
+    assert "}(x);" in text
     assert (
         f"::{PARENT_OBJ}->send(esphome__bytes.data(), esphome__bytes.size());" in text
     )
     assert "progmem_memcpy" not in text
 
 
+@pytest.mark.asyncio
+async def test_apply_bytes_keeps_a_mutable_reference_arg(
+    registries: tuple[Registry, Registry], mock_cg: MockCodegen
+) -> None:
+    """A reference trigger arg such as climate's `ClimateCall &` is passed through as is."""
+    fields = (ApplyBytes("data", "send({})"),)
+    config = {"data": Lambda("return {0x01};")}
+    call_ref = cg.MockObj("ClimateCall").operator("ref")
+    await _run_apply_action(registries, fields, config, args=[(call_ref, "x")])
+    text = _apply_definition(mock_cg)
+    assert text.startswith("static void esphome__obj_1__fn(ClimateCall & x) {")
+    assert "[](ClimateCall & x) -> std::vector<uint8_t> {" in text
+    assert "}(x);" in text
+
+
 @pytest.mark.asyncio
 async def test_apply_bytes_constant_is_a_shared_table(
     registries: tuple[Registry, Registry], mock_cg: MockCodegen