sandbox: load registries in the private hass + make the compat lane engage the sandbox

Two critical review findings (plan-review-overhead E1+E2), each the test
that would have caught the other:

E1 — FlowRunner.create built a bare HomeAssistant and never ran the
registry loads bootstrap does, so er.async_get returned an unloaded
EntityRegistry with no .entities and every real EntityPlatform add died
with AttributeError while entry_setup still ACKed ok — the sandbox
bridged zero entities for real integrations. Load the
area/category/device/entity/floor/issue/label registries in create(),
before the channel opens, so their Stores bind to the local tempdir.
Regression-tested by driving the real sun integration through
EntryRunner end-to-end.

E2 — both compat pytest plugins only installed the entry autotag; the
sandbox_inprocess/sandbox_subprocess fixtures were opt-in and no vanilla
test requests them, so hass.config_entries.router stayed None and every
tagged entry set up locally: COMPAT.md's 99.97% measured vanilla tests,
not sandbox compatibility. The plugins now inject their sandbox fixture
into every hass-using test at collection, count router-driven
entry_setups via a test-side wrapper, and report the count in the
terminal summary; run_compat.py records the count per integration,
marks green-but-never-engaged rows no_op, and fails the run outright if
nothing engaged. Probe: sun now runs 10 routed entry setups, 98/101
passing (3 real compat gaps for the lane to report).

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

sandbox/hass_client/hass_client/flow_runner.py

@@ -16,6 +16,7 @@ that can't be serialised. The docstring in ``_marshal_result`` is the
 load-bearing note for how the schema is later marshalled.
 """
 
+import asyncio
 from collections.abc import Callable, Mapping
 import contextlib
 import ipaddress
@@ -30,6 +31,18 @@ from homeassistant.config_entries import (
 )
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import BaseServiceInfo, FlowResultType, UnknownFlow
+from homeassistant.helpers import (
+    area_registry as ar,
+    category_registry as cr,
+    device_registry as dr,
+    entity,
+    entity_registry as er,
+    floor_registry as fr,
+    frame,
+    issue_registry as ir,
+    label_registry as lr,
+    translation,
+)
 from homeassistant.helpers.discovery_flow import DiscoveryKey
 from homeassistant.helpers.service_info.dhcp import DhcpServiceInfo
 from homeassistant.helpers.service_info.hassio import HassioServiceInfo
@@ -111,6 +124,25 @@ class FlowRunner:
         # has built its default one, so we inherit all the wiring.
         hass.config_entries.flow = _SandboxFlowManager(hass, hass.config_entries, {})
         loader.async_setup(hass)
+        # The registry subset of bootstrap's async_load_base_functionality —
+        # a bare hass never runs bootstrap, and an unloaded EntityRegistry
+        # has no ``.entities``, so a real EntityPlatform cannot add a single
+        # entity without these. Runs before the runtime opens the channel and
+        # sets ``current_sandbox``, so registry Stores bind to the sandbox's
+        # local tempdir (fresh per process), never route to main.
+        entity.async_setup(hass)
+        frame.async_setup(hass)
+        translation.async_setup(hass)
+        dr.async_setup(hass)
+        await asyncio.gather(
+            ar.async_load(hass),
+            cr.async_load(hass),
+            dr.async_load(hass),
+            er.async_load(hass),
+            fr.async_load(hass),
+            ir.async_load(hass),
+            lr.async_load(hass),
+        )
         return cls(hass)
 
     def register(self, channel: Channel) -> None:

sandbox/hass_client/hass_client/sandbox/__init__.py

@@ -162,14 +162,14 @@ class SandboxRuntime:
             # every coroutine the runtime spawns inherits it (asyncio copies
             # the context at `create_task` time).
             #
-            # Ordering caveat (see the plan's touch-points audit): registries
-            # whose `Store` is constructed AND first loaded inside
-            # `FlowRunner.create` already ran their `async_load` against the
-            # sandbox tempdir before this point, so they keep their local
-            # file backing. `restore_state`'s `async_load` runs *after* this
-            # set, so it routes to main — which is what we want. If a future
-            # refactor moves a registry's first `async_load` to straddle this
-            # line, that registry would silently start routing to main.
+            # Ordering caveat: `FlowRunner.create` explicitly loads the
+            # area/category/device/entity/floor/issue/label registries before
+            # this point, so their Stores bind to the sandbox tempdir and keep
+            # their local file backing. `restore_state`'s `async_load` runs
+            # *after* this set, so it routes to main — which is what we want.
+            # If a future refactor moves a registry's first `async_load` to
+            # straddle this line, that registry would silently start routing
+            # to main.
             assert current_sandbox.get() is None, (
                 "current_sandbox already set — two sandbox runtimes sharing "
                 "one event loop? (see plan Risk #3)"

sandbox/hass_client/hass_client/testing/_autotag.py

@@ -65,6 +65,66 @@ def classify_domain_sync(domain: str) -> str | None:
     return GROUP_BUILT_IN
 
 
+_ENGAGEMENT = {"entry_setups": 0}
+
+
+def engagement_count() -> int:
+    """Router-driven sandbox ``entry_setup`` count for this pytest session."""
+    return _ENGAGEMENT["entry_setups"]
+
+
+def install_router_engagement_counter() -> Callable[[], None]:
+    """Count every router-driven sandbox entry setup.
+
+    The compat lane's honesty guard: a run whose counter stays at zero
+    never routed anything through a sandbox, so ``run_compat.py`` can
+    flag it as a no-op instead of reporting vanilla-test results as
+    sandbox compatibility. Idempotent; returns an unpatch callable.
+    """
+    # Lazy import: the HA integration tree must not load at plugin
+    # import time (same rule as the classifier import below).
+    from homeassistant.components.sandbox.router import (  # noqa: PLC0415
+        SandboxFlowRouter,
+    )
+
+    if getattr(SandboxFlowRouter, "_sandbox_counter_patched", False):
+        return lambda: None
+
+    original = SandboxFlowRouter.async_setup_entry
+
+    async def counted(self: Any, entry: Any) -> bool | None:
+        result = await original(self, entry)
+        if result is not None:
+            _ENGAGEMENT["entry_setups"] += 1
+        return result
+
+    SandboxFlowRouter.async_setup_entry = counted
+    SandboxFlowRouter._sandbox_counter_patched = True  # noqa: SLF001
+
+    def restore() -> None:
+        SandboxFlowRouter.async_setup_entry = original
+        with contextlib.suppress(AttributeError):
+            delattr(SandboxFlowRouter, "_sandbox_counter_patched")
+
+    return restore
+
+
+def configure_compat_plugin() -> Callable[[], None]:
+    """Install the autotag patch + engagement counter; returns the undo.
+
+    Shared by both compat plugins' ``pytest_configure`` so the install /
+    restore pairing lives in one place.
+    """
+    unpatch_autotag = install_mock_config_entry_autotag()
+    unpatch_counter = install_router_engagement_counter()
+
+    def restore() -> None:
+        unpatch_counter()
+        unpatch_autotag()
+
+    return restore
+
+
 def install_mock_config_entry_autotag() -> Callable[[], None]:
     """Patch :meth:`MockConfigEntry.add_to_hass` to inject the sandbox group.
 
@@ -103,4 +163,10 @@ def install_mock_config_entry_autotag() -> Callable[[], None]:
     return restore
 
 
-__all__ = ["classify_domain_sync", "install_mock_config_entry_autotag"]
+__all__ = [
+    "classify_domain_sync",
+    "configure_compat_plugin",
+    "engagement_count",
+    "install_mock_config_entry_autotag",
+    "install_router_engagement_counter",
+]

sandbox/hass_client/hass_client/testing/conftest_sandbox.py

@@ -32,12 +32,12 @@ a :class:`SubprocessSandbox` handle once the subprocess is running.
 from collections.abc import AsyncIterator, Callable
 from dataclasses import dataclass
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import pytest
 import pytest_asyncio
 
-from hass_client.testing._autotag import install_mock_config_entry_autotag
+from hass_client.testing._autotag import configure_compat_plugin, engagement_count
 
 if TYPE_CHECKING:
     from homeassistant.core import HomeAssistant
@@ -48,27 +48,36 @@ DEFAULT_GROUP = "built-in"
 
 _MARK_NO_FREEZER = "no_sandbox_freezer"
 
-_unpatch_autotag: Callable[[], None] | None = None
+_unconfigure: Callable[[], None] | None = None
 
 
 def pytest_configure(config: pytest.Config) -> None:
-    """Register the ``no_sandbox_freezer`` marker and install the autotag patch."""
+    """Register the freezer marker; install autotag + engagement counter."""
     config.addinivalue_line(
         "markers",
         f"{_MARK_NO_FREEZER}: skip the test when the real-subprocess sandbox"
         " plugin is active (freezer + subprocess clock skew hangs the channel)",
     )
-    global _unpatch_autotag  # noqa: PLW0603
-    if _unpatch_autotag is None:
-        _unpatch_autotag = install_mock_config_entry_autotag()
+    global _unconfigure  # noqa: PLW0603
+    if _unconfigure is None:
+        _unconfigure = configure_compat_plugin()
 
 
 def pytest_unconfigure(config: pytest.Config) -> None:
-    """Restore the original ``MockConfigEntry.add_to_hass`` on session exit."""
-    global _unpatch_autotag  # noqa: PLW0603
-    if _unpatch_autotag is not None:
-        _unpatch_autotag()
-        _unpatch_autotag = None
+    """Restore the patched hooks on session exit."""
+    global _unconfigure  # noqa: PLW0603
+    if _unconfigure is not None:
+        _unconfigure()
+        _unconfigure = None
+
+
+def pytest_terminal_summary(
+    terminalreporter: Any, exitstatus: int, config: pytest.Config
+) -> 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)"
+    )
 
 
 def pytest_collection_modifyitems(
@@ -95,9 +104,17 @@ def pytest_collection_modifyitems(
         if item.get_closest_marker(_MARK_NO_FREEZER) is not None:
             item.add_marker(skip_freezer)
             continue
-        fixtures = getattr(item, "fixturenames", ())
+        fixtures = getattr(item, "fixturenames", None)
+        if fixtures is None:
+            continue
         if "freezer" in fixtures:
             item.add_marker(skip_freezer)
+            continue
+        # Inject the subprocess sandbox into every hass-using test — the
+        # autotag alone leaves ``hass.config_entries.router`` None, and a
+        # tagged entry then quietly sets up locally (lane no-op).
+        if "hass" in fixtures and "sandbox_subprocess" not in fixtures:
+            fixtures.append("sandbox_subprocess")
 
 
 @dataclass

sandbox/hass_client/hass_client/testing/pytest_plugin.py

@@ -33,7 +33,7 @@ import pytest
 import pytest_asyncio
 
 from hass_client.sandbox import SandboxRuntime
-from hass_client.testing._autotag import install_mock_config_entry_autotag
+from hass_client.testing._autotag import configure_compat_plugin, engagement_count
 from hass_client.testing._inproc import make_inproc_channel_pair
 
 if TYPE_CHECKING:
@@ -43,22 +43,53 @@ _LOGGER = logging.getLogger(__name__)
 
 DEFAULT_GROUP = "built-in"
 
-_unpatch_autotag: Callable[[], None] | None = None
+_unconfigure: Callable[[], None] | None = None
 
 
 def pytest_configure(config: pytest.Config) -> None:
-    """Patch ``MockConfigEntry.add_to_hass`` so the classifier path fires."""
-    global _unpatch_autotag  # noqa: PLW0603
-    if _unpatch_autotag is None:
-        _unpatch_autotag = install_mock_config_entry_autotag()
+    """Install the autotag patch + router engagement counter."""
+    global _unconfigure  # noqa: PLW0603
+    if _unconfigure is None:
+        _unconfigure = configure_compat_plugin()
 
 
 def pytest_unconfigure(config: pytest.Config) -> None:
-    """Restore the original ``MockConfigEntry.add_to_hass`` on session exit."""
-    global _unpatch_autotag  # noqa: PLW0603
-    if _unpatch_autotag is not None:
-        _unpatch_autotag()
-        _unpatch_autotag = None
+    """Restore the patched hooks on session exit."""
+    global _unconfigure  # noqa: PLW0603
+    if _unconfigure is not None:
+        _unconfigure()
+        _unconfigure = None
+
+
+def pytest_collection_modifyitems(
+    config: pytest.Config, items: list[pytest.Item]
+) -> None:
+    """Inject the in-process sandbox into every ``hass``-using test.
+
+    The autotag alone only marks entries with a group; with no sandbox
+    set up, ``hass.config_entries.router`` stays ``None`` and every
+    tagged entry quietly sets up locally — the lane measures nothing.
+    Requesting the fixture here makes the router path real for each test.
+    """
+    for item in items:
+        fixtures = getattr(item, "fixturenames", None)
+        if fixtures is None or "hass" not in fixtures:
+            continue
+        if "sandbox_inprocess" not in fixtures:
+            fixtures.append("sandbox_inprocess")
+
+
+def pytest_terminal_summary(
+    terminalreporter: Any, exitstatus: int, config: pytest.Config
+) -> None:
+    """Report how often the sandbox router actually engaged.
+
+    ``run_compat.py`` parses this line; zero engagements on a suite that
+    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)"
+    )
 
 
 @dataclass

sandbox/hass_client/tests/test_entity_bridge.py

@@ -567,12 +567,8 @@ async def test_device_registry_update_resends_linked_entities(
     sandbox.start()
 
     # Link the entity to a device in the registry so the device-update
-    # handler can find it via its device_id. The sandbox-private hass does
-    # not bootstrap the registries, so set them up explicitly and register a
-    # config entry the device can hang off.
-    dr.async_setup(hass)
-    await dr.async_load(hass, load_empty=True)
-    await er.async_load(hass, load_empty=True)
+    # handler can find it via its device_id (FlowRunner.create loads the
+    # registries), and register a config entry the device can hang off.
     config_entry = ConfigEntry(
         version=1,
         minor_version=1,

sandbox/hass_client/tests/test_entry_runner.py

@@ -20,6 +20,7 @@ import pytest
 from homeassistant import config_entries as ha_config_entries, loader as ha_loader
 from homeassistant.config_entries import ConfigEntry, ConfigFlow
 from homeassistant.exceptions import ServiceValidationError
+from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.entity_component import DATA_INSTANCES
 
 
@@ -390,3 +391,37 @@ async def test_entity_query_method_raises(
     with pytest.raises(ChannelRemoteError) as err:
         await main.call("sandbox/entity_query", msg)
     assert err.value.error_type == "ServiceValidationError"
+
+
+async def test_entry_setup_real_platform_adds_entities(
+    channels: tuple[Channel, Channel], runner: EntryRunner
+) -> None:
+    """A real integration's entity platform adds entities on the private hass.
+
+    Regression test for the missing registry loads: an unloaded
+    ``EntityRegistry`` has no ``.entities``, so every
+    ``EntityPlatform._async_add_entity`` died with ``AttributeError`` while
+    ``entry_setup`` still reported ok — the sandbox bridged zero entities.
+    Driving the real ``sun`` integration end-to-end pins the whole path.
+    """
+    main, sandbox = channels
+    runner.register(sandbox)
+    main.start()
+    sandbox.start()
+
+    payload = pb.EntrySetup(
+        entry_id="sun_entry",
+        domain="sun",
+        title="Sun",
+        source="user",
+        version=1,
+        minor_version=1,
+    )
+    result = await main.call("sandbox/entry_setup", payload)
+    assert result.ok, result.reason
+
+    hass = runner.hass
+    await hass.async_block_till_done()
+    assert hass.states.get("sun.sun") is not None
+    ent_reg = er.async_get(hass)
+    assert ent_reg.async_get_entity_id("sensor", "sun", "sun_entry-next_dawn")

sandbox/run_compat.py

@@ -58,6 +58,8 @@ _SUMMARY_RE = {
     "skipped": re.compile(r"(\d+) skipped"),
 }
 
+_ENGAGED_RE = re.compile(r"sandbox-compat: router entry_setup engaged (\d+) time")
+
 
 @dataclass
 class Result:
@@ -68,6 +70,7 @@ class Result:
     failed: int = 0
     errors: int = 0
     skipped: int = 0
+    engaged: int = 0
     status: str = "no_tests"
 
     @property
@@ -129,13 +132,19 @@ def run_one(integration: str, plugin: str, *, timeout: float = 300.0) -> Result:
         for field, pattern in _SUMMARY_RE.items():
             if (match := pattern.search(line)) is not None:
                 setattr(result, field, int(match.group(1)))
+    if (match := _ENGAGED_RE.search(output)) is not None:
+        result.engaged = int(match.group(1))
 
     if result.total == 0:
         result.status = "no_tests"
-    elif result.failed == 0 and result.errors == 0:
-        result.status = "pass"
-    else:
+    elif result.failed != 0 or result.errors != 0:
         result.status = "issues"
+    elif result.engaged == 0 and result.passed > 0:
+        # Tests passed but nothing ever routed through a sandbox — the
+        # plugin regressed to a no-op; do NOT report this as compatibility.
+        result.status = "no_op"
+    else:
+        result.status = "pass"
 
     if result.status in ("issues", "timeout"):
         ERRORS_DIR.mkdir(parents=True, exist_ok=True)
@@ -147,7 +156,9 @@ def write_csv(results: list[Result], path: Path) -> None:
     """Persist per-integration results as CSV."""
     with path.open("w", newline="") as fh:
         writer = csv.writer(fh)
-        writer.writerow(["integration", "status", "passed", "failed", "errors", "skipped"])
+        writer.writerow(
+            ["integration", "status", "passed", "failed", "errors", "skipped", "engaged"]
+        )
         for result in results:
             writer.writerow(
                 [
@@ -157,13 +168,20 @@ def write_csv(results: list[Result], path: Path) -> None:
                     result.failed,
                     result.errors,
                     result.skipped,
+                    result.engaged,
                 ]
             )
 
 
 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, "issues": 0, "timeout": 0, "no_tests": 0}
+    counts: dict[str, int] = {
+        "pass": 0,
+        "issues": 0,
+        "timeout": 0,
+        "no_tests": 0,
+        "no_op": 0,
+    }
     totals = {"passed": 0, "failed": 0, "errors": 0, "skipped": 0}
     for result in results:
         counts[result.status] = counts.get(result.status, 0) + 1
@@ -181,6 +199,7 @@ def write_report(results: list[Result], plugin: str, path: Path) -> None:
         "",
         f"- Integrations passing: **{counts.get('pass', 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)}**",
         f"- No tests collected: **{counts.get('no_tests', 0)}**",
         "",
@@ -191,13 +210,14 @@ def write_report(results: list[Result], plugin: str, path: Path) -> None:
         "",
         "## Per-integration results",
         "",
-        "| integration | status | passed | failed | errors | skipped |",
-        "| --- | --- | ---: | ---: | ---: | ---: |",
+        "| integration | status | passed | failed | errors | skipped | engaged |",
+        "| --- | --- | ---: | ---: | ---: | ---: | ---: |",
     ]
     for result in results:
         lines.append(
             f"| {result.integration} | {result.status} | {result.passed} |"
             f" {result.failed} | {result.errors} | {result.skipped} |"
+            f" {result.engaged} |"
         )
     path.write_text("\n".join(lines) + "\n")
 
@@ -262,6 +282,14 @@ def main(argv: list[str] | None = None) -> int:
     write_report(results, plugin, args.report)
     print(f"\nWrote {args.csv}")
     print(f"Wrote {args.report}")
+
+    if results and all(result.engaged == 0 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.",
+            file=sys.stderr,
+        )
+        return 1
     return 0
 
 

tests/components/sandbox/test_testing_plugins.py

@@ -100,7 +100,7 @@ def _make_item(
 ) -> MagicMock:
     """Construct a fake pytest item with the given fixtures/markers."""
     item = MagicMock(spec=pytest.Item)
-    item.fixturenames = fixturenames
+    item.fixturenames = list(fixturenames)
     item.get_closest_marker.side_effect = lambda name: (
         MagicMock() if name in markers else None
     )
@@ -128,11 +128,35 @@ def test_conftest_sandbox_skips_marker_tagged_tests() -> None:
     assert args[0].name == "skip"
 
 
-def test_conftest_sandbox_leaves_unrelated_tests_alone() -> None:
-    """Tests without the freezer fixture or marker are left untouched."""
+def test_conftest_sandbox_injects_sandbox_into_hass_tests() -> None:
+    """A hass-using test gets the ``sandbox_subprocess`` fixture injected.
+
+    Without the injection the plugin only tags entries — the router stays
+    ``None`` and the compat lane is a no-op.
+    """
     item = _make_item(fixturenames=("hass",))
     cs_plugin.pytest_collection_modifyitems(MagicMock(), [item])
     assert not item.add_marker.called
+    assert "sandbox_subprocess" in item.fixturenames
+
+
+def test_conftest_sandbox_leaves_non_hass_tests_alone() -> None:
+    """Tests without hass, the freezer fixture, or the marker are untouched."""
+    item = _make_item(fixturenames=("tmp_path",))
+    cs_plugin.pytest_collection_modifyitems(MagicMock(), [item])
+    assert not item.add_marker.called
+    assert item.fixturenames == ["tmp_path"]
+
+
+def test_inprocess_plugin_injects_sandbox_into_hass_tests() -> None:
+    """The in-process plugin injects ``sandbox_inprocess`` the same way."""
+    from hass_client.testing import pytest_plugin as inproc_plugin  # noqa: PLC0415
+
+    item = _make_item(fixturenames=("hass",))
+    other = _make_item(fixturenames=("tmp_path",))
+    inproc_plugin.pytest_collection_modifyitems(MagicMock(), [item, other])
+    assert "sandbox_inprocess" in item.fixturenames
+    assert other.fixturenames == ["tmp_path"]
 
 
 def test_autotag_sets_mock_config_entry_sandbox() -> None: