handles newlines, add test to prove it

Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>

.github/workflows/codeql.yml

@@ -58,7 +58,7 @@ jobs:
 
       # Initializes the CodeQL tools for scanning.
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@0499de31b99561a6d14a36a5f662c2a54f91beee # v4.31.2
+        uses: github/codeql-action/init@014f16e7ab1402f30e7c3329d33797e7948572db # v4.31.3
         with:
           languages: ${{ matrix.language }}
           build-mode: ${{ matrix.build-mode }}
@@ -86,6 +86,6 @@ jobs:
           exit 1
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@0499de31b99561a6d14a36a5f662c2a54f91beee # v4.31.2
+        uses: github/codeql-action/analyze@014f16e7ab1402f30e7c3329d33797e7948572db # v4.31.3
         with:
           category: "/language:${{matrix.language}}"

.pre-commit-config.yaml

@@ -11,7 +11,7 @@ ci:
 repos:
   - repo: https://github.com/astral-sh/ruff-pre-commit
     # Ruff version.
-    rev: v0.14.4
+    rev: v0.14.5
     hooks:
       # Run the linter.
       - id: ruff

esphome/components/esp32/__init__.py

@@ -381,8 +381,9 @@ PLATFORM_VERSION_LOOKUP = {
 }
 
 
-def _check_versions(value):
-    value = value.copy()
+def _check_versions(config):
+    config = config.copy()
+    value = config[CONF_FRAMEWORK]
 
     if value[CONF_VERSION] in PLATFORM_VERSION_LOOKUP:
         if CONF_SOURCE in value or CONF_PLATFORM_VERSION in value:
@@ -447,7 +448,7 @@ def _check_versions(value):
             "If there are connectivity or build issues please remove the manual version."
         )
 
-    return value
+    return config
 
 
 def _parse_platform_version(value):
@@ -497,6 +498,8 @@ def final_validate(config):
     from esphome.components.psram import DOMAIN as PSRAM_DOMAIN
 
     errs = []
+    conf_fw = config[CONF_FRAMEWORK]
+    advanced = conf_fw[CONF_ADVANCED]
     full_config = fv.full_config.get()
     if pio_options := full_config[CONF_ESPHOME].get(CONF_PLATFORMIO_OPTIONS):
         pio_flash_size_key = "board_upload.flash_size"
@@ -513,22 +516,14 @@ def final_validate(config):
                     f"Please specify {CONF_FLASH_SIZE} within esp32 configuration only"
                 )
             )
-    if (
-        config[CONF_VARIANT] != VARIANT_ESP32
-        and CONF_ADVANCED in (conf_fw := config[CONF_FRAMEWORK])
-        and CONF_IGNORE_EFUSE_MAC_CRC in conf_fw[CONF_ADVANCED]
-    ):
+    if config[CONF_VARIANT] != VARIANT_ESP32 and advanced[CONF_IGNORE_EFUSE_MAC_CRC]:
         errs.append(
             cv.Invalid(
                 f"'{CONF_IGNORE_EFUSE_MAC_CRC}' is not supported on {config[CONF_VARIANT]}",
                 path=[CONF_FRAMEWORK, CONF_ADVANCED, CONF_IGNORE_EFUSE_MAC_CRC],
             )
         )
-    if (
-        config.get(CONF_FRAMEWORK, {})
-        .get(CONF_ADVANCED, {})
-        .get(CONF_EXECUTE_FROM_PSRAM)
-    ):
+    if advanced[CONF_EXECUTE_FROM_PSRAM]:
         if config[CONF_VARIANT] != VARIANT_ESP32S3:
             errs.append(
                 cv.Invalid(
@@ -544,6 +539,17 @@ def final_validate(config):
                 )
             )
 
+    if (
+        config[CONF_FLASH_SIZE] == "32MB"
+        and "ota" in full_config
+        and not advanced[CONF_ENABLE_IDF_EXPERIMENTAL_FEATURES]
+    ):
+        errs.append(
+            cv.Invalid(
+                f"OTA with 32MB flash requires '{CONF_ENABLE_IDF_EXPERIMENTAL_FEATURES}' to be set in the '{CONF_ADVANCED}' section of the esp32 configuration",
+                path=[CONF_FLASH_SIZE],
+            )
+        )
     if errs:
         raise cv.MultipleInvalid(errs)
 
@@ -598,89 +604,74 @@ def _validate_idf_component(config: ConfigType) -> ConfigType:
 
 FRAMEWORK_ESP_IDF = "esp-idf"
 FRAMEWORK_ARDUINO = "arduino"
-FRAMEWORK_SCHEMA = cv.All(
-    cv.Schema(
-        {
-            cv.Optional(CONF_TYPE, default=FRAMEWORK_ARDUINO): cv.one_of(
-                FRAMEWORK_ESP_IDF, FRAMEWORK_ARDUINO
-            ),
-            cv.Optional(CONF_VERSION, default="recommended"): cv.string_strict,
-            cv.Optional(CONF_RELEASE): cv.string_strict,
-            cv.Optional(CONF_SOURCE): cv.string_strict,
-            cv.Optional(CONF_PLATFORM_VERSION): _parse_platform_version,
-            cv.Optional(CONF_SDKCONFIG_OPTIONS, default={}): {
-                cv.string_strict: cv.string_strict
-            },
-            cv.Optional(CONF_LOG_LEVEL, default="ERROR"): cv.one_of(
-                *LOG_LEVELS_IDF, upper=True
-            ),
-            cv.Optional(CONF_ADVANCED, default={}): cv.Schema(
-                {
-                    cv.Optional(CONF_ASSERTION_LEVEL): cv.one_of(
-                        *ASSERTION_LEVELS, upper=True
-                    ),
-                    cv.Optional(CONF_COMPILER_OPTIMIZATION, default="SIZE"): cv.one_of(
-                        *COMPILER_OPTIMIZATIONS, upper=True
-                    ),
-                    cv.Optional(CONF_ENABLE_IDF_EXPERIMENTAL_FEATURES): cv.boolean,
-                    cv.Optional(CONF_ENABLE_LWIP_ASSERT, default=True): cv.boolean,
-                    cv.Optional(
-                        CONF_IGNORE_EFUSE_CUSTOM_MAC, default=False
-                    ): cv.boolean,
-                    cv.Optional(CONF_IGNORE_EFUSE_MAC_CRC): cv.boolean,
-                    # DHCP server is needed for WiFi AP mode. When WiFi component is used,
-                    # it will handle disabling DHCP server when AP is not configured.
-                    # Default to false (disabled) when WiFi is not used.
-                    cv.OnlyWithout(
-                        CONF_ENABLE_LWIP_DHCP_SERVER, "wifi", default=False
-                    ): cv.boolean,
-                    cv.Optional(
-                        CONF_ENABLE_LWIP_MDNS_QUERIES, default=True
-                    ): cv.boolean,
-                    cv.Optional(
-                        CONF_ENABLE_LWIP_BRIDGE_INTERFACE, default=False
-                    ): cv.boolean,
-                    cv.Optional(
-                        CONF_ENABLE_LWIP_TCPIP_CORE_LOCKING, default=True
-                    ): cv.boolean,
-                    cv.Optional(
-                        CONF_ENABLE_LWIP_CHECK_THREAD_SAFETY, default=True
-                    ): cv.boolean,
-                    cv.Optional(
-                        CONF_DISABLE_LIBC_LOCKS_IN_IRAM, default=True
-                    ): cv.boolean,
-                    cv.Optional(
-                        CONF_DISABLE_VFS_SUPPORT_TERMIOS, default=True
-                    ): cv.boolean,
-                    cv.Optional(
-                        CONF_DISABLE_VFS_SUPPORT_SELECT, default=True
-                    ): cv.boolean,
-                    cv.Optional(CONF_DISABLE_VFS_SUPPORT_DIR, default=True): cv.boolean,
-                    cv.Optional(CONF_EXECUTE_FROM_PSRAM): cv.boolean,
-                    cv.Optional(CONF_LOOP_TASK_STACK_SIZE, default=8192): cv.int_range(
-                        min=8192, max=32768
-                    ),
-                }
-            ),
-            cv.Optional(CONF_COMPONENTS, default=[]): cv.ensure_list(
-                cv.All(
-                    cv.Schema(
-                        {
-                            cv.Required(CONF_NAME): cv.string_strict,
-                            cv.Optional(CONF_SOURCE): cv.git_ref,
-                            cv.Optional(CONF_REF): cv.string,
-                            cv.Optional(CONF_PATH): cv.string,
-                            cv.Optional(CONF_REFRESH): cv.All(
-                                cv.string, cv.source_refresh
-                            ),
-                        }
-                    ),
-                    _validate_idf_component,
-                )
-            ),
-        }
-    ),
-    _check_versions,
+FRAMEWORK_SCHEMA = cv.Schema(
+    {
+        cv.Optional(CONF_TYPE): cv.one_of(FRAMEWORK_ESP_IDF, FRAMEWORK_ARDUINO),
+        cv.Optional(CONF_VERSION, default="recommended"): cv.string_strict,
+        cv.Optional(CONF_RELEASE): cv.string_strict,
+        cv.Optional(CONF_SOURCE): cv.string_strict,
+        cv.Optional(CONF_PLATFORM_VERSION): _parse_platform_version,
+        cv.Optional(CONF_SDKCONFIG_OPTIONS, default={}): {
+            cv.string_strict: cv.string_strict
+        },
+        cv.Optional(CONF_LOG_LEVEL, default="ERROR"): cv.one_of(
+            *LOG_LEVELS_IDF, upper=True
+        ),
+        cv.Optional(CONF_ADVANCED, default={}): cv.Schema(
+            {
+                cv.Optional(CONF_ASSERTION_LEVEL): cv.one_of(
+                    *ASSERTION_LEVELS, upper=True
+                ),
+                cv.Optional(CONF_COMPILER_OPTIMIZATION, default="SIZE"): cv.one_of(
+                    *COMPILER_OPTIMIZATIONS, upper=True
+                ),
+                cv.Optional(
+                    CONF_ENABLE_IDF_EXPERIMENTAL_FEATURES, default=False
+                ): cv.boolean,
+                cv.Optional(CONF_ENABLE_LWIP_ASSERT, default=True): cv.boolean,
+                cv.Optional(CONF_IGNORE_EFUSE_CUSTOM_MAC, default=False): cv.boolean,
+                cv.Optional(CONF_IGNORE_EFUSE_MAC_CRC, default=False): cv.boolean,
+                # DHCP server is needed for WiFi AP mode. When WiFi component is used,
+                # it will handle disabling DHCP server when AP is not configured.
+                # Default to false (disabled) when WiFi is not used.
+                cv.OnlyWithout(
+                    CONF_ENABLE_LWIP_DHCP_SERVER, "wifi", default=False
+                ): cv.boolean,
+                cv.Optional(CONF_ENABLE_LWIP_MDNS_QUERIES, default=True): cv.boolean,
+                cv.Optional(
+                    CONF_ENABLE_LWIP_BRIDGE_INTERFACE, default=False
+                ): cv.boolean,
+                cv.Optional(
+                    CONF_ENABLE_LWIP_TCPIP_CORE_LOCKING, default=True
+                ): cv.boolean,
+                cv.Optional(
+                    CONF_ENABLE_LWIP_CHECK_THREAD_SAFETY, default=True
+                ): cv.boolean,
+                cv.Optional(CONF_DISABLE_LIBC_LOCKS_IN_IRAM, default=True): cv.boolean,
+                cv.Optional(CONF_DISABLE_VFS_SUPPORT_TERMIOS, default=True): cv.boolean,
+                cv.Optional(CONF_DISABLE_VFS_SUPPORT_SELECT, default=True): cv.boolean,
+                cv.Optional(CONF_DISABLE_VFS_SUPPORT_DIR, default=True): cv.boolean,
+                cv.Optional(CONF_EXECUTE_FROM_PSRAM, default=False): cv.boolean,
+                cv.Optional(CONF_LOOP_TASK_STACK_SIZE, default=8192): cv.int_range(
+                    min=8192, max=32768
+                ),
+            }
+        ),
+        cv.Optional(CONF_COMPONENTS, default=[]): cv.ensure_list(
+            cv.All(
+                cv.Schema(
+                    {
+                        cv.Required(CONF_NAME): cv.string_strict,
+                        cv.Optional(CONF_SOURCE): cv.git_ref,
+                        cv.Optional(CONF_REF): cv.string,
+                        cv.Optional(CONF_PATH): cv.string,
+                        cv.Optional(CONF_REFRESH): cv.All(cv.string, cv.source_refresh),
+                    }
+                ),
+                _validate_idf_component,
+            )
+        ),
+    }
 )
 
 
@@ -743,11 +734,11 @@ def _show_framework_migration_message(name: str, variant: str) -> None:
 
 
 def _set_default_framework(config):
+    config = config.copy()
     if CONF_FRAMEWORK not in config:
-        config = config.copy()
-
-        variant = config[CONF_VARIANT]
         config[CONF_FRAMEWORK] = FRAMEWORK_SCHEMA({})
+    if CONF_TYPE not in config[CONF_FRAMEWORK]:
+        variant = config[CONF_VARIANT]
         if variant in ARDUINO_ALLOWED_VARIANTS:
             config[CONF_FRAMEWORK][CONF_TYPE] = FRAMEWORK_ARDUINO
             _show_framework_migration_message(
@@ -787,6 +778,7 @@ CONFIG_SCHEMA = cv.All(
     ),
     _detect_variant,
     _set_default_framework,
+    _check_versions,
     set_core_data,
     cv.has_at_least_one_key(CONF_BOARD, CONF_VARIANT),
 )
@@ -805,9 +797,7 @@ def _configure_lwip_max_sockets(conf: dict) -> None:
     from esphome.components.socket import KEY_SOCKET_CONSUMERS
 
     # Check if user manually specified CONFIG_LWIP_MAX_SOCKETS
-    user_max_sockets = conf.get(CONF_SDKCONFIG_OPTIONS, {}).get(
-        "CONFIG_LWIP_MAX_SOCKETS"
-    )
+    user_max_sockets = conf[CONF_SDKCONFIG_OPTIONS].get("CONFIG_LWIP_MAX_SOCKETS")
 
     socket_consumers: dict[str, int] = CORE.data.get(KEY_SOCKET_CONSUMERS, {})
     total_sockets = sum(socket_consumers.values())
@@ -977,23 +967,18 @@ async def to_code(config):
     # WiFi component handles its own optimization when AP mode is not used
     # When using Arduino with Ethernet, DHCP server functions must be available
     # for the Network library to compile, even if not actively used
-    if (
-        CONF_ENABLE_LWIP_DHCP_SERVER in advanced
-        and not advanced[CONF_ENABLE_LWIP_DHCP_SERVER]
-        and not (
-            conf[CONF_TYPE] == FRAMEWORK_ARDUINO
-            and "ethernet" in CORE.loaded_integrations
-        )
+    if advanced.get(CONF_ENABLE_LWIP_DHCP_SERVER) is False and not (
+        conf[CONF_TYPE] == FRAMEWORK_ARDUINO and "ethernet" in CORE.loaded_integrations
     ):
         add_idf_sdkconfig_option("CONFIG_LWIP_DHCPS", False)
-    if not advanced.get(CONF_ENABLE_LWIP_MDNS_QUERIES, True):
+    if not advanced[CONF_ENABLE_LWIP_MDNS_QUERIES]:
         add_idf_sdkconfig_option("CONFIG_LWIP_DNS_SUPPORT_MDNS_QUERIES", False)
-    if not advanced.get(CONF_ENABLE_LWIP_BRIDGE_INTERFACE, False):
+    if not advanced[CONF_ENABLE_LWIP_BRIDGE_INTERFACE]:
         add_idf_sdkconfig_option("CONFIG_LWIP_BRIDGEIF_MAX_PORTS", 0)
 
     _configure_lwip_max_sockets(conf)
 
-    if advanced.get(CONF_EXECUTE_FROM_PSRAM, False):
+    if advanced[CONF_EXECUTE_FROM_PSRAM]:
         add_idf_sdkconfig_option("CONFIG_SPIRAM_FETCH_INSTRUCTIONS", True)
         add_idf_sdkconfig_option("CONFIG_SPIRAM_RODATA", True)
 
@@ -1004,23 +989,22 @@ async def to_code(config):
     # - select() on 4 sockets: ~190μs (Arduino/core locking) vs ~235μs (ESP-IDF default)
     # - Up to 200% slower under load when all operations queue through tcpip_thread
     # Enabling this makes ESP-IDF socket performance match Arduino framework.
-    if advanced.get(CONF_ENABLE_LWIP_TCPIP_CORE_LOCKING, True):
+    if advanced[CONF_ENABLE_LWIP_TCPIP_CORE_LOCKING]:
         add_idf_sdkconfig_option("CONFIG_LWIP_TCPIP_CORE_LOCKING", True)
-    if advanced.get(CONF_ENABLE_LWIP_CHECK_THREAD_SAFETY, True):
+    if advanced[CONF_ENABLE_LWIP_CHECK_THREAD_SAFETY]:
         add_idf_sdkconfig_option("CONFIG_LWIP_CHECK_THREAD_SAFETY", True)
 
     # Disable placing libc locks in IRAM to save RAM
     # This is safe for ESPHome since no IRAM ISRs (interrupts that run while cache is disabled)
     # use libc lock APIs. Saves approximately 1.3KB (1,356 bytes) of IRAM.
-    if advanced.get(CONF_DISABLE_LIBC_LOCKS_IN_IRAM, True):
+    if advanced[CONF_DISABLE_LIBC_LOCKS_IN_IRAM]:
         add_idf_sdkconfig_option("CONFIG_LIBC_LOCKS_PLACE_IN_IRAM", False)
 
     # Disable VFS support for termios (terminal I/O functions)
     # ESPHome doesn't use termios functions on ESP32 (only used in host UART driver).
     # Saves approximately 1.8KB of flash when disabled (default).
     add_idf_sdkconfig_option(
-        "CONFIG_VFS_SUPPORT_TERMIOS",
-        not advanced.get(CONF_DISABLE_VFS_SUPPORT_TERMIOS, True),
+        "CONFIG_VFS_SUPPORT_TERMIOS", not advanced[CONF_DISABLE_VFS_SUPPORT_TERMIOS]
     )
 
     # Disable VFS support for select() with file descriptors
@@ -1034,8 +1018,7 @@ async def to_code(config):
     else:
         # No component needs it - allow user to control (default: disabled)
         add_idf_sdkconfig_option(
-            "CONFIG_VFS_SUPPORT_SELECT",
-            not advanced.get(CONF_DISABLE_VFS_SUPPORT_SELECT, True),
+            "CONFIG_VFS_SUPPORT_SELECT", not advanced[CONF_DISABLE_VFS_SUPPORT_SELECT]
         )
 
     # Disable VFS support for directory functions (opendir, readdir, mkdir, etc.)
@@ -1048,8 +1031,7 @@ async def to_code(config):
     else:
         # No component needs it - allow user to control (default: disabled)
         add_idf_sdkconfig_option(
-            "CONFIG_VFS_SUPPORT_DIR",
-            not advanced.get(CONF_DISABLE_VFS_SUPPORT_DIR, True),
+            "CONFIG_VFS_SUPPORT_DIR", not advanced[CONF_DISABLE_VFS_SUPPORT_DIR]
         )
 
     cg.add_platformio_option("board_build.partitions", "partitions.csv")
@@ -1063,7 +1045,7 @@ async def to_code(config):
             add_idf_sdkconfig_option(flag, assertion_level == key)
 
     add_idf_sdkconfig_option("CONFIG_COMPILER_OPTIMIZATION_DEFAULT", False)
-    compiler_optimization = advanced.get(CONF_COMPILER_OPTIMIZATION)
+    compiler_optimization = advanced[CONF_COMPILER_OPTIMIZATION]
     for key, flag in COMPILER_OPTIMIZATIONS.items():
         add_idf_sdkconfig_option(flag, compiler_optimization == key)
 
@@ -1072,18 +1054,20 @@ async def to_code(config):
         conf[CONF_ADVANCED][CONF_ENABLE_LWIP_ASSERT],
     )
 
-    if advanced.get(CONF_IGNORE_EFUSE_MAC_CRC):
+    if advanced[CONF_IGNORE_EFUSE_MAC_CRC]:
         add_idf_sdkconfig_option("CONFIG_ESP_MAC_IGNORE_MAC_CRC_ERROR", True)
         add_idf_sdkconfig_option("CONFIG_ESP_PHY_CALIBRATION_AND_DATA_STORAGE", False)
-    if advanced.get(CONF_ENABLE_IDF_EXPERIMENTAL_FEATURES):
+    if advanced[CONF_ENABLE_IDF_EXPERIMENTAL_FEATURES]:
         _LOGGER.warning(
             "Using experimental features in ESP-IDF may result in unexpected failures."
         )
         add_idf_sdkconfig_option("CONFIG_IDF_EXPERIMENTAL_FEATURES", True)
+        if config[CONF_FLASH_SIZE] == "32MB":
+            add_idf_sdkconfig_option(
+                "CONFIG_BOOTLOADER_CACHE_32BIT_ADDR_QUAD_FLASH", True
+            )
 
-    cg.add_define(
-        "ESPHOME_LOOP_TASK_STACK_SIZE", advanced.get(CONF_LOOP_TASK_STACK_SIZE)
-    )
+    cg.add_define("ESPHOME_LOOP_TASK_STACK_SIZE", advanced[CONF_LOOP_TASK_STACK_SIZE])
 
     cg.add_define(
         "USE_ESP_IDF_VERSION_CODE",

esphome/components/light/light_call.cpp

@@ -52,8 +52,10 @@ static void log_invalid_parameter(const char *name, const LogString *message) {
   }
 
 static const LogString *color_mode_to_human(ColorMode color_mode) {
-  if (color_mode == ColorMode::UNKNOWN)
-    return LOG_STR("Unknown");
+  if (color_mode == ColorMode::ON_OFF)
+    return LOG_STR("On/Off");
+  if (color_mode == ColorMode::BRIGHTNESS)
+    return LOG_STR("Brightness");
   if (color_mode == ColorMode::WHITE)
     return LOG_STR("White");
   if (color_mode == ColorMode::COLOR_TEMPERATURE)
@@ -68,7 +70,7 @@ static const LogString *color_mode_to_human(ColorMode color_mode) {
     return LOG_STR("RGB + cold/warm white");
   if (color_mode == ColorMode::RGB_COLOR_TEMPERATURE)
     return LOG_STR("RGB + color temperature");
-  return LOG_STR("");
+  return LOG_STR("Unknown");
 }
 
 // Helper to log percentage values

esphome/components/uart/__init__.py

@@ -1,3 +1,4 @@
+from logging import getLogger
 import math
 import re
 
@@ -35,6 +36,8 @@ from esphome.core import CORE, ID
 import esphome.final_validate as fv
 from esphome.yaml_util import make_data_base
 
+_LOGGER = getLogger(__name__)
+
 CODEOWNERS = ["@esphome/core"]
 uart_ns = cg.esphome_ns.namespace("uart")
 UARTComponent = uart_ns.class_("UARTComponent")
@@ -130,6 +133,21 @@ def validate_host_config(config):
     return config
 
 
+def validate_rx_buffer_size(config):
+    if CORE.is_esp32:
+        # ESP32 UART hardware FIFO is 128 bytes (LP UART is 16 bytes, but we use 128 as safe minimum)
+        # rx_buffer_size must be greater than the hardware FIFO length
+        min_buffer_size = 128
+        if config[CONF_RX_BUFFER_SIZE] <= min_buffer_size:
+            _LOGGER.warning(
+                "UART rx_buffer_size (%d bytes) is too small and must be greater than the hardware "
+                "FIFO size (%d bytes). The buffer size will be automatically adjusted at runtime.",
+                config[CONF_RX_BUFFER_SIZE],
+                min_buffer_size,
+            )
+    return config
+
+
 def _uart_declare_type(value):
     if CORE.is_esp8266:
         return cv.declare_id(ESP8266UartComponent)(value)
@@ -247,6 +265,7 @@ CONFIG_SCHEMA = cv.All(
     ).extend(cv.COMPONENT_SCHEMA),
     cv.has_at_least_one_key(CONF_TX_PIN, CONF_RX_PIN, CONF_PORT),
     validate_host_config,
+    validate_rx_buffer_size,
 )
 
 

esphome/components/uart/uart_component_esp_idf.cpp

@@ -91,6 +91,16 @@ void IDFUARTComponent::setup() {
   this->uart_num_ = static_cast<uart_port_t>(next_uart_num++);
   this->lock_ = xSemaphoreCreateMutex();
 
+#if (SOC_UART_LP_NUM >= 1)
+  size_t fifo_len = ((this->uart_num_ < SOC_UART_HP_NUM) ? SOC_UART_FIFO_LEN : SOC_LP_UART_FIFO_LEN);
+#else
+  size_t fifo_len = SOC_UART_FIFO_LEN;
+#endif
+  if (this->rx_buffer_size_ <= fifo_len) {
+    ESP_LOGW(TAG, "rx_buffer_size is too small, must be greater than %zu", fifo_len);
+    this->rx_buffer_size_ = fifo_len * 2;
+  }
+
   xSemaphoreTake(this->lock_, portMAX_DELAY);
 
   this->load_settings(false);
@@ -237,8 +247,12 @@ void IDFUARTComponent::set_rx_timeout(size_t rx_timeout) {
 
 void IDFUARTComponent::write_array(const uint8_t *data, size_t len) {
   xSemaphoreTake(this->lock_, portMAX_DELAY);
-  uart_write_bytes(this->uart_num_, data, len);
+  int32_t write_len = uart_write_bytes(this->uart_num_, data, len);
   xSemaphoreGive(this->lock_);
+  if (write_len != (int32_t) len) {
+    ESP_LOGW(TAG, "uart_write_bytes failed: %d != %zu", write_len, len);
+    this->mark_failed();
+  }
 #ifdef USE_UART_DEBUGGER
   for (size_t i = 0; i < len; i++) {
     this->debug_callback_.call(UART_DIRECTION_TX, data[i]);
@@ -267,6 +281,7 @@ bool IDFUARTComponent::peek_byte(uint8_t *data) {
 
 bool IDFUARTComponent::read_array(uint8_t *data, size_t len) {
   size_t length_to_read = len;
+  int32_t read_len = 0;
   if (!this->check_read_timeout_(len))
     return false;
   xSemaphoreTake(this->lock_, portMAX_DELAY);
@@ -277,25 +292,31 @@ bool IDFUARTComponent::read_array(uint8_t *data, size_t len) {
     this->has_peek_ = false;
   }
   if (length_to_read > 0)
-    uart_read_bytes(this->uart_num_, data, length_to_read, 20 / portTICK_PERIOD_MS);
+    read_len = uart_read_bytes(this->uart_num_, data, length_to_read, 20 / portTICK_PERIOD_MS);
   xSemaphoreGive(this->lock_);
 #ifdef USE_UART_DEBUGGER
   for (size_t i = 0; i < len; i++) {
     this->debug_callback_.call(UART_DIRECTION_RX, data[i]);
   }
 #endif
-  return true;
+  return read_len == (int32_t) length_to_read;
 }
 
 int IDFUARTComponent::available() {
-  size_t available;
+  size_t available = 0;
+  esp_err_t err;
 
   xSemaphoreTake(this->lock_, portMAX_DELAY);
-  uart_get_buffered_data_len(this->uart_num_, &available);
-  if (this->has_peek_)
-    available++;
+  err = uart_get_buffered_data_len(this->uart_num_, &available);
   xSemaphoreGive(this->lock_);
 
+  if (err != ESP_OK) {
+    ESP_LOGW(TAG, "uart_get_buffered_data_len failed: %s", esp_err_to_name(err));
+    this->mark_failed();
+  }
+  if (this->has_peek_) {
+    available++;
+  }
   return available;
 }
 

esphome/components/wifi/__init__.py

@@ -12,7 +12,6 @@ from esphome.components.network import (
 from esphome.components.psram import is_guaranteed as psram_is_guaranteed
 from esphome.config_helpers import filter_source_files_from_platform
 import esphome.config_validation as cv
-from esphome.config_validation import only_with_esp_idf
 from esphome.const import (
     CONF_AP,
     CONF_BSSID,
@@ -352,7 +351,7 @@ CONFIG_SCHEMA = cv.All(
                 single=True
             ),
             cv.Optional(CONF_USE_PSRAM): cv.All(
-                only_with_esp_idf, cv.requires_component("psram"), cv.boolean
+                cv.only_on_esp32, cv.requires_component("psram"), cv.boolean
             ),
         }
     ),

esphome/cpp_generator.py

@@ -19,11 +19,21 @@ from esphome.core import (
     TimePeriodNanoseconds,
     TimePeriodSeconds,
 )
+from esphome.coroutine import CoroPriority, coroutine_with_priority
 from esphome.helpers import cpp_string_escape, indent_all_but_first_and_last
 from esphome.types import Expression, SafeExpType, TemplateArgsType
 from esphome.util import OrderedDict
 from esphome.yaml_util import ESPHomeDataBase
 
+# Keys for lambda deduplication storage in CORE.data
+_KEY_LAMBDA_DEDUP = "lambda_dedup"
+_KEY_LAMBDA_DEDUP_DECLARATIONS = "lambda_dedup_declarations"
+
+# Regex patterns for static variable detection (compiled once)
+_RE_CPP_SINGLE_LINE_COMMENT = re.compile(r"//.*?$", re.MULTILINE)
+_RE_CPP_MULTI_LINE_COMMENT = re.compile(r"/\*.*?\*/", re.DOTALL)
+_RE_STATIC_VARIABLE = re.compile(r"\bstatic\s+(?!cast|assert|pointer_cast)\w+\s+\w+")
+
 
 class RawExpression(Expression):
     __slots__ = ("text",)
@@ -188,7 +198,7 @@ class LambdaExpression(Expression):
 
     def __init__(
         self, parts, parameters, capture: str = "=", return_type=None, source=None
-    ):
+    ) -> None:
         self.parts = parts
         if not isinstance(parameters, ParameterListExpression):
             parameters = ParameterListExpression(*parameters)
@@ -197,16 +207,21 @@ class LambdaExpression(Expression):
         self.capture = capture
         self.return_type = safe_exp(return_type) if return_type is not None else None
 
-    def __str__(self):
+    def format_body(self) -> str:
+        """Format the lambda body with source directive and content."""
+        body = ""
+        if self.source is not None:
+            body += f"{self.source.as_line_directive}\n"
+        body += self.content
+        return body
+
+    def __str__(self) -> str:
         # Stateless lambdas (empty capture) implicitly convert to function pointers
         # when assigned to function pointer types - no unary + needed
         cpp = f"[{self.capture}]({self.parameters})"
         if self.return_type is not None:
             cpp += f" -> {self.return_type}"
-        cpp += " {\n"
-        if self.source is not None:
-            cpp += f"{self.source.as_line_directive}\n"
-        cpp += f"{self.content}\n}}"
+        cpp += f" {{\n{self.format_body()}\n}}"
         return indent_all_but_first_and_last(cpp)
 
     @property
@@ -214,6 +229,37 @@ class LambdaExpression(Expression):
         return "".join(str(part) for part in self.parts)
 
 
+class SharedFunctionLambdaExpression(LambdaExpression):
+    """A lambda expression that references a shared deduplicated function.
+
+    This class wraps a function pointer but maintains the LambdaExpression
+    interface so calling code works unchanged.
+    """
+
+    __slots__ = ("_func_name",)
+
+    def __init__(
+        self,
+        func_name: str,
+        parameters: TemplateArgsType,
+        return_type: SafeExpType | None = None,
+    ) -> None:
+        # Initialize parent with empty parts since we're just a function reference
+        super().__init__(
+            [], parameters, capture="", return_type=return_type, source=None
+        )
+        self._func_name = func_name
+
+    def __str__(self) -> str:
+        # Just return the function name - it's already a function pointer
+        return self._func_name
+
+    @property
+    def content(self) -> str:
+        # No content, just a function reference
+        return ""
+
+
 # pylint: disable=abstract-method
 class Literal(Expression, metaclass=abc.ABCMeta):
     __slots__ = ()
@@ -583,6 +629,25 @@ def add_global(expression: SafeExpType | Statement, prepend: bool = False):
     CORE.add_global(expression, prepend)
 
 
+@coroutine_with_priority(CoroPriority.FINAL)
+async def flush_lambda_dedup_declarations() -> None:
+    """Flush all deferred lambda deduplication declarations to global scope.
+
+    This is a coroutine that runs with FINAL priority (after all components)
+    to ensure all referenced variables are declared before the shared
+    lambda functions that use them.
+    """
+    if _KEY_LAMBDA_DEDUP_DECLARATIONS not in CORE.data:
+        return
+
+    declarations = CORE.data[_KEY_LAMBDA_DEDUP_DECLARATIONS]
+    for func_declaration in declarations:
+        add_global(RawStatement(func_declaration))
+
+    # Clear the list so we don't add them again
+    CORE.data[_KEY_LAMBDA_DEDUP_DECLARATIONS] = []
+
+
 def add_library(name: str, version: str | None, repository: str | None = None):
     """Add a library to the codegen library storage.
 
@@ -656,6 +721,93 @@ async def get_variable_with_full_id(id_: ID) -> tuple[ID, "MockObj"]:
     return await CORE.get_variable_with_full_id(id_)
 
 
+def _has_static_variables(code: str) -> bool:
+    """Check if code contains static variable definitions.
+
+    Static variables in lambdas should not be deduplicated because each lambda
+    instance should have its own static variable state.
+
+    Args:
+        code: The lambda body code to check
+
+    Returns:
+        True if code contains static variable definitions
+    """
+    # Remove C++ comments to avoid false positives
+    # Remove single-line comments (// ...)
+    code_no_comments = _RE_CPP_SINGLE_LINE_COMMENT.sub("", code)
+    # Remove multi-line comments (/* ... */)
+    code_no_comments = _RE_CPP_MULTI_LINE_COMMENT.sub("", code_no_comments)
+
+    # Match: static <type> <identifier>
+    # But not: static_cast, static_assert, static_pointer_cast
+    return bool(_RE_STATIC_VARIABLE.search(code_no_comments))
+
+
+def _get_shared_lambda_name(lambda_expr: LambdaExpression) -> str | None:
+    """Get the shared function name for a lambda expression.
+
+    If an identical lambda was already generated, returns the existing shared
+    function name. Otherwise, creates a new shared function and returns its name.
+
+    Lambdas with static variables are not deduplicated to preserve their
+    independent state.
+
+    Args:
+        lambda_expr: The lambda expression to deduplicate
+
+    Returns:
+        The name of the shared function for this lambda (either existing or newly created),
+        or None if the lambda should not be deduplicated (e.g., contains static variables)
+    """
+    # Create a unique key from the lambda content, parameters, and return type
+    content = lambda_expr.content
+
+    # Don't deduplicate lambdas with static variables - each instance needs its own state
+    if _has_static_variables(content):
+        return None
+    param_str = str(lambda_expr.parameters)
+    return_str = (
+        str(lambda_expr.return_type) if lambda_expr.return_type is not None else "void"
+    )
+
+    # Use tuple of (content, params, return_type) as key
+    lambda_key = (content, param_str, return_str)
+
+    # Initialize deduplication storage in CORE.data if not exists
+    if _KEY_LAMBDA_DEDUP not in CORE.data:
+        CORE.data[_KEY_LAMBDA_DEDUP] = {}
+        # Register the flush job to run after all components (FINAL priority)
+        # This ensures all variables are declared before shared lambda functions
+        CORE.add_job(flush_lambda_dedup_declarations)
+
+    lambda_cache = CORE.data[_KEY_LAMBDA_DEDUP]
+
+    # Check if we've seen this lambda before
+    if lambda_key in lambda_cache:
+        # Return name of existing shared function
+        return lambda_cache[lambda_key]
+
+    # First occurrence - create a shared function
+    # Use the cache size as the function number
+    func_name = f"shared_lambda_{len(lambda_cache)}"
+
+    # Build the function declaration using lambda's body formatting
+    func_declaration = (
+        f"{return_str} {func_name}({param_str}) {{\n{lambda_expr.format_body()}\n}}"
+    )
+
+    # Store the declaration to be added later (after all variable declarations)
+    # We can't add it immediately because it might reference variables not yet declared
+    CORE.data.setdefault(_KEY_LAMBDA_DEDUP_DECLARATIONS, []).append(func_declaration)
+
+    # Store in cache
+    lambda_cache[lambda_key] = func_name
+
+    # Return the function name (this is the first occurrence, but we still generate shared function)
+    return func_name
+
+
 async def process_lambda(
     value: Lambda,
     parameters: TemplateArgsType,
@@ -713,6 +865,19 @@ async def process_lambda(
         location.line += value.content_offset
     else:
         location = None
+
+    # Lambda deduplication: Only deduplicate stateless lambdas (empty capture).
+    # Stateful lambdas cannot be shared as they capture different contexts.
+    # Lambdas with static variables are also not deduplicated to preserve independent state.
+    if capture == "":
+        lambda_expr = LambdaExpression(
+            parts, parameters, capture, return_type, location
+        )
+        func_name = _get_shared_lambda_name(lambda_expr)
+        if func_name is not None:
+            # Return a shared function reference instead of inline lambda
+            return SharedFunctionLambdaExpression(func_name, parameters, return_type)
+
     return LambdaExpression(parts, parameters, capture, return_type, location)
 
 

requirements_test.txt

@@ -1,6 +1,6 @@
-pylint==4.0.2
+pylint==4.0.3
 flake8==7.3.0  # also change in .pre-commit-config.yaml when updating
-ruff==0.14.4  # also change in .pre-commit-config.yaml when updating
+ruff==0.14.5  # also change in .pre-commit-config.yaml when updating
 pyupgrade==3.21.1  # also change in .pre-commit-config.yaml when updating
 pre-commit
 

tests/component_tests/text/test_text.py

@@ -1,4 +1,6 @@
-"""Tests for the binary sensor component."""
+"""Tests for the text component."""
+
+from esphome.core import CORE
 
 
 def test_text_is_setup(generate_main):
@@ -56,15 +58,22 @@ def test_text_config_value_mode_set(generate_main):
     assert "it_3->traits.set_mode(text::TEXT_MODE_PASSWORD);" in main_cpp
 
 
-def test_text_config_lamda_is_set(generate_main):
+def test_text_config_lambda_is_set(generate_main) -> None:
     """
-    Test if lambda is set for lambda mode (optimized with stateless lambda)
+    Test if lambda is set for lambda mode (optimized with stateless lambda and deduplication)
     """
     # Given
 
     # When
     main_cpp = generate_main("tests/component_tests/text/test_text.yaml")
 
+    # Get both global and main sections to find the shared lambda definition
+    full_cpp = CORE.cpp_global_section + main_cpp
+
     # Then
-    assert "it_4->set_template([]() -> esphome::optional<std::string> {" in main_cpp
-    assert 'return std::string{"Hello"};' in main_cpp
+    # Lambda is deduplicated into a shared function (reference in main section)
+    assert "it_4->set_template(shared_lambda_" in main_cpp
+    # Lambda body should be in the code somewhere
+    assert 'return std::string{"Hello"};' in full_cpp
+    # Verify the shared lambda function is defined (in global section)
+    assert "esphome::optional<std::string> shared_lambda_" in full_cpp

tests/components/esp32/test.esp32-p4-idf.yaml

@@ -0,0 +1,27 @@
+esp32:
+  variant: esp32p4
+  flash_size: 32MB
+  cpu_frequency: 400MHz
+  framework:
+    type: esp-idf
+    advanced:
+      enable_idf_experimental_features: yes
+
+ota:
+  platform: esphome
+
+wifi:
+  ssid: MySSID
+  password: password1
+
+esp32_hosted:
+  variant: ESP32C6
+  slot: 1
+  active_high: true
+  reset_pin: GPIO15
+  cmd_pin: GPIO13
+  clk_pin: GPIO12
+  d0_pin: GPIO11
+  d1_pin: GPIO10
+  d2_pin: GPIO9
+  d3_pin: GPIO8

tests/unit_tests/test_lambda_dedup.py

@@ -0,0 +1,286 @@
+"""Tests for lambda deduplication in cpp_generator."""
+
+from esphome import cpp_generator as cg
+from esphome.core import CORE
+
+
+def test_deduplicate_identical_lambdas() -> None:
+    """Test that identical stateless lambdas are deduplicated."""
+    # Create two identical lambda expressions
+    lambda1 = cg.LambdaExpression(
+        parts=["return 42;"],
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    lambda2 = cg.LambdaExpression(
+        parts=["return 42;"],
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    # Try to deduplicate them
+    func_name1 = cg._get_shared_lambda_name(lambda1)
+    func_name2 = cg._get_shared_lambda_name(lambda2)
+
+    # Both should get the same function name (deduplication happened)
+    assert func_name1 == func_name2
+    assert func_name1 == "shared_lambda_0"
+
+
+def test_different_lambdas_not_deduplicated() -> None:
+    """Test that different lambdas get different function names."""
+    lambda1 = cg.LambdaExpression(
+        parts=["return 42;"],
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    lambda2 = cg.LambdaExpression(
+        parts=["return 24;"],  # Different content
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    func_name1 = cg._get_shared_lambda_name(lambda1)
+    func_name2 = cg._get_shared_lambda_name(lambda2)
+
+    # Different lambdas should get different function names
+    assert func_name1 != func_name2
+    assert func_name1 == "shared_lambda_0"
+    assert func_name2 == "shared_lambda_1"
+
+
+def test_different_return_types_not_deduplicated() -> None:
+    """Test that lambdas with different return types are not deduplicated."""
+    lambda1 = cg.LambdaExpression(
+        parts=["return 42;"],
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    lambda2 = cg.LambdaExpression(
+        parts=["return 42;"],  # Same content
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("float"),  # Different return type
+    )
+
+    func_name1 = cg._get_shared_lambda_name(lambda1)
+    func_name2 = cg._get_shared_lambda_name(lambda2)
+
+    # Different return types = different functions
+    assert func_name1 != func_name2
+
+
+def test_different_parameters_not_deduplicated() -> None:
+    """Test that lambdas with different parameters are not deduplicated."""
+    lambda1 = cg.LambdaExpression(
+        parts=["return x;"],
+        parameters=[("int", "x")],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    lambda2 = cg.LambdaExpression(
+        parts=["return x;"],  # Same content
+        parameters=[("float", "x")],  # Different parameter type
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    func_name1 = cg._get_shared_lambda_name(lambda1)
+    func_name2 = cg._get_shared_lambda_name(lambda2)
+
+    # Different parameters = different functions
+    assert func_name1 != func_name2
+
+
+def test_flush_lambda_dedup_declarations() -> None:
+    """Test that deferred declarations are properly stored for later flushing."""
+    # Create a lambda which will create a deferred declaration
+    lambda1 = cg.LambdaExpression(
+        parts=["return 42;"],
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    cg._get_shared_lambda_name(lambda1)
+
+    # Check that declaration was stored
+    assert cg._KEY_LAMBDA_DEDUP_DECLARATIONS in CORE.data
+    assert len(CORE.data[cg._KEY_LAMBDA_DEDUP_DECLARATIONS]) == 1
+
+    # Verify the declaration content is correct
+    declaration = CORE.data[cg._KEY_LAMBDA_DEDUP_DECLARATIONS][0]
+    assert "shared_lambda_0" in declaration
+    assert "return 42;" in declaration
+
+    # Note: The actual flushing happens via CORE.add_job with FINAL priority
+    # during real code generation, so we don't test that here
+
+
+def test_shared_function_lambda_expression() -> None:
+    """Test SharedFunctionLambdaExpression behaves correctly."""
+    shared_lambda = cg.SharedFunctionLambdaExpression(
+        func_name="shared_lambda_0",
+        parameters=[],
+        return_type=cg.RawExpression("int"),
+    )
+
+    # Should output just the function name
+    assert str(shared_lambda) == "shared_lambda_0"
+
+    # Should have empty capture (stateless)
+    assert shared_lambda.capture == ""
+
+    # Should have empty content (just a reference)
+    assert shared_lambda.content == ""
+
+
+def test_lambda_deduplication_counter() -> None:
+    """Test that lambda counter increments correctly."""
+    # Create 3 different lambdas
+    for i in range(3):
+        lambda_expr = cg.LambdaExpression(
+            parts=[f"return {i};"],
+            parameters=[],
+            capture="",
+            return_type=cg.RawExpression("int"),
+        )
+        func_name = cg._get_shared_lambda_name(lambda_expr)
+        assert func_name == f"shared_lambda_{i}"
+
+
+def test_lambda_format_body() -> None:
+    """Test that format_body correctly formats lambda body with source."""
+    # Without source
+    lambda1 = cg.LambdaExpression(
+        parts=["return 42;"],
+        parameters=[],
+        capture="",
+        return_type=None,
+        source=None,
+    )
+    assert lambda1.format_body() == "return 42;"
+
+    # With source would need a proper source object, skip for now
+
+
+def test_stateful_lambdas_not_deduplicated() -> None:
+    """Test that stateful lambdas (non-empty capture) are not deduplicated."""
+    # _get_shared_lambda_name is only called for stateless lambdas (capture == "")
+    # Stateful lambdas bypass deduplication entirely in process_lambda
+
+    # Verify that a stateful lambda would NOT get deduplicated
+    # by checking it's not in the stateless dedup cache
+    stateful_lambda = cg.LambdaExpression(
+        parts=["return x + y;"],
+        parameters=[],
+        capture="=",  # Non-empty capture means stateful
+        return_type=cg.RawExpression("int"),
+    )
+
+    # Stateful lambdas should NOT be passed to _get_shared_lambda_name
+    # This is enforced by the `if capture == ""` check in process_lambda
+    # We verify the lambda has a non-empty capture
+    assert stateful_lambda.capture != ""
+    assert stateful_lambda.capture == "="
+
+
+def test_static_variable_detection() -> None:
+    """Test detection of static variables in lambda code."""
+    # Should detect static variables
+    assert cg._has_static_variables("static int counter = 0;")
+    assert cg._has_static_variables("static bool flag = false; return flag;")
+    assert cg._has_static_variables("  static  float  value  =  1.0;  ")
+
+    # Should NOT detect static_cast, static_assert, etc. (with underscores)
+    assert not cg._has_static_variables("return static_cast<int>(value);")
+    assert not cg._has_static_variables("static_assert(sizeof(int) == 4);")
+    assert not cg._has_static_variables("auto ptr = static_pointer_cast<Foo>(bar);")
+
+    # Edge case: 'cast', 'assert', 'pointer_cast' are NOT C++ keywords
+    # Someone could use them as type names, but we should NOT flag them
+    # because they're not actually static variables with state
+    # NOTE: These are valid C++ but extremely unlikely in ESPHome lambdas
+    assert not cg._has_static_variables("static cast obj;")  # 'cast' as type name
+    assert not cg._has_static_variables("static assert value;")  # 'assert' as type name
+    assert not cg._has_static_variables(
+        "static pointer_cast ptr;"
+    )  # 'pointer_cast' as type
+
+    # Should NOT detect in comments
+    assert not cg._has_static_variables("// static int x = 0;\nreturn 42;")
+    assert not cg._has_static_variables("/* static int y = 0; */ return 42;")
+
+    # Should detect even with comments elsewhere
+    assert cg._has_static_variables("// comment\nstatic int x = 0;\nreturn x;")
+
+    # Should NOT detect non-static code
+    assert not cg._has_static_variables("int counter = 0; return counter++;")
+    assert not cg._has_static_variables("return 42;")
+
+    # Should handle newlines between static and type/variable
+    assert cg._has_static_variables("static int\nfoo = 0;")
+    assert cg._has_static_variables("static\nint\nbar = 0;")
+    assert cg._has_static_variables(
+        "static  int  \n  foo  =  0;"
+    )  # Mixed spaces/newlines
+
+
+def test_lambdas_with_static_not_deduplicated() -> None:
+    """Test that lambdas with static variables are not deduplicated."""
+    # Two identical lambdas with static variables
+    lambda1 = cg.LambdaExpression(
+        parts=["static int counter = 0; return counter++;"],
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    lambda2 = cg.LambdaExpression(
+        parts=["static int counter = 0; return counter++;"],
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    # Should return None (not deduplicated)
+    func_name1 = cg._get_shared_lambda_name(lambda1)
+    func_name2 = cg._get_shared_lambda_name(lambda2)
+
+    assert func_name1 is None
+    assert func_name2 is None
+
+
+def test_lambdas_without_static_still_deduplicated() -> None:
+    """Test that lambdas without static variables are still deduplicated."""
+    # Two identical lambdas WITHOUT static variables
+    lambda1 = cg.LambdaExpression(
+        parts=["int counter = 0; return counter++;"],  # No static
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    lambda2 = cg.LambdaExpression(
+        parts=["int counter = 0; return counter++;"],  # No static
+        parameters=[],
+        capture="",
+        return_type=cg.RawExpression("int"),
+    )
+
+    # Should be deduplicated (same function name)
+    func_name1 = cg._get_shared_lambda_name(lambda1)
+    func_name2 = cg._get_shared_lambda_name(lambda2)
+
+    assert func_name1 is not None
+    assert func_name2 is not None
+    assert func_name1 == func_name2