Add thread safety checks to async_create_task

homeassistant/bootstrap.py

@@ -731,7 +731,7 @@ async def async_setup_multi_components(
     # to wait to be imported, and the sooner we can get the base platforms
     # loaded the sooner we can start loading the rest of the integrations.
     futures = {
-        domain: hass.async_create_task(
+        domain: hass.async_create_task_internal(
             async_setup_component(hass, domain, config),
             f"setup component {domain}",
             eager_start=True,

homeassistant/config_entries.py

@@ -1087,7 +1087,7 @@ def async_create_task(
 
         target: target to call.
         """
-        task = hass.async_create_task(
+        task = hass.async_create_task_internal(
             target, f"{name} {self.title} {self.domain} {self.entry_id}", eager_start
         )
         if eager_start and task.done():
@@ -1643,7 +1643,7 @@ async def async_remove(self, entry_id: str) -> dict[str, Any]:
         # starting a new flow with the 'unignore' step. If the integration doesn't
         # implement async_step_unignore then this will be a no-op.
         if entry.source == SOURCE_IGNORE:
-            self.hass.async_create_task(
+            self.hass.async_create_task_internal(
                 self.hass.config_entries.flow.async_init(
                     entry.domain,
                     context={"source": SOURCE_UNIGNORE},

homeassistant/core.py

@@ -785,7 +785,9 @@ def create_task(
         target: target to call.
         """
         self.loop.call_soon_threadsafe(
-            functools.partial(self.async_create_task, target, name, eager_start=True)
+            functools.partial(
+                self.async_create_task_internal, target, name, eager_start=True
+            )
         )
 
     @callback
@@ -800,6 +802,37 @@ def async_create_task(
         This method must be run in the event loop. If you are using this in your
         integration, use the create task methods on the config entry instead.
 
+        target: target to call.
+        """
+        # We turned on asyncio debug in April 2024 in the dev containers
+        # in the hope of catching some of the issues that have been
+        # reported. It will take a while to get all the issues fixed in
+        # custom components.
+        #
+        # In 2025.5 we should guard the `verify_event_loop_thread`
+        # check with a check for the `hass.config.debug` flag being set as
+        # we long term, we don't want to be checking this in production
+        # environments since it is a performance hit.
+        self.verify_event_loop_thread("async_create_task")
+        return self.async_create_task_internal(target, name, eager_start)
+
+    @callback
+    def async_create_task_internal(
+        self,
+        target: Coroutine[Any, Any, _R],
+        name: str | None = None,
+        eager_start: bool = True,
+    ) -> asyncio.Task[_R]:
+        """Create a task from within the event loop, internal use only.
+
+        This method is intended to only be used by core internally
+        and should not be considered a stable API. We will make
+        breaking change to this function in the future and it
+        should not be used in integrations.
+
+        This method must be run in the event loop. If you are using this in your
+        integration, use the create task methods on the config entry instead.
+
         target: target to call.
         """
         if eager_start:
@@ -2697,7 +2730,7 @@ async def async_call(
 
         coro = self._execute_service(handler, service_call)
         if not blocking:
-            self._hass.async_create_task(
+            self._hass.async_create_task_internal(
                 self._run_service_call_catch_exceptions(coro, service_call),
                 f"service call background {service_call.domain}.{service_call.service}",
                 eager_start=True,

homeassistant/helpers/integration_platform.py

@@ -85,7 +85,7 @@ def _async_integration_platform_component_loaded(
 
     # At least one of the platforms is not loaded, we need to load them
     # so we have to fall back to creating a task.
-    hass.async_create_task(
+    hass.async_create_task_internal(
         _async_process_integration_platforms_for_component(
             hass, integration, platforms_that_exist, integration_platforms_by_name
         ),
@@ -206,7 +206,7 @@ async def async_process_integration_platforms(
     # We use hass.async_create_task instead of asyncio.create_task because
     # we want to make sure that startup waits for the task to complete.
     #
-    future = hass.async_create_task(
+    future = hass.async_create_task_internal(
         _async_process_integration_platforms(
             hass, platform_name, top_level_components.copy(), process_job
         ),

homeassistant/helpers/storage.py

@@ -468,7 +468,7 @@ def _async_schedule_callback_delayed_write(self) -> None:
             # wrote. Reschedule the timer to the next write time.
             self._async_reschedule_delayed_write(self._next_write_time)
             return
-        self.hass.async_create_task(
+        self.hass.async_create_task_internal(
             self._async_callback_delayed_write(), eager_start=True
         )
 

homeassistant/setup.py

@@ -600,7 +600,7 @@ async def when_setup() -> None:
             _LOGGER.exception("Error handling when_setup callback for %s", component)
 
     if component in hass.config.components:
-        hass.async_create_task(
+        hass.async_create_task_internal(
             when_setup(), f"when setup {component}", eager_start=True
         )
         return

tests/test_core.py

@@ -329,7 +329,7 @@ async def test_async_create_task_schedule_coroutine() -> None:
     async def job():
         pass
 
-    ha.HomeAssistant.async_create_task(hass, job(), eager_start=False)
+    ha.HomeAssistant.async_create_task_internal(hass, job(), eager_start=False)
     assert len(hass.loop.call_soon.mock_calls) == 0
     assert len(hass.loop.create_task.mock_calls) == 1
     assert len(hass.add_job.mock_calls) == 0
@@ -342,7 +342,7 @@ async def test_async_create_task_eager_start_schedule_coroutine() -> None:
     async def job():
         pass
 
-    ha.HomeAssistant.async_create_task(hass, job(), eager_start=True)
+    ha.HomeAssistant.async_create_task_internal(hass, job(), eager_start=True)
     # Should create the task directly since 3.12 supports eager_start
     assert len(hass.loop.create_task.mock_calls) == 0
     assert len(hass.add_job.mock_calls) == 0
@@ -355,7 +355,7 @@ async def test_async_create_task_schedule_coroutine_with_name() -> None:
     async def job():
         pass
 
-    task = ha.HomeAssistant.async_create_task(
+    task = ha.HomeAssistant.async_create_task_internal(
         hass, job(), "named task", eager_start=False
     )
     assert len(hass.loop.call_soon.mock_calls) == 0
@@ -3480,3 +3480,15 @@ async def test_async_remove_thread_safety(hass: HomeAssistant) -> None:
         await hass.async_add_executor_job(
             hass.services.async_remove, "test_domain", "test_service"
         )
+
+
+async def test_async_create_task_thread_safety(hass: HomeAssistant) -> None:
+    """Test async_create_task thread safety."""
+
+    async def _any_coro():
+        pass
+
+    with pytest.raises(
+        RuntimeError, match="Detected code that calls async_create_task from a thread."
+    ):
+        await hass.async_add_executor_job(hass.async_create_task, _any_coro)