[sx126x][sx127x] Require rx_start for layers that need replies

- Reject a radio with rx_start false when the link or a peer has a layer
  that needs packets back from the peer, as the radio sleeps after sending
- Log a send over a long busy period at debug level, so a lost reception
  can be traced

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

esphome/components/packet_interface/radio_link.cpp

@@ -21,7 +21,7 @@ bool RadioLinkBase::may_send_(bool busy) {
   if (!this->listen_before_talk_.may_send(busy, now))
     return false;
   if (busy) {
-    ESP_LOGV(TAG, "Radio busy for %" PRIu32 "ms, sending anyway", this->listen_before_talk_.busy_for(now));
+    ESP_LOGD(TAG, "Radio busy for %" PRIu32 "ms, sending anyway", this->listen_before_talk_.busy_for(now));
   }
   return true;
 }

esphome/components/packet_interface/radio_link.py

@@ -13,6 +13,7 @@ import esphome.final_validate as fv
 from esphome.types import ConfigType
 
 from . import new_packet_interface, packet_interface_schema
+from .layers import CONF_ADDRESSING, CONF_PEERS, STACK_ORDER
 
 # The radios' payload length register is 8 bits
 MAX_PACKET_SIZE = 255
@@ -77,12 +78,26 @@ def _radio_config(config: ConfigType, radio_id_key: str) -> ConfigType:
     )
 
 
+def _two_way_layer(config: ConfigType) -> str | None:
+    """The first layer on the link or in a peer that keeps state for one peer, so needs packets back from it."""
+    for stack in (config, *config.get(CONF_ADDRESSING, {}).get(CONF_PEERS, [])):
+        for kind in STACK_ORDER:
+            if (layer := stack.get(kind.key)) is not None and kind.get_type(
+                layer
+            ).single_peer(layer):
+                return kind.key
+    return None
+
+
 def radio_link_final_validate(
     radio_id_key: str,
     payload_length_key: str,
+    rx_start_key: str,
     check: Callable[[ConfigType, ConfigType], None] | None = None,
 ) -> Callable[[ConfigType], ConfigType]:
-    """Final validation for a radio link: packets vary in length, and check(link, radio) checks the rest."""
+    """Final validation for a radio link: packets vary in length, the radio listens after sending when a layer needs
+    replies, and check(link, radio) checks the rest.
+    """
 
     def validator(config: ConfigType) -> ConfigType:
         radio = _radio_config(config, radio_id_key)
@@ -91,6 +106,12 @@ def radio_link_final_validate(
                 f"Packets vary in length, so the radio's '{payload_length_key}' must be 0",
                 path=[radio_id_key],
             )
+        if not radio[rx_start_key] and (layer := _two_way_layer(config)) is not None:
+            raise cv.Invalid(
+                f"This '{layer}' needs packets from the peer, but the radio sleeps after sending unless "
+                f"'{rx_start_key}' is true",
+                path=[radio_id_key],
+            )
         if check is not None:
             check(config, radio)
         return config

esphome/components/sx126x/packet_interface/__init__.py

@@ -8,6 +8,7 @@ from esphome.types import ConfigType
 
 from .. import (
     CONF_PAYLOAD_LENGTH,
+    CONF_RX_START,
     CONF_SX126X_ID,
     DOMAIN,
     SX126x,
@@ -24,7 +25,9 @@ SX126xPacketInterface = sx126x_ns.class_(
 
 CONFIG_SCHEMA = radio_link_schema(SX126xPacketInterface, CONF_SX126X_ID, SX126x)
 
-FINAL_VALIDATE_SCHEMA = radio_link_final_validate(CONF_SX126X_ID, CONF_PAYLOAD_LENGTH)
+FINAL_VALIDATE_SCHEMA = radio_link_final_validate(
+    CONF_SX126X_ID, CONF_PAYLOAD_LENGTH, CONF_RX_START
+)
 
 
 async def to_code(config: ConfigType) -> None:

esphome/components/sx127x/packet_interface/__init__.py

@@ -12,6 +12,7 @@ from .. import (
     CONF_MODULATION,
     CONF_PACKET_MODE,
     CONF_PAYLOAD_LENGTH,
+    CONF_RX_START,
     CONF_SX127X_ID,
     DOMAIN,
     SX127x,
@@ -49,7 +50,7 @@ def _check_fsk(config: ConfigType, radio: ConfigType) -> None:
 
 
 FINAL_VALIDATE_SCHEMA = radio_link_final_validate(
-    CONF_SX127X_ID, CONF_PAYLOAD_LENGTH, _check_fsk
+    CONF_SX127X_ID, CONF_PAYLOAD_LENGTH, CONF_RX_START, _check_fsk
 )
 
 

tests/unit_tests/test_lora_packet_interface.py

@@ -74,7 +74,7 @@ class _FullConfig(dict):
     """Enough of the full configuration for the radio links' final validation."""
 
     def __init__(self, domain: str, radio: dict) -> None:
-        super().__init__({domain: [radio]})
+        super().__init__({domain: [{"rx_start": True, **radio}]})
         self.domain = domain
 
     def get_path_for_id(self, id_: object) -> list:
@@ -140,3 +140,38 @@ def test_radio_settings_checked(
             assert module.FINAL_VALIDATE_SCHEMA(config)
     finally:
         fv.full_config.reset(token)
+
+
+@pytest.mark.parametrize(
+    ("link", "match"),
+    [
+        ({}, None),
+        ({"encryption": "secret"}, None),
+        ({"reliability": {}}, "reliability"),
+        (
+            {
+                "addressing": {
+                    "address": 1,
+                    "peers": [
+                        {"id": "a", "destination": 2, "session": {}},
+                        {"id": "b", "destination": 3},
+                    ],
+                }
+            },
+            "session",
+        ),
+    ],
+)
+def test_rx_start_needed_for_replies(setup_core, link: dict, match: str | None) -> None:
+    """A radio that sleeps after sending suits only a link whose layers need no packets back."""
+    sleeping = {"payload_length": 0, "rx_start": False}
+    config = sx126x_packet_interface.CONFIG_SCHEMA({"sx126x_id": "radio", **link})
+    token = fv.full_config.set(_FullConfig("sx126x", sleeping))
+    try:
+        if match is None:
+            assert sx126x_packet_interface.FINAL_VALIDATE_SCHEMA(config)
+            return
+        with pytest.raises(cv.Invalid, match=f"'{match}'.*rx_start"):
+            sx126x_packet_interface.FINAL_VALIDATE_SCHEMA(config)
+    finally:
+        fv.full_config.reset(token)