[tcp] Fixes from code review

- Publish the connected sensor as off at setup, so a link that never
  comes up does not show as unknown
- Add a timeout option, defaulting to three keep-alive intervals and
  longer than one, as the peer's keep_alive sets how often packets
  arrive; limit keep_alive to 1 h and timeout to 3 h
- Load binary_sensor for an empty config from tooling
- Name the layer in the error when a provided layer is configured

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

esphome/components/packet_interface/__init__.py

@@ -140,7 +140,12 @@ def packet_interface_schema(
 ) -> cv.Schema:
     """Schema for a link platform; provides names the layer keys, such as reliability, that the link does itself."""
     link_layers = {
-        key: schema for key, schema in LINK_LAYER_SCHEMAS.items() if key not in provides
+        key: (
+            cv.invalid(f"This link provides '{key}' itself, so it cannot be configured")
+            if key in provides
+            else schema
+        )
+        for key, schema in LINK_LAYER_SCHEMAS.items()
     }
     return (
         cv.Schema(

esphome/components/tcp/packet_interface/__init__.py

@@ -22,6 +22,7 @@ import esphome.config_validation as cv
 from esphome.const import (
     CONF_ID,
     CONF_PORT,
+    CONF_TIMEOUT,
     DEVICE_CLASS_CONNECTIVITY,
     ENTITY_CATEGORY_DIAGNOSTIC,
 )
@@ -33,7 +34,7 @@ DEPENDENCIES = ["network"]
 
 def AUTO_LOAD(config: ConfigType | None = None) -> list[str]:
     """The binary_sensor component only for the connected sensor; without a config, as tooling calls it, both."""
-    if config is None or CONF_CONNECTED in config:
+    if not config or CONF_CONNECTED in config:
         return ["socket", "binary_sensor"]
     return ["socket"]
 
@@ -46,6 +47,19 @@ TcpPacketInterface = tcp_ns.class_(
     "TcpPacketInterface", packet_interface.PacketInterface
 )
 
+
+def _timeout(config: ConfigType) -> ConfigType:
+    """Default the timeout to three keep-alive intervals, and require it to be longer than one."""
+    keep_alive = config[CONF_KEEP_ALIVE].total_milliseconds
+    if (timeout := config.get(CONF_TIMEOUT)) is None:
+        config[CONF_TIMEOUT] = cv.positive_time_period_milliseconds(
+            f"{keep_alive * 3}ms"
+        )
+    elif timeout.total_milliseconds <= keep_alive:
+        raise cv.Invalid("Must be longer than keep_alive", path=[CONF_TIMEOUT])
+    return config
+
+
 BASE_SCHEMA = packet_interface_schema(
     TcpPacketInterface,
     MAX_PACKET_SIZE,
@@ -60,7 +74,12 @@ BASE_SCHEMA = packet_interface_schema(
         ): cv.positive_time_period_milliseconds,
         cv.Optional(CONF_KEEP_ALIVE, default="10s"): cv.All(
             cv.positive_time_period_milliseconds,
-            cv.Range(min=cv.TimePeriod(seconds=1)),
+            cv.Range(min=cv.TimePeriod(seconds=1), max=cv.TimePeriod(hours=1)),
+        ),
+        # Nothing received for this long closes the connection; it must be longer than the peer's keep_alive
+        cv.Optional(CONF_TIMEOUT): cv.All(
+            cv.positive_time_period_milliseconds,
+            cv.Range(max=cv.TimePeriod(hours=3)),
         ),
         cv.Optional(CONF_CONNECTED): binary_sensor.binary_sensor_schema(
             device_class=DEVICE_CLASS_CONNECTIVITY,
@@ -72,9 +91,15 @@ BASE_SCHEMA = packet_interface_schema(
 CONFIG_SCHEMA = cv.All(
     cv.typed_schema(
         {
-            "client": BASE_SCHEMA.extend({cv.Required(CONF_HOST): socket.ipv4_host}),
-            "server": BASE_SCHEMA.extend(
-                {cv.Optional(CONF_ALLOWED_IPS): socket.IPV4_ALLOW_SCHEMA}
+            "client": cv.All(
+                BASE_SCHEMA.extend({cv.Required(CONF_HOST): socket.ipv4_host}),
+                _timeout,
+            ),
+            "server": cv.All(
+                BASE_SCHEMA.extend(
+                    {cv.Optional(CONF_ALLOWED_IPS): socket.IPV4_ALLOW_SCHEMA}
+                ),
+                _timeout,
             ),
         },
         key=CONF_ROLE,
@@ -104,5 +129,6 @@ async def to_code(config: ConfigType) -> None:
     cg.add(var.set_port(config[CONF_PORT]))
     cg.add(var.set_reconnect_interval(config[CONF_RECONNECT_INTERVAL]))
     cg.add(var.set_keep_alive(config[CONF_KEEP_ALIVE]))
+    cg.add(var.set_receive_timeout(config[CONF_TIMEOUT]))
     binary_sensors = binary_sensor.sub_binary_sensors(config)
     await binary_sensors(CONF_CONNECTED, var.set_connected_sensor)

esphome/components/tcp/packet_interface/tcp_packet_interface.cpp

@@ -12,8 +12,6 @@ ESPHOME_LOG_TAG(TAG, "tcp.packet_interface");
 
 // The packet length before each packet
 static constexpr size_t LENGTH_SIZE = 2;
-// The peer is gone when nothing has been received for this many keep-alive intervals
-static constexpr uint32_t KEEP_ALIVE_MISSES = 3;
 
 TcpPacketInterface::TcpPacketInterface(std::span<uint8_t> rx_buffer)
     : packet_interface::PacketInterface(rx_buffer.size()), rx_buffer_(rx_buffer) {
@@ -26,6 +24,10 @@ void TcpPacketInterface::setup() {
 #ifdef USE_SOCKET_TCP_LISTENER
   this->listener_.begin(TAG);
 #endif
+#ifdef USE_BINARY_SENSOR
+  if (this->connected_sensor_ != nullptr)
+    this->connected_sensor_->publish_initial_state(false);
+#endif
 }
 
 void TcpPacketInterface::dump_config() {
@@ -39,10 +41,11 @@ void TcpPacketInterface::dump_config() {
                 "  Address: %s:%u\n"
                 "  Reconnect Interval: %" PRIu32 "ms\n"
                 "  Keep Alive: %" PRIu32 "ms\n"
+                "  Timeout: %" PRIu32 "ms\n"
                 "  Max Packet Size: %zu",
                 this->name_, server ? LOG_STR_LITERAL("server") : LOG_STR_LITERAL("client"),
                 server ? LOG_STR_LITERAL("*") : this->link_.host(), this->link_.port(),
-                this->link_.reconnect_interval(), this->keep_alive_ms_, this->max_packet_size_);
+                this->link_.reconnect_interval(), this->keep_alive_ms_, this->timeout_ms_, this->max_packet_size_);
 #ifdef USE_SOCKET_TCP_LISTENER
   this->listener_.dump_config();
 #endif
@@ -77,9 +80,8 @@ void TcpPacketInterface::loop() {
   if (!this->connected_)
     return;
   uint32_t now = App.get_loop_component_start_time();
-  if (now - this->last_rx_ms_ > KEEP_ALIVE_MISSES * this->keep_alive_ms_) {
-    ESP_LOGW(TAG, "%s: Nothing received for %" PRIu32 "ms, closing the connection", this->name_,
-             KEEP_ALIVE_MISSES * this->keep_alive_ms_);
+  if (now - this->last_rx_ms_ > this->timeout_ms_) {
+    ESP_LOGW(TAG, "%s: Nothing received for %" PRIu32 "ms, closing the connection", this->name_, this->timeout_ms_);
     this->link_.close();
     this->link_.note_attempt();
     this->on_connection_(false);

esphome/components/tcp/packet_interface/tcp_packet_interface.h

@@ -27,7 +27,7 @@ namespace esphome::tcp {
  *
  * A zero length packet is a keep-alive, sent when nothing else has been sent for keep_alive. TCP can take minutes to
  * notice a peer that has gone without closing the connection, so the link closes it, and reconnects or accepts again,
- * when nothing has been received for three keep-alive intervals.
+ * when nothing has been received for the timeout, which must be longer than the peer's keep-alive interval.
  */
 class TcpPacketInterface : public packet_interface::PacketInterface {
  public:
@@ -45,6 +45,8 @@ class TcpPacketInterface : public packet_interface::PacketInterface {
   void set_port(uint16_t port) { this->link_.set_port(port); }
   void set_reconnect_interval(uint32_t interval_ms) { this->link_.set_reconnect_interval(interval_ms); }
   void set_keep_alive(uint32_t keep_alive_ms) { this->keep_alive_ms_ = keep_alive_ms; }
+  /// The connection is closed when nothing has been received for this long; the peer's keep_alive sets the pace.
+  void set_receive_timeout(uint32_t timeout_ms) { this->timeout_ms_ = timeout_ms; }
 #ifdef USE_SOCKET_TCP_LISTENER
   void set_server(bool server) { this->server_ = server; }
 #ifdef USE_SOCKET_IPV4_ALLOW
@@ -72,6 +74,7 @@ class TcpPacketInterface : public packet_interface::PacketInterface {
 #endif
   const std::span<uint8_t> rx_buffer_;
   uint32_t keep_alive_ms_{10000};
+  uint32_t timeout_ms_{30000};
   uint32_t last_rx_ms_{0};
   uint32_t last_tx_ms_{0};
   uint16_t rx_length_{0};

tests/components/tcp/__init__.py

@@ -1,4 +1,5 @@
 from esphome.components import socket as socket_component
+from esphome.core import CORE
 from esphome.types import ConfigType
 from tests.testing_helpers import ComponentManifestOverride
 
@@ -6,6 +7,12 @@ from tests.testing_helpers import ComponentManifestOverride
 def override_manifest(manifest: ComponentManifestOverride) -> None:
     async def to_code_testing(config: ConfigType) -> None:
         socket_component.require_tcp_listener()
+        # No entity platforms are configured, so count the connected sensor to emit USE_BINARY_SENSOR
+        CORE.platform_counts["binary_sensor"] += 1
 
     manifest.to_code = to_code_testing
-    manifest.dependencies = manifest.dependencies + ["packet_interface", "socket"]
+    manifest.dependencies = manifest.dependencies + [
+        "binary_sensor",
+        "packet_interface",
+        "socket",
+    ]

tests/components/tcp/tcp_packet_interface.cpp

@@ -140,11 +140,35 @@ TEST_F(TcpPacketInterfaceTest, KeepAliveKeepsIdleLinkUp) {
 
 TEST_F(TcpPacketInterfaceTest, SilentPeerClosesLink) {
   set_time(1000 + 30001);
-  this->a.link.loop();  // b has sent nothing for three keep-alive intervals
+  this->a.link.loop();  // b has sent nothing for the default timeout of 30 s
   EXPECT_FALSE(this->a.link.is_connected());
   EXPECT_EQ(this->a.states, (std::vector<bool>{true, false}));
 }
 
+TEST_F(TcpPacketInterfaceTest, SlowerPeerMatchedByTimeout) {
+  // b sends keep-alives every 20 s; three of a's own 5 s intervals would close the link
+  this->a.link.set_keep_alive(5000);
+  this->a.link.set_receive_timeout(60000);
+  this->b.link.set_keep_alive(20000);
+  this->b.link.set_receive_timeout(60000);
+  for (uint32_t now = 1000; now <= 121000; now += 5000) {
+    set_time(now);
+    this->loops(2);
+  }
+  EXPECT_TRUE(this->a.link.is_connected());
+  EXPECT_TRUE(this->b.link.is_connected());
+  EXPECT_EQ(this->a.states, std::vector<bool>{true});
+}
+
+TEST_F(TcpPacketInterfaceTest, ConnectedSensorStartsFalse) {
+  TestTcp link;
+  binary_sensor::BinarySensor sensor;
+  link.set_connected_sensor(&sensor);
+  link.setup();
+  EXPECT_TRUE(sensor.has_state());
+  EXPECT_FALSE(sensor.state);
+}
+
 TEST_F(TcpPacketInterfaceTest, OversizedLengthClosesLink) {
   End c;
   int fds[2];

tests/unit_tests/test_tcp_packet_interface.py

@@ -22,7 +22,7 @@ def test_client_and_server(setup_core) -> None:
 @pytest.mark.parametrize("layer", ["reliability", "session", "addressing"])
 def test_provided_layers_rejected(setup_core, layer: str) -> None:
     """TCP provides these itself, so they cannot be configured on it."""
-    with pytest.raises(cv.Invalid, match=layer):
+    with pytest.raises(cv.Invalid, match=f"provides '{layer}' itself"):
         tcp_packet_interface.CONFIG_SCHEMA(
             {"host": "192.168.1.50", "port": 6060, layer: {"address": 1}}
         )
@@ -31,12 +31,31 @@ def test_provided_layers_rejected(setup_core, layer: str) -> None:
 def test_auto_load_binary_sensor_for_connected(setup_core) -> None:
     """The binary_sensor component is loaded only for the connected sensor."""
     assert tcp_packet_interface.AUTO_LOAD() == ["socket", "binary_sensor"]
+    assert tcp_packet_interface.AUTO_LOAD({}) == ["socket", "binary_sensor"]
     config = {"host": "192.168.1.50", "port": 6060}
     assert tcp_packet_interface.AUTO_LOAD(config) == ["socket"]
     config["connected"] = {"name": "Link"}
     assert tcp_packet_interface.AUTO_LOAD(config) == ["socket", "binary_sensor"]
 
 
+def test_timeout(setup_core) -> None:
+    """The timeout follows keep_alive unless given, and must be longer than it."""
+    base = {"host": "192.168.1.50", "port": 6060}
+    config = tcp_packet_interface.CONFIG_SCHEMA({**base, "keep_alive": "2s"})
+    assert config["timeout"].total_milliseconds == 6000
+    # A peer that sends keep-alives less often is matched with a longer timeout
+    config = tcp_packet_interface.CONFIG_SCHEMA({**base, "timeout": "90s"})
+    assert config["timeout"].total_milliseconds == 90000
+    with pytest.raises(cv.Invalid, match="longer than keep_alive"):
+        tcp_packet_interface.CONFIG_SCHEMA(
+            {**base, "keep_alive": "5s", "timeout": "5s"}
+        )
+    with pytest.raises(cv.Invalid):
+        tcp_packet_interface.CONFIG_SCHEMA({**base, "keep_alive": "2h"})
+    with pytest.raises(cv.Invalid):
+        tcp_packet_interface.CONFIG_SCHEMA({**base, "timeout": "4h"})
+
+
 @pytest.mark.parametrize(
     ("role", "tcp", "listen"), [("client", 1, 0), ("server", 1, 1)]
 )