diff --git a/esphome/codegen.py b/esphome/codegen.py
index f0deb6e8d3..6d55c6023d 100644
--- a/esphome/codegen.py
+++ b/esphome/codegen.py
@@ -51,7 +51,6 @@ from esphome.cpp_helpers import (  # noqa: F401
     past_safe_mode,
     register_component,
     register_parented,
-    require_wake_loop_threadsafe,
 )
 from esphome.cpp_types import (  # noqa: F401
     NAN,
diff --git a/esphome/components/esp32_ble/__init__.py b/esphome/components/esp32_ble/__init__.py
index d3db1db70c..ced7e3fec9 100644
--- a/esphome/components/esp32_ble/__init__.py
+++ b/esphome/components/esp32_ble/__init__.py
@@ -7,6 +7,7 @@ from typing import Any
 
 from esphome import automation
 import esphome.codegen as cg
+from esphome.components import socket
 from esphome.components.esp32 import add_idf_sdkconfig_option, const, get_esp32_variant
 import esphome.config_validation as cv
 from esphome.const import (
@@ -21,6 +22,7 @@ from esphome.core import CORE, CoroPriority, TimePeriod, coroutine_with_priority
 import esphome.final_validate as fv
 
 DEPENDENCIES = ["esp32"]
+AUTO_LOAD = ["socket"]
 CODEOWNERS = ["@jesserockz", "@Rapsssito", "@bdraco"]
 DOMAIN = "esp32_ble"
 
@@ -481,10 +483,10 @@ async def to_code(config):
         cg.add(var.set_name(name))
     await cg.register_component(var, config)
 
-    # BLE uses the core wake_loop_threadsafe() mechanism to wake the main loop from BLE tasks
+    # BLE uses the socket wake_loop_threadsafe() mechanism to wake the main loop from BLE tasks
     # This enables low-latency (~12μs) BLE event processing instead of waiting for
     # select() timeout (0-16ms). The wake socket is shared across all components.
-    cg.require_wake_loop_threadsafe()
+    socket.require_wake_loop_threadsafe()
 
     # Define max connections for use in C++ code (e.g., ble_server.h)
     max_connections = config.get(CONF_MAX_CONNECTIONS, DEFAULT_MAX_CONNECTIONS)
diff --git a/esphome/components/socket/__init__.py b/esphome/components/socket/__init__.py
index e6a4cfc07f..4c2ea7f088 100644
--- a/esphome/components/socket/__init__.py
+++ b/esphome/components/socket/__init__.py
@@ -3,6 +3,7 @@ from collections.abc import Callable, MutableMapping
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.core import CORE
+from esphome.cpp_generator import add_define
 
 CODEOWNERS = ["@esphome/core"]
 
@@ -15,6 +16,9 @@ IMPLEMENTATION_BSD_SOCKETS = "bsd_sockets"
 # Components register their socket needs and platforms read this to configure appropriately
 KEY_SOCKET_CONSUMERS = "socket_consumers"
 
+# Wake loop threadsafe support tracking
+KEY_WAKE_LOOP_THREADSAFE_REQUIRED = "wake_loop_threadsafe_required"
+
 
 def consume_sockets(
     value: int, consumer: str
@@ -37,6 +41,30 @@ def consume_sockets(
     return _consume_sockets
 
 
+def require_wake_loop_threadsafe() -> None:
+    """Mark that wake_loop_threadsafe support is required by a component.
+
+    Call this from components that need to wake the main event loop from background threads.
+    This enables the shared UDP loopback socket mechanism (~208 bytes RAM).
+    The socket is shared across all components that use this feature.
+
+    IMPORTANT: This is for background thread context only, NOT ISR context.
+    Socket operations are not safe to call from ISR handlers.
+
+    Example:
+        from esphome.components import socket
+
+        async def to_code(config):
+            socket.require_wake_loop_threadsafe()
+    """
+    # Only set up once (idempotent - multiple components can call this)
+    if not CORE.data.get(KEY_WAKE_LOOP_THREADSAFE_REQUIRED, False):
+        CORE.data[KEY_WAKE_LOOP_THREADSAFE_REQUIRED] = True
+        add_define("USE_WAKE_LOOP_THREADSAFE")
+        # Consume 1 socket for the shared wake notification socket
+        consume_sockets(1, "socket.wake_loop_threadsafe")({})
+
+
 CONFIG_SCHEMA = cv.Schema(
     {
         cv.SplitDefault(
diff --git a/esphome/cpp_helpers.py b/esphome/cpp_helpers.py
index 8b1fd1db27..2698b9b3d5 100644
--- a/esphome/cpp_helpers.py
+++ b/esphome/cpp_helpers.py
@@ -9,7 +9,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, ID, coroutine
 from esphome.coroutine import FakeAwaitable
-from esphome.cpp_generator import LogStringLiteral, add, add_define, get_variable
+from esphome.cpp_generator import LogStringLiteral, add, get_variable
 from esphome.cpp_types import App
 from esphome.types import ConfigFragmentType, ConfigType
 from esphome.util import Registry, RegistryEntry
@@ -124,34 +124,3 @@ async def past_safe_mode():
             yield
 
     return await FakeAwaitable(_safe_mode_generator())
-
-
-# Wake loop threadsafe support tracking
-# Components that need to wake the main event loop from FreeRTOS tasks can call require_wake_loop_threadsafe()
-KEY_WAKE_LOOP_THREADSAFE_REQUIRED = "wake_loop_threadsafe_required"
-
-
-def require_wake_loop_threadsafe() -> None:
-    """Mark that wake_loop_threadsafe support is required by a component.
-
-    Call this from components that need to wake the main event loop from FreeRTOS tasks.
-    This enables the shared UDP loopback socket mechanism (~208 bytes RAM).
-    The socket is shared across all components that use this feature.
-
-    IMPORTANT: This is for FreeRTOS task context only, NOT ISR context.
-    Socket operations are not safe to call from ISR handlers.
-
-    Example:
-        import esphome.codegen as cg
-
-        async def to_code(config):
-            cg.require_wake_loop_threadsafe()
-    """
-    # Only set up once (idempotent - multiple components can call this)
-    if not CORE.data.get(KEY_WAKE_LOOP_THREADSAFE_REQUIRED, False):
-        from esphome.components import socket
-
-        CORE.data[KEY_WAKE_LOOP_THREADSAFE_REQUIRED] = True
-        add_define("USE_WAKE_LOOP_THREADSAFE")
-        # Consume 1 socket for the shared wake notification socket
-        socket.consume_sockets(1, "core.wake_loop_threadsafe")({})
diff --git a/tests/components/socket/conftest.py b/tests/components/socket/conftest.py
new file mode 100644
index 0000000000..5d93cac232
--- /dev/null
+++ b/tests/components/socket/conftest.py
@@ -0,0 +1,12 @@
+"""Configuration file for socket component tests."""
+
+import pytest
+
+from esphome.core import CORE
+
+
+@pytest.fixture(autouse=True)
+def reset_core():
+    """Reset CORE after each test."""
+    yield
+    CORE.reset()
diff --git a/tests/components/socket/test_init.py b/tests/components/socket/test_init.py
new file mode 100644
index 0000000000..45e5ea2211
--- /dev/null
+++ b/tests/components/socket/test_init.py
@@ -0,0 +1,42 @@
+from esphome.components import socket
+from esphome.core import CORE
+
+
+def test_require_wake_loop_threadsafe__first_call() -> None:
+    """Test that first call sets up define and consumes socket."""
+    socket.require_wake_loop_threadsafe()
+
+    # Verify CORE.data was updated
+    assert CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
+
+    # Verify the define was added
+    assert any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in CORE.defines)
+
+
+def test_require_wake_loop_threadsafe__idempotent() -> None:
+    """Test that subsequent calls are idempotent."""
+    # Set up initial state as if already called
+    CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] = True
+
+    # Call again - should not raise or fail
+    socket.require_wake_loop_threadsafe()
+
+    # Verify state is still True
+    assert CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
+
+    # Define should not be added since flag was already True
+    assert not any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in CORE.defines)
+
+
+def test_require_wake_loop_threadsafe__multiple_calls() -> None:
+    """Test that multiple calls only set up once."""
+    # Call three times
+    socket.require_wake_loop_threadsafe()
+    socket.require_wake_loop_threadsafe()
+    socket.require_wake_loop_threadsafe()
+
+    # Verify CORE.data was set
+    assert CORE.data[socket.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
+
+    # Verify the define was added (only once, but we can just check it exists)
+    assert any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in CORE.defines)
diff --git a/tests/unit_tests/test_cpp_helpers.py b/tests/unit_tests/test_cpp_helpers.py
index 89a474f44d..2618803fec 100644
--- a/tests/unit_tests/test_cpp_helpers.py
+++ b/tests/unit_tests/test_cpp_helpers.py
@@ -70,43 +70,3 @@ async def test_register_component__with_setup_priority(monkeypatch):
     assert add_mock.call_count == 4
     app_mock.register_component.assert_called_with(var)
     assert core_mock.component_ids == []
-
-
-def test_require_wake_loop_threadsafe__first_call() -> None:
-    """Test that first call sets up define and consumes socket."""
-    ch.require_wake_loop_threadsafe()
-
-    # Verify CORE.data was updated
-    assert ch.CORE.data[ch.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
-
-    # Verify the define was added
-    assert any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in ch.CORE.defines)
-
-
-def test_require_wake_loop_threadsafe__idempotent() -> None:
-    """Test that subsequent calls are idempotent."""
-    # Set up initial state as if already called
-    ch.CORE.data[ch.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] = True
-
-    # Call again - should not raise or fail
-    ch.require_wake_loop_threadsafe()
-
-    # Verify state is still True
-    assert ch.CORE.data[ch.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
-
-    # Define should not be added since flag was already True
-    assert not any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in ch.CORE.defines)
-
-
-def test_require_wake_loop_threadsafe__multiple_calls() -> None:
-    """Test that multiple calls only set up once."""
-    # Call three times
-    ch.require_wake_loop_threadsafe()
-    ch.require_wake_loop_threadsafe()
-    ch.require_wake_loop_threadsafe()
-
-    # Verify CORE.data was set
-    assert ch.CORE.data[ch.KEY_WAKE_LOOP_THREADSAFE_REQUIRED] is True
-
-    # Verify the define was added (only once, but we can just check it exists)
-    assert any(d.name == "USE_WAKE_LOOP_THREADSAFE" for d in ch.CORE.defines)