Name the CLI data timeout, pin it in a test, word the device comment as observed not proven

esphome/components/esphome/ota/ota_esphome.cpp

@@ -41,9 +41,9 @@ const noise::NoiseContext &ESPHomeOTAComponent::noise_context_() const {
 #endif
 static constexpr uint16_t OTA_BLOCK_SIZE = 8192;
 static constexpr uint32_t OTA_SOCKET_TIMEOUT_HANDSHAKE = 20000;  // milliseconds for initial handshake
-// Milliseconds for data transfer. Outlasts lwIP retransmitting a lost chunk ack
-// six times (1.5 + 3 + 6 + 12 + 24 + 48 s); the CLI waits longer (160 s) so the
-// device is always free again before the CLI retries
+// Milliseconds for data transfer. Covers the lwIP retransmit run seen in
+// practice for a lost chunk ack (1.5 + 3 + 6 + 12 + 24 + 48 s); the CLI waits
+// longer (espota2.DATA_PHASE_TIMEOUT) so the device is free before it retries
 static constexpr uint32_t OTA_SOCKET_TIMEOUT_DATA = 105000;
 
 // Single-instance pointer — multi-port configs are rejected in final_validate.

esphome/espota2.py

@@ -96,6 +96,10 @@ UPLOAD_BUFFER_SIZE = UPLOAD_BLOCK_SIZE * 8
 # across the addresses on top of that.
 EXTRA_UPLOAD_ATTEMPTS = 2
 UPLOAD_RETRY_DELAY = 5.0
+# Data phase timeout; must stay longer than the device's OTA_SOCKET_TIMEOUT_DATA
+# (105 s) so a stalled session is gone before a retry, and long enough for lwIP
+# to get a lost chunk ack through after the retransmit run seen in practice
+DATA_PHASE_TIMEOUT = 160.0
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -694,10 +698,7 @@ def perform_ota(
 
     _LOGGER.info("Handshake complete")
 
-    # Longer than the device's OTA_SOCKET_TIMEOUT_DATA (105 s) so a stalled
-    # session is gone before a retry, and long enough for lwIP to get a lost
-    # chunk ack through after six retransmissions
-    sock.settimeout(160.0)
+    sock.settimeout(DATA_PHASE_TIMEOUT)
 
     if extended_proto:
         send_check(sock, ota_type, "ota type")

tests/unit_tests/test_espota2.py

@@ -754,8 +754,12 @@ def test_run_ota_impl_successful(
     assert result_code == 0
     assert result_host == "192.168.1.100"
 
-    # Verify socket was configured correctly
-    mock_socket.settimeout.assert_called_with(20.0)
+    # Verify socket was configured correctly: connect timeout first, then the
+    # data phase timeout, which must outlast the device's 105 s data timeout
+    timeouts = [c.args[0] for c in mock_socket.settimeout.call_args_list]
+    assert timeouts[0] == 20.0
+    assert espota2.DATA_PHASE_TIMEOUT in timeouts
+    assert espota2.DATA_PHASE_TIMEOUT > 105.0
     mock_socket.connect.assert_called_once_with(("192.168.1.100", 3232))
     mock_socket.close.assert_called_once()