sandbox: route entries on every setup path — the router hook was bypassable at boot

setup.py's component-setup path (the path HA boot uses for every
existing entry, and the path async_setup_component-style tests use)
calls entry.async_setup_locked directly — it never went through
ConfigEntries.async_setup, so the router hook there was skipped and a
sandboxed entry would quietly run its integration locally on main at
every restart.

The router consult moves onto ConfigEntry.async_setup — the single
funnel every setup path uses (manager, boot/component setup, the
SETUP_RETRY timer) — so a routed entry can never slip into local setup;
the manager-level helpers are deleted. ConfigEntries.async_setup keeps
a shortcut for tagged entries so a routed custom (HACS) integration
with no code on main doesn't fail component setup before reaching the
router. Regression test drives async_setup_component end-to-end and
asserts the RPC crossed and local async_setup_entry never ran
(stash-verified to fail on the unfixed tree).

The compat lane now also reports how many entries were TAGGED next to
how many engaged, and run_compat classifies tagged==0 suites as 'main'
(camera/tts/system/ALWAYS_MAIN integrations legitimately measure
vanilla behavior) instead of lumping them into the suspicious no_op
bucket — no_op now precisely means 'tagged but never routed'.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QCotUYum6AoisyrxshoiJJ

homeassistant/config_entries.py

@@ -707,6 +707,23 @@ class ConfigEntry[_DataT = Any]:
         if self.source == SOURCE_IGNORE or self.disabled_by:
             return
 
+        # The router consult lives here — the single funnel every setup path
+        # uses (ConfigEntries.async_setup, setup.py's component-setup path at
+        # boot, the SETUP_RETRY timer) — so a routed entry can never slip
+        # into local setup. Core owns the state on both sides of the router
+        # contract: True marks LOADED, ConfigEntryError marks SETUP_ERROR
+        # with the message, None falls through to local setup.
+        if (router := hass.config_entries.router) is not None:
+            try:
+                result = await router.async_setup_entry(self)
+            except ConfigEntryError as err:
+                self._async_set_state(hass, ConfigEntryState.SETUP_ERROR, str(err))
+                return
+            if result is not None:
+                if result:
+                    self._async_set_state(hass, ConfigEntryState.LOADED, None)
+                return
+
         current_entry.set(self)
         try:
             await self.__async_setup_with_context(hass, integration)
@@ -2463,9 +2480,19 @@ class ConfigEntries:
                 f" be in the {ConfigEntryState.NOT_LOADED} state"
             )
 
-        if self.router is not None:
-            if (result := await self._async_router_setup(entry, _lock)) is not None:
-                return result
+        # A routed entry never needs its component set up on main — a custom
+        # (HACS) integration has no code here, so async_setup_component would
+        # fail before the entry ever reached the router. entry.async_setup
+        # performs the actual router consult.
+        if self.router is not None and entry.sandbox is not None:
+            if _lock:
+                async with entry.setup_lock:
+                    await entry.async_setup(self.hass)
+            else:
+                await entry.async_setup(self.hass)
+            return (
+                entry.state is ConfigEntryState.LOADED  # type: ignore[comparison-overlap]
+            )
 
         # Setup Component if not set up yet
         if entry.domain in self.hass.config.components:
@@ -2487,32 +2514,6 @@ class ConfigEntries:
             entry.state is ConfigEntryState.LOADED  # type: ignore[comparison-overlap]
         )
 
-    async def _async_router_setup(self, entry: ConfigEntry, _lock: bool) -> bool | None:
-        """Run the router's remote setup under the entry's setup lock.
-
-        Core owns the entry state on both sides of the router contract:
-        True marks LOADED, a ConfigEntryError marks SETUP_ERROR with the
-        message, None falls through to the default setup path (the lock is
-        released first, so the default path can take it again).
-        """
-        if _lock:
-            async with entry.setup_lock:
-                return await self._async_router_setup_locked(entry)
-        return await self._async_router_setup_locked(entry)
-
-    async def _async_router_setup_locked(self, entry: ConfigEntry) -> bool | None:
-        assert self.router is not None
-        try:
-            result = await self.router.async_setup_entry(entry)
-        except ConfigEntryError as err:
-            entry._async_set_state(  # noqa: SLF001
-                self.hass, ConfigEntryState.SETUP_ERROR, str(err)
-            )
-            return False
-        if result:
-            entry._async_set_state(self.hass, ConfigEntryState.LOADED, None)  # noqa: SLF001
-        return result
-
     async def async_unload(self, entry_id: str, _lock: bool = True) -> bool:
         """Unload a config entry."""
         entry = self.async_get_known_entry(entry_id)

sandbox/hass_client/hass_client/testing/_autotag.py

@@ -65,7 +65,7 @@ def classify_domain_sync(domain: str) -> str | None:
     return GROUP_BUILT_IN
 
 
-_ENGAGEMENT = {"entry_setups": 0}
+_ENGAGEMENT = {"entry_setups": 0, "tagged": 0}
 
 
 def engagement_count() -> int:
@@ -73,6 +73,16 @@ def engagement_count() -> int:
     return _ENGAGEMENT["entry_setups"]
 
 
+def tagged_count() -> int:
+    """How many MockConfigEntry instances the autotag routed to a sandbox.
+
+    Zero means the integration legitimately classifies to main (camera/
+    tts/ALWAYS_MAIN/system) — vanilla behavior is the correct measurement
+    and a zero engagement count is expected, not a lane regression.
+    """
+    return _ENGAGEMENT["tagged"]
+
+
 def install_router_engagement_counter() -> Callable[[], None]:
     """Count every router-driven sandbox entry setup.
 
@@ -150,6 +160,7 @@ def install_mock_config_entry_autotag() -> Callable[[], None]:
         if self.sandbox is None:
             group = classify_domain_sync(self.domain)
             if group is not None:
+                _ENGAGEMENT["tagged"] += 1
                 # ``ConfigEntry`` enforces ``sandbox`` updates via
                 # ``async_update_entry``; in tests the entry hasn't been
                 # registered yet so we mirror the framework's
@@ -174,4 +185,5 @@ __all__ = [
     "engagement_count",
     "install_mock_config_entry_autotag",
     "install_router_engagement_counter",
+    "tagged_count",
 ]

sandbox/hass_client/hass_client/testing/conftest_sandbox.py

@@ -37,7 +37,11 @@ from typing import TYPE_CHECKING, Any
 import pytest
 import pytest_asyncio
 
-from hass_client.testing._autotag import configure_compat_plugin, engagement_count
+from hass_client.testing._autotag import (
+    configure_compat_plugin,
+    engagement_count,
+    tagged_count,
+)
 
 if TYPE_CHECKING:
     from homeassistant.core import HomeAssistant
@@ -76,7 +80,8 @@ def pytest_terminal_summary(
 ) -> None:
     """Report how often the sandbox router actually engaged (see run_compat)."""
     terminalreporter.write_line(
-        f"sandbox-compat: router entry_setup engaged {engagement_count()} time(s)"
+        f"sandbox-compat: router entry_setup engaged {engagement_count()} time(s),"
+        f" {tagged_count()} entries tagged"
     )
 
 

sandbox/hass_client/hass_client/testing/pytest_plugin.py

@@ -33,7 +33,11 @@ import pytest
 import pytest_asyncio
 
 from hass_client.sandbox import SandboxRuntime
-from hass_client.testing._autotag import configure_compat_plugin, engagement_count
+from hass_client.testing._autotag import (
+    configure_compat_plugin,
+    engagement_count,
+    tagged_count,
+)
 from hass_client.testing._inproc import make_inproc_channel_pair
 
 if TYPE_CHECKING:
@@ -88,7 +92,8 @@ def pytest_terminal_summary(
     sets up config entries means the lane regressed to a no-op.
     """
     terminalreporter.write_line(
-        f"sandbox-compat: router entry_setup engaged {engagement_count()} time(s)"
+        f"sandbox-compat: router entry_setup engaged {engagement_count()} time(s),"
+        f" {tagged_count()} entries tagged"
     )
 
 

sandbox/run_compat.py

@@ -59,7 +59,10 @@ _SUMMARY_RE = {
     "skipped": re.compile(r"(\d+) skipped"),
 }
 
-_ENGAGED_RE = re.compile(r"sandbox-compat: router entry_setup engaged (\d+) time")
+_ENGAGED_RE = re.compile(
+    r"sandbox-compat: router entry_setup engaged (\d+) time\(s\),"
+    r" (\d+) entries tagged"
+)
 
 
 @dataclass
@@ -72,6 +75,7 @@ class Result:
     errors: int = 0
     skipped: int = 0
     engaged: int = 0
+    tagged: int = 0
     status: str = "no_tests"
 
     @property
@@ -135,13 +139,18 @@ def run_one(integration: str, plugin: str, *, timeout: float = 300.0) -> Result:
                 setattr(result, field, int(match.group(1)))
     if (match := _ENGAGED_RE.search(output)) is not None:
         result.engaged = int(match.group(1))
+        result.tagged = int(match.group(2))
 
     if result.total == 0:
         result.status = "no_tests"
     elif result.failed != 0 or result.errors != 0:
         result.status = "issues"
+    elif result.tagged == 0:
+        # The integration classifies to main (camera/tts/system/ALWAYS_MAIN)
+        # — vanilla behavior is the correct measurement here.
+        result.status = "main"
     elif result.engaged == 0 and result.passed > 0:
-        # Tests passed but nothing ever routed through a sandbox — the
+        # Entries were tagged for a sandbox but nothing ever routed — the
         # plugin regressed to a no-op; do NOT report this as compatibility.
         result.status = "no_op"
     else:
@@ -178,6 +187,7 @@ def write_report(results: list[Result], plugin: str, path: Path) -> None:
     """Write a short Markdown summary suitable for review."""
     counts: dict[str, int] = {
         "pass": 0,
+        "main": 0,
         "issues": 0,
         "timeout": 0,
         "no_tests": 0,
@@ -198,7 +208,8 @@ def write_report(results: list[Result], plugin: str, path: Path) -> None:
         "",
         "## Summary",
         "",
-        f"- Integrations passing: **{counts.get('pass', 0)}**",
+        f"- Integrations passing (sandboxed): **{counts.get('pass', 0)}**",
+        f"- Integrations on main by classification: **{counts.get('main', 0)}**",
         f"- Integrations with issues: **{counts.get('issues', 0)}**",
         f"- No-op runs (sandbox never engaged): **{counts.get('no_op', 0)}**",
         f"- Timeouts: **{counts.get('timeout', 0)}**",
@@ -301,7 +312,9 @@ def main(argv: list[str] | None = None) -> int:
     print(f"\nWrote {args.csv}")
     print(f"Wrote {args.report}")
 
-    if results and all(result.engaged == 0 for result in results):
+    if results and all(
+        result.engaged == 0 for result in results if result.tagged
+    ) and any(result.tagged for result in results):
         print(
             "ERROR: no test in the entire run routed an entry through a"
             " sandbox — the compat lane is a no-op.",

tests/components/sandbox/test_router.py

@@ -22,10 +22,11 @@ from homeassistant.config_entries import (
     ConfigFlowContext,
 )
 from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
 
 from ._helpers import FakeSandboxManager, make_channel_pair
 
-from tests.common import MockConfigEntry, MockModule, mock_integration
+from tests.common import MockConfigEntry, MockModule, mock_integration, mock_platform
 
 
 @pytest.fixture(name="manager")
@@ -263,3 +264,56 @@ async def test_async_setup_entry_returns_none_when_not_sandboxed(
 
     assert result is None
     assert manager.start_calls == []
+
+
+async def test_component_setup_path_routes_tagged_entries(
+    hass: HomeAssistant, manager: FakeSandboxManager
+) -> None:
+    """setup.py's per-entry boot path consults the router.
+
+    HA boot (and any ``async_setup_component``) sets existing entries up via
+    ``entry.async_setup_locked`` — never ``ConfigEntries.async_setup`` — so
+    the router consult must live on ``ConfigEntry.async_setup`` or a
+    sandboxed entry silently runs its integration on main at every restart.
+    """
+    channel_a, channel_b = make_channel_pair()
+    received: list[pb.EntrySetup] = []
+
+    async def _entry_setup(payload: pb.EntrySetup) -> pb.EntrySetupResult:
+        received.append(payload)
+        return pb.EntrySetupResult(ok=True)
+
+    channel_b.register("sandbox/entry_setup", _entry_setup)
+    channel_a.start()
+    channel_b.start()
+    manager.install("built-in", channel_a)
+
+    local_setup_calls: list[str] = []
+
+    async def _local_setup_entry(hass_: HomeAssistant, entry_: ConfigEntry) -> bool:
+        local_setup_calls.append(entry_.entry_id)
+        return True
+
+    mock_integration(
+        hass,
+        MockModule("test_bootpath", async_setup_entry=_local_setup_entry),
+    )
+    mock_platform(hass, "test_bootpath.config_flow", None)
+    entry = MockConfigEntry(
+        domain="test_bootpath",
+        title="Boot",
+        sandbox="built-in",
+    )
+    entry.add_to_hass(hass)
+    hass.config_entries.router = SandboxFlowRouter(hass, cast(SandboxManager, manager))
+
+    try:
+        assert await async_setup_component(hass, "test_bootpath", {})
+        await hass.async_block_till_done()
+    finally:
+        await channel_a.close()
+        await channel_b.close()
+
+    assert [msg.entry_id for msg in received] == [entry.entry_id]
+    assert local_setup_calls == []
+    assert entry.state is ConfigEntryState.LOADED