[esphome.ota] Fix cleanup race, tighten error message and comments, add tests

Addresses copilot review on #15636.

1. Fix cleanup_connection_() race with queued listener events.
   While an OTA session was active, a second incoming connection would
   fire esphome_socket_event_callback → esphome_wake_ota_component_any_context,
   which sets pending_enable_loop_ on the (still-active) OTA component.
   enable_pending_loops_() only scans the inactive section, so that flag
   goes invisible. When cleanup_connection_() then called disable_loop(),
   the component dropped to LOOP_DONE with a stale pending flag and
   nothing to re-trigger the scan — the queued client sat forever until
   some unrelated socket activity woke the main loop.

   Fix: don't call disable_loop() from cleanup_connection_(). loop() has
   the idempotent idle check at its top; one more dispatch after cleanup
   is cheap and guarantees we re-read server_->ready() and either accept
   the queued client or disable cleanly.

2. Tighten the multi-port error message. Merging is fine — the constraint
   is single-port. Reworded: "Only a single port is supported for 'ota'
   'platform: esphome'. Got ports [...]. Consolidate onto a single port;
   configs sharing a port are merged automatically."

3. Comment drift: three call sites and the fast-select extern declaration
   still referred to enable_loop_soon_any_context() and implied the hook
   wakes the main loop. Updated to reflect the current mechanism (sets
   pending-enable flags only; callers have already woken the main loop).
   Also clarified that esphome_wake_ota_component_any_context fires on
   every RCVPLUS event across all monitored sockets, so false wakes are
   expected and OTA::loop() disables itself again when idle.

4. Added tests/component_tests/ota/test_esphome_ota.py covering
   ota_esphome_final_validate: single instance accepted, same-port
   configs merge, different-port configs rejected with cv.Invalid,
   non-esphome platforms unaffected.

esphome/components/esphome/ota/__init__.py

@@ -80,10 +80,10 @@ def ota_esphome_final_validate(config):
 
     if len(merged_ota_esphome_configs_by_port) > 1:
         raise cv.Invalid(
-            f"Only a single '{CONF_OTA}' '{CONF_PLATFORM}: {CONF_ESPHOME}' instance is "
-            f"supported, but multiple were configured on different ports "
-            f"({sorted(merged_ota_esphome_configs_by_port.keys())}). Remove the extra "
-            f"configurations or place them on the same port so they can be merged."
+            f"Only a single port is supported for '{CONF_OTA}' "
+            f"'{CONF_PLATFORM}: {CONF_ESPHOME}'. Got ports "
+            f"{sorted(merged_ota_esphome_configs_by_port.keys())}. Consolidate "
+            f"onto a single port; configs sharing a port are merged automatically."
         )
 
     new_ota_conf.extend(merged_ota_esphome_configs_by_port.values())

esphome/components/esphome/ota/ota_esphome.cpp

@@ -577,8 +577,12 @@ void ESPHomeOTAComponent::cleanup_connection_() {
 #ifdef USE_OTA_PASSWORD
   this->cleanup_auth_();
 #endif
-  // Back to idle — sleep until the next incoming connection wakes us.
-  this->disable_loop();
+  // Do not disable_loop() here. loop() itself disables when idle. If a second
+  // connection was queued on the listener while we were busy, the wake flag was
+  // set while this component was in LOOP state — enable_pending_loops_() only
+  // scans the inactive section and would never clear it. Letting loop() run one
+  // more iteration guarantees we re-check server_->ready() and either accept the
+  // queued client or disable ourselves cleanly.
 }
 
 void ESPHomeOTAComponent::yield_and_feed_watchdog_() {

esphome/components/socket/lwip_raw_tcp_impl.cpp

@@ -860,9 +860,11 @@ err_t LWIPRawListenImpl::accept_fn_(struct tcp_pcb *newpcb, err_t err) {
   // Wake the main loop immediately so it can accept the new connection.
   esphome::wake_loop_any_context();
 #ifdef USE_OTA
-  // Re-enable the OTA component loop if it disabled itself while idle.
-  // enable_loop_soon_any_context() is IRAM/IRQ-safe, which is required on RP2040
-  // where this callback runs in a low-priority user IRQ context.
+  // Mark the OTA component loop to be re-enabled if it disabled itself while idle.
+  // This only sets pending-enable flags; the wake_loop_any_context() call above has
+  // already woken the main loop, which will process the pending enable on its next
+  // iteration. Safe to call from RP2040's low-priority user IRQ context — it only
+  // writes volatile bools, no heap or locks.
   esphome::App.wake_ota_component_any_context();
 #endif
   return ERR_OK;

esphome/core/application.cpp

@@ -451,7 +451,8 @@ void Application::enable_pending_loops_() {
 
 #if defined(USE_OTA) && defined(USE_LWIP_FAST_SELECT)
 // Called from the LwIP TCP/IP task via esphome_socket_event_callback() on NETCONN_EVT_RCVPLUS.
-// enable_loop_soon_any_context() is task-safe and IRAM-resident.
+// Only marks the OTA component as pending loop-enable; the fast-select callback itself has
+// already woken the main task via xTaskNotifyGive().
 extern "C" void IRAM_ATTR esphome_wake_ota_component_any_context() { App.wake_ota_component_any_context(); }
 #endif
 

esphome/core/lwip_fast_select.c

@@ -159,8 +159,10 @@ static netconn_callback s_original_callback = NULL;
 
 #ifdef USE_OTA
 // Extern wake hook for the OTA component (implemented in application.cpp). Called from the
-// TCP/IP task so the OTA component's disabled loop can be re-enabled when a new connection
-// arrives on its listening socket. Safe from task context via enable_loop_soon_any_context().
+// TCP/IP task on every NETCONN_EVT_RCVPLUS — not just OTA's listener, so this can be a false
+// wake from an unrelated monitored socket. OTA::loop() handles that by disabling itself again
+// when there is no pending work. The hook only marks the OTA component as pending loop-enable;
+// it does not itself wake the main task (the caller below already does that).
 extern void esphome_wake_ota_component_any_context(void);
 #endif
 
@@ -183,7 +185,9 @@ static void esphome_socket_event_callback(struct netconn *conn, enum netconn_evt
       xTaskNotifyGive(task);
     }
 #ifdef USE_OTA
-    // Re-enable the OTA component loop if it disabled itself while idle.
+    // Mark the OTA component loop to be re-enabled if it disabled itself while idle.
+    // Only sets pending-enable flags — the xTaskNotifyGive above has already woken
+    // the main task, which will process the pending enable on its next iteration.
     esphome_wake_ota_component_any_context();
 #endif
   }

tests/component_tests/ota/test_esphome_ota.py

@@ -0,0 +1,105 @@
+"""Tests for the esphome OTA platform final_validate logic."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import pytest
+
+from esphome import config_validation as cv
+from esphome.components.esphome.ota import ota_esphome_final_validate
+from esphome.const import (
+    CONF_ESPHOME,
+    CONF_ID,
+    CONF_OTA,
+    CONF_PASSWORD,
+    CONF_PLATFORM,
+    CONF_PORT,
+    CONF_VERSION,
+)
+from esphome.core import ID
+import esphome.final_validate as fv
+
+
+def _make_ota_config(port: int = 3232, **kwargs: Any) -> dict[str, Any]:
+    config: dict[str, Any] = {
+        CONF_PLATFORM: CONF_ESPHOME,
+        CONF_ID: ID(f"ota_esphome_{port}", is_manual=False),
+        CONF_VERSION: 2,
+        CONF_PORT: port,
+    }
+    config.update(kwargs)
+    return config
+
+
+def test_single_esphome_ota_instance_accepted() -> None:
+    """A single ESPHome OTA config passes final_validate untouched."""
+    full_conf = {CONF_OTA: [_make_ota_config(port=3232)]}
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 1
+        assert updated[CONF_OTA][0][CONF_PORT] == 3232
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_same_port_configs_merge(caplog: pytest.LogCaptureFixture) -> None:
+    """Two ESPHome OTA configs on the same port merge into one instance."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_PASSWORD: "pw"}),
+            _make_ota_config(port=3232),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with caplog.at_level(logging.WARNING):
+            ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 1
+        assert updated[CONF_OTA][0][CONF_PORT] == 3232
+        assert any("Found and merged" in record.message for record in caplog.records), (
+            "Expected merge warning not found in log"
+        )
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_multiple_ports_rejected() -> None:
+    """Two ESPHome OTA configs on different ports raise cv.Invalid."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232),
+            _make_ota_config(port=3233),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(
+            cv.Invalid,
+            match=r"Only a single port is supported for 'ota' 'platform: esphome'",
+        ):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_non_esphome_ota_unaffected() -> None:
+    """Non-esphome OTA platforms are not subject to the single-instance rule."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+            {CONF_PLATFORM: "http_request", CONF_ID: ID("ota_hr", is_manual=False)},
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 3
+    finally:
+        fv.full_config.reset(token)