diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index c0c8dd4cce78..eb89b351924e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -76,7 +76,8 @@ repos: stages: [manual] files: ^sandbox/proto/sandbox\.proto$ # Drift guard for the hand-mirrored sandbox wire modules (channel.py, - # codec_protobuf.py, messages.py). A plain byte-for-byte diff with no + # codec_protobuf.py, messages.py) and the checked-in protobuf gencode + # pair (_proto/sandbox_pb2.py/.pyi). A plain byte-for-byte diff with no # external tooling, so — unlike the proto gencode guard above — it runs as # a regular hook whenever either copy of a mirrored file changes. - id: sandbox-mirror-drift @@ -84,7 +85,7 @@ repos: entry: sandbox/proto/check_mirror_drift.sh language: script pass_filenames: false - files: ^(homeassistant/components/sandbox|sandbox/hass_client/hass_client)/(channel|codec_protobuf|messages)\.py$ + files: ^(homeassistant/components/sandbox|sandbox/hass_client/hass_client)/((channel|codec_protobuf|messages)\.py|_proto/sandbox_pb2\.(py|pyi))$ # Run mypy through our wrapper script in order to get the possible # pyenv and/or virtualenv activated; it may not have been e.g. if # committing from a GUI tool that was not launched from an activated diff --git a/homeassistant/components/sandbox/__init__.py b/homeassistant/components/sandbox/__init__.py index 22111e0ed33c..330b4b3b0b93 100644 --- a/homeassistant/components/sandbox/__init__.py +++ b/homeassistant/components/sandbox/__init__.py @@ -46,7 +46,6 @@ class SandboxData: manager: SandboxManager | None = None router: SandboxFlowRouter | None = None - channels: dict[str, Channel] = field(default_factory=dict) bridges: dict[str, SandboxBridge] = field(default_factory=dict) # A bridge displaced by a restart, held until the fresh process goes # ready so its proxies + platform slots can be torn down right before @@ -68,7 +67,6 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: # re-registers, so the entities stay visible (Phase 2 marks them # unavailable) across the restart gap instead of vanishing. old_bridge = data.bridges.get(group) - data.channels[group] = channel data.bridges[group] = async_create_bridge(hass, group=group, channel=channel) if old_bridge is None: return @@ -177,7 +175,6 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: unregister_translation_provider() await manager.async_graceful_shutdown_all(timeout=manager.shutdown_grace) await manager.async_stop_all() - data.channels.clear() data.bridges.clear() data.pending_teardown.clear() diff --git a/homeassistant/components/sandbox/bridge.py b/homeassistant/components/sandbox/bridge.py index bc44d862770a..da724375db09 100644 --- a/homeassistant/components/sandbox/bridge.py +++ b/homeassistant/components/sandbox/bridge.py @@ -76,8 +76,7 @@ from homeassistant.util.file import write_utf8_file_atomic from ._proto import sandbox_pb2 as pb from .channel import Channel, ChannelClosedError, ChannelRemoteError from .const import UNIQUE_ID_SEPARATOR -from .messages import decode_json, decode_json_dict, encode_json -from .protocol import ( +from .messages import ( MSG_CALL_SERVICE, MSG_ENTITY_QUERY, MSG_FIRE_EVENT, @@ -89,6 +88,9 @@ from .protocol import ( MSG_STORE_SAVE, MSG_UNREGISTER_ENTITY, MSG_UNREGISTER_SERVICE, + decode_json, + decode_json_dict, + encode_json, ) from .schema_bridge import reconstruct_schema @@ -156,7 +158,6 @@ class SandboxEntityDescription: initial_state: str | None = None initial_attributes: dict[str, Any] = field(default_factory=dict) device_info: dict[str, Any] | None = None - device_id: str | None = None @classmethod def from_proto(cls, msg: pb.EntityDescription) -> SandboxEntityDescription: @@ -478,7 +479,7 @@ class SandboxBridge: # scoped to the (now-verified-owned) entry. self._reject_foreign_device_merge(device_registry, description) try: - device = device_registry.async_get_or_create( + device_registry.async_get_or_create( config_entry_id=description.entry_id, **description.device_info, ) @@ -487,7 +488,6 @@ class SandboxBridge: f"register_entity: invalid device_info for " f"{description.sandbox_entity_id!r}: {err}" ) from err - description.device_id = device.id # MSG_REGISTER_ENTITY is an upsert: a re-send for an already-tracked # entity (the client re-describes on registry/device updates) refreshes # the existing proxy in place rather than adding a duplicate. The @@ -820,9 +820,7 @@ class SandboxBridge: """ entry_ids = {eid for (eid, _domain) in list(self._platforms)} entry_ids.update( - entry_id - for proxy in list(self._entities.values()) - if (entry_id := getattr(proxy.description, "entry_id", None)) is not None + proxy.description.entry_id for proxy in list(self._entities.values()) ) for entry_id in entry_ids: await self._async_teardown_entry(entry_id) @@ -860,7 +858,7 @@ class SandboxBridge: self._entities = { sid: proxy for sid, proxy in self._entities.items() - if getattr(proxy.description, "entry_id", None) != entry_id + if proxy.description.entry_id != entry_id } @@ -1060,14 +1058,10 @@ def _deserialise_device_info(info: pb.DeviceInfo) -> dict[str, Any] | None: return out or None -def _parse_supports_response(value: Any) -> SupportsResponse: +def _parse_supports_response(value: str) -> SupportsResponse: """Coerce the wire ``supports_response`` field into the enum.""" - if isinstance(value, SupportsResponse): - return value - if value is None: - return SupportsResponse.NONE try: - return SupportsResponse(str(value).lower()) + return SupportsResponse(value.lower()) except ValueError: return SupportsResponse.NONE @@ -1152,9 +1146,7 @@ def _translate_remote_error(err: ChannelRemoteError) -> Exception: msg = err.error if name in {"Invalid", "MultipleInvalid"}: return TypeError(msg) - if name in {"ServiceNotFound", "ServiceValidationError"}: - return HomeAssistantError(msg) - if name == "HomeAssistantError": + if name in {"ServiceNotFound", "ServiceValidationError", "HomeAssistantError"}: return HomeAssistantError(msg) return HomeAssistantError(f"sandbox error ({name or 'unknown'}): {msg}") diff --git a/homeassistant/components/sandbox/channel.py b/homeassistant/components/sandbox/channel.py index 9f3a9488b4c3..2428a2fb7939 100644 --- a/homeassistant/components/sandbox/channel.py +++ b/homeassistant/components/sandbox/channel.py @@ -16,7 +16,7 @@ dispatch core: :class:`StreamTransport` length-prefixes each frame (4-byte big-endian length + body) over an :class:`asyncio.StreamReader` / :class:`asyncio.StreamWriter` pair (stdio, unix socket). A future - ``WebSocketTransport`` drops in via :meth:`Channel.from_transport` using + ``WebSocketTransport`` drops in via ``Channel(transport=...)`` using aiohttp's native binary framing. The :class:`Frame` shape mirrors the three message kinds that cross the @@ -309,8 +309,7 @@ class Channel: The common case passes a ``reader``/``writer`` pair, framed with :class:`StreamTransport` (length-prefixed). To run over a non-stream - transport (e.g. websockets), pass ``transport=`` instead — see - :meth:`from_transport`. + transport (e.g. websockets), pass ``transport=`` instead. ``codec`` is required — production passes :class:`~.codec_protobuf.ProtobufCodec`; a forgotten codec is a @@ -340,24 +339,6 @@ class Channel: self._inflight_sem = asyncio.Semaphore(max_inflight) self._max_queued = max_queued - @classmethod - def from_transport( - cls, - transport: Transport, - *, - codec: Codec, - name: str = "channel", - max_inflight: int = DEFAULT_MAX_INFLIGHT, - ) -> Channel: - """Build a channel over an arbitrary :class:`Transport`. - - This is the seam a future ``WebSocketTransport`` drops into — the - dispatch core is identical regardless of how frames reach the wire. - """ - return cls( - transport=transport, codec=codec, name=name, max_inflight=max_inflight - ) - @property def closed(self) -> bool: """Return True once the channel has been closed.""" @@ -400,7 +381,8 @@ class Channel: if self._closed: raise ChannelClosedError(f"channel {self._name!r} is closed") call_id = self._next_id - self._next_id += 1 + # Wrap within the uint32 wire field, skipping 0 (id 0 marks a push). + self._next_id = self._next_id % 0xFFFFFFFF + 1 future: asyncio.Future[Any] = asyncio.get_running_loop().create_future() self._pending[call_id] = future try: @@ -463,8 +445,7 @@ class Channel: ``timeout`` — any handler still running afterwards is left for ``close()`` to cancel. Does not itself cancel anything. """ - inflight = [task for task in self._inflight if task is not self._reader_task] - if inflight: + if inflight := list(self._inflight): await asyncio.wait(inflight, timeout=timeout) async def _write(self, frame: Frame) -> None: diff --git a/homeassistant/components/sandbox/codec_protobuf.py b/homeassistant/components/sandbox/codec_protobuf.py index eea043e09e76..80b77c518f46 100644 --- a/homeassistant/components/sandbox/codec_protobuf.py +++ b/homeassistant/components/sandbox/codec_protobuf.py @@ -74,9 +74,13 @@ class ProtobufCodec: def _serialize_body(body: Any, cls: type[Message] | None) -> bytes: - """Serialise a proto-message body; ``None`` becomes an empty message.""" + """Serialise a proto-message body; ``None`` becomes an empty message. + + An empty proto message serialises to zero bytes, so ``None`` maps to + ``b""`` whether or not the type has a registered class. + """ if body is None: - return cls().SerializeToString() if cls is not None else b"" + return b"" if isinstance(body, Message): return body.SerializeToString() raise TypeError( diff --git a/homeassistant/components/sandbox/manager.py b/homeassistant/components/sandbox/manager.py index 1ee58857d8a9..c68bfebab324 100644 --- a/homeassistant/components/sandbox/manager.py +++ b/homeassistant/components/sandbox/manager.py @@ -40,7 +40,7 @@ from homeassistant.core import HomeAssistant from .channel import Channel, ChannelClosedError, ChannelRemoteError from .codec_protobuf import ProtobufCodec -from .protocol import MSG_READY, MSG_SHUTDOWN +from .messages import MSG_READY, MSG_SHUTDOWN _LOGGER = logging.getLogger(__name__) diff --git a/homeassistant/components/sandbox/messages.py b/homeassistant/components/sandbox/messages.py index 1fd4a9a7e6cf..066ab650604a 100644 --- a/homeassistant/components/sandbox/messages.py +++ b/homeassistant/components/sandbox/messages.py @@ -1,23 +1,115 @@ -"""Typed protobuf message registry + dynamic-payload JSON codec. +"""Wire-protocol constants, typed proto registry + dynamic-payload JSON codec. -This module is the codec's view of the wire: the ``type → (request_cls, -result_cls)`` registry plus the single encoder/decoder pair for the genuinely -dynamic payloads (service_data, target, state attributes, capabilities, the -wrapped Store envelope, flow ``data``/``errors``/``context``, the serialized -voluptuous schema). Those cross as orjson-encoded JSON in ``bytes`` fields: -measured ~13x faster than the ``google.protobuf.Struct`` fields they replaced, -with native number fidelity (Struct stored every number as a double) and one -coercer — :func:`encode_json` embeds HA's rich-type JSON encoding, so -producers never pre-coerce. +The integration and the sandbox runtime exchange typed protobuf messages over +the :class:`Channel`. Each message type is namespaced ``sandbox/…``; this +module holds the ``MSG_*`` type-string constants, the ``type → (request_cls, +result_cls)`` registry (the codec resolves it on both encode and decode), and +the single encoder/decoder pair for the genuinely dynamic payloads +(service_data, target, state attributes, capabilities, the wrapped Store +envelope, flow ``data``/``errors``/``context``, the serialized voluptuous +schema). Those cross as orjson-encoded JSON in ``bytes`` fields: measured ~13x +faster than the ``google.protobuf.Struct`` fields they replaced, with native +number fidelity (Struct stored every number as a double) and one coercer — +:func:`encode_json` embeds HA's rich-type JSON encoding, so producers never +pre-coerce. Mirrored verbatim across the no-cross-import boundary, exactly like -:mod:`channel` / :mod:`protocol`: the same file lives at -``hass_client.messages``. The relative ``._proto`` import resolves to each -side's own checked-in gencode, so the two copies are byte-identical — and -``sandbox/proto/check_mirror_drift.sh`` fails the build if they drift apart. +:mod:`channel`: the same file lives at ``hass_client.messages``. The relative +``._proto`` import resolves to each side's own checked-in gencode, so the two +copies are byte-identical — and ``sandbox/proto/check_mirror_drift.sh`` fails +the build if they drift apart. + +Each ``MSG_*`` type maps to a request/result proto message pair in +``REGISTRY``, generated from ``sandbox/proto/sandbox.proto``. The payload +shapes described below are the *logical* contract for each call — they are +carried as those typed proto messages, not free-form dicts. A registry-free +line-oriented JSON codec lives in the test helpers as the channel-core +test/debug wire. + +Main → Sandbox calls: + +* ``sandbox/entry_setup`` — push a serialised :class:`ConfigEntry` into + the sandbox, asking it to load the owning integration and run + ``async_setup_entry``. Returns ``{"ok": bool, "reason": str | None}``. + Carries an ``integration_source`` sub-message telling a stateless sandbox + where to fetch the integration code: ``{kind: "builtin"}`` (the bundled + ``homeassistant`` package provides it — a no-op) or ``{kind: "git", url, + ref, tag, domain, subdir}`` for custom (HACS) integrations. ``ref`` is an + exact commit sha (main pins tag→sha; see ``sources.py``); the sandbox + fetches the code before setup (see ``hass_client.sources``). +* ``sandbox/entry_unload`` — ask the sandbox to unload an entry by id. +* ``sandbox/call_service`` — generic service dispatch (shared with + the main→sandbox service mirroring path). Payload mirrors a + ``ServiceCall``: ``(domain, service, target, service_data, context, + return_response)``. Returns either ``None`` or a service-response dict. +* ``sandbox/entity_query`` — generic request/response RPC for the + server-side entity queries with no ``SupportsResponse`` service to ride + (media search, update release notes, vacuum segments, the WS-only calendar + event edits). Payload ``{sandbox_entity_id, method, args, context_id}``; + the sandbox resolves the entity, invokes ``method`` with ``args`` as kwargs, + and returns the serialised result wrapped as ``{"value": }``. + Ops that map to a ``SupportsResponse`` service use ``call_service`` instead. +* ``sandbox/get_translations`` — pull a sandboxed integration's frontend + translation strings. Payload ``{language, domains: [str]}`` (main batches + every owned custom domain of one group into a single request). Response + ``{language, strings: {domain: }}`` — the + un-flattened nesting a ``translations/.json`` holds, with ``title`` + pre-filled from the integration name (main has no ``Integration`` for a + custom domain, so it cannot run that fallback). Built-in domains never + cross the wire — main reads its byte-identical disk copy. +* ``sandbox/ping`` — liveness probe; the runtime echoes an empty result. +* ``sandbox/flow_init`` / ``sandbox/flow_step`` / ``sandbox/flow_abort`` — + config-flow forwarding: bootstrap a sandbox-side flow, drive one step, + tear a flow down. See ``proxy_flow`` (main) / ``flow_runner`` (sandbox). + +Sandbox → Main calls: + +* ``sandbox/register_entity`` — sandbox tells main "I just added an + entity, here's its description". Main builds the proxy and replies + ``{"entity_id": }`` so the sandbox can route later + ``call_service`` requests back to the right local entity. Optional + ``device_info`` field: a JSON-flattened ``DeviceInfo`` dict + — sets become lists of two-element lists (``identifiers`` / + ``connections``), tuples become lists (``via_device``), and + ``entry_type`` is the enum's string value. When present, main calls + :func:`device_registry.async_get_or_create` so the sandbox's devices + surface in main's device_registry tied to the sandboxed entry. +* ``sandbox/unregister_entity`` — symmetric counterpart. +* ``sandbox/state_changed`` — push (no response). Carries the + marshalled state delta for one entity. +* ``sandbox/register_service`` — sandbox tells main "I just + registered a service, please mirror it". Main installs a thin handler + that forwards calls back over the shared ``sandbox/call_service`` + channel. +* ``sandbox/unregister_service`` — symmetric counterpart. +* ``sandbox/fire_event`` — push (no response). The sandbox + forwards each ``_*`` event so main listeners (notably + ``automation``) can react as if the integration ran locally. +* ``sandbox/store_load`` — sandbox-side ``Store.async_load`` + proxies to this RPC. Payload ``{"key": str}``; response is the wrapped + ``{"version", "minor_version", "key", "data"}`` dict the sandbox last + saved, or ``None`` if no data exists yet. The group is implicit from + the channel — each :class:`SandboxBridge` only ever serves one group. +* ``sandbox/store_save`` — sandbox-side ``Store`` flush. + Payload ``{"key": str, "data": dict}``; main writes the wrapped dict + to ``/.storage/sandbox//`` atomically. Response + is ``{"ok": True}``. +* ``sandbox/store_remove`` — sandbox-side + ``Store.async_remove``. Payload ``{"key": str}``; main unlinks the + file (if any). Response is ``{"ok": True}``. + +Main → Sandbox shutdown: + +* ``sandbox/shutdown`` — ask the runtime to unload its entries, dump + ``RestoreEntity`` state, fire ``EVENT_HOMEASSISTANT_FINAL_WRITE`` so any + pending Stores flush to main via the ``current_sandbox`` store bridge, + and exit cleanly. Response ``{"ok": True, "unloaded": int, "restored": + int}``. The runtime sets its shutdown event right after writing the + reply, so the subprocess exits 0 on its own — main only needs SIGTERM + if the round-trip times out. """ -from typing import Any +from typing import Any, Final from google.protobuf.message import Message import orjson @@ -26,36 +118,65 @@ from homeassistant.helpers.json import json_encoder_default from ._proto import sandbox_pb2 as pb +# Handshake (Sandbox → Main): the runtime's first frame on the channel. +# Replaces the old ``sandbox:ready`` stdout text marker — the manager +# registers a handler for this push and treats its arrival as "running", +# so stdout carries nothing but channel frames. +MSG_READY: Final = "sandbox/ready" + +# Main → Sandbox +MSG_ENTRY_SETUP: Final = "sandbox/entry_setup" +MSG_ENTRY_UNLOAD: Final = "sandbox/entry_unload" +MSG_CALL_SERVICE: Final = "sandbox/call_service" +MSG_ENTITY_QUERY: Final = "sandbox/entity_query" +MSG_GET_TRANSLATIONS: Final = "sandbox/get_translations" +MSG_SHUTDOWN: Final = "sandbox/shutdown" +MSG_PING: Final = "sandbox/ping" +MSG_FLOW_INIT: Final = "sandbox/flow_init" +MSG_FLOW_STEP: Final = "sandbox/flow_step" +MSG_FLOW_ABORT: Final = "sandbox/flow_abort" + +# Sandbox → Main +MSG_REGISTER_ENTITY: Final = "sandbox/register_entity" +MSG_UNREGISTER_ENTITY: Final = "sandbox/unregister_entity" +MSG_STATE_CHANGED: Final = "sandbox/state_changed" +MSG_REGISTER_SERVICE: Final = "sandbox/register_service" +MSG_UNREGISTER_SERVICE: Final = "sandbox/unregister_service" +MSG_FIRE_EVENT: Final = "sandbox/fire_event" +MSG_STORE_LOAD: Final = "sandbox/store_load" +MSG_STORE_SAVE: Final = "sandbox/store_save" +MSG_STORE_REMOVE: Final = "sandbox/store_remove" + # Wire type → (request message class, result message class). The result class # is ``None`` for one-way pushes (ready / state_changed / fire_event). The # codec resolves these from ``frame.type`` on both encode and decode. REGISTRY: dict[str, tuple[type[Message], type[Message] | None]] = { # handshake (push) - "sandbox/ready": (pb.Ready, None), + MSG_READY: (pb.Ready, None), # main → sandbox - "sandbox/entry_setup": (pb.EntrySetup, pb.EntrySetupResult), - "sandbox/entry_unload": (pb.EntryUnload, pb.EntryUnloadResult), - "sandbox/call_service": (pb.CallService, pb.CallServiceResult), - "sandbox/entity_query": (pb.EntityQuery, pb.EntityQueryResult), - "sandbox/get_translations": (pb.GetTranslations, pb.GetTranslationsResult), - "sandbox/shutdown": (pb.Shutdown, pb.ShutdownResult), - "sandbox/ping": (pb.Ping, pb.PingResult), - "sandbox/flow_init": (pb.FlowInit, pb.FlowResult), - "sandbox/flow_step": (pb.FlowStep, pb.FlowResult), - "sandbox/flow_abort": (pb.FlowAbort, pb.FlowAbortResult), + MSG_ENTRY_SETUP: (pb.EntrySetup, pb.EntrySetupResult), + MSG_ENTRY_UNLOAD: (pb.EntryUnload, pb.EntryUnloadResult), + MSG_CALL_SERVICE: (pb.CallService, pb.CallServiceResult), + MSG_ENTITY_QUERY: (pb.EntityQuery, pb.EntityQueryResult), + MSG_GET_TRANSLATIONS: (pb.GetTranslations, pb.GetTranslationsResult), + MSG_SHUTDOWN: (pb.Shutdown, pb.ShutdownResult), + MSG_PING: (pb.Ping, pb.PingResult), + MSG_FLOW_INIT: (pb.FlowInit, pb.FlowResult), + MSG_FLOW_STEP: (pb.FlowStep, pb.FlowResult), + MSG_FLOW_ABORT: (pb.FlowAbort, pb.FlowAbortResult), # sandbox → main - "sandbox/register_entity": (pb.EntityDescription, pb.RegisterEntityResult), - "sandbox/unregister_entity": (pb.UnregisterEntity, pb.UnregisterEntityResult), - "sandbox/state_changed": (pb.StateChanged, None), - "sandbox/register_service": (pb.RegisterService, pb.RegisterServiceResult), - "sandbox/unregister_service": ( + MSG_REGISTER_ENTITY: (pb.EntityDescription, pb.RegisterEntityResult), + MSG_UNREGISTER_ENTITY: (pb.UnregisterEntity, pb.UnregisterEntityResult), + MSG_STATE_CHANGED: (pb.StateChanged, None), + MSG_REGISTER_SERVICE: (pb.RegisterService, pb.RegisterServiceResult), + MSG_UNREGISTER_SERVICE: ( pb.UnregisterService, pb.UnregisterServiceResult, ), - "sandbox/fire_event": (pb.FireEvent, None), - "sandbox/store_load": (pb.StoreLoad, pb.StoreLoadResult), - "sandbox/store_save": (pb.StoreSave, pb.StoreSaveResult), - "sandbox/store_remove": (pb.StoreRemove, pb.StoreRemoveResult), + MSG_FIRE_EVENT: (pb.FireEvent, None), + MSG_STORE_LOAD: (pb.StoreLoad, pb.StoreLoadResult), + MSG_STORE_SAVE: (pb.StoreSave, pb.StoreSaveResult), + MSG_STORE_REMOVE: (pb.StoreRemove, pb.StoreRemoveResult), } @@ -203,6 +324,26 @@ def make_entity_description( __all__ = [ + "MSG_CALL_SERVICE", + "MSG_ENTITY_QUERY", + "MSG_ENTRY_SETUP", + "MSG_ENTRY_UNLOAD", + "MSG_FIRE_EVENT", + "MSG_FLOW_ABORT", + "MSG_FLOW_INIT", + "MSG_FLOW_STEP", + "MSG_GET_TRANSLATIONS", + "MSG_PING", + "MSG_READY", + "MSG_REGISTER_ENTITY", + "MSG_REGISTER_SERVICE", + "MSG_SHUTDOWN", + "MSG_STATE_CHANGED", + "MSG_STORE_LOAD", + "MSG_STORE_REMOVE", + "MSG_STORE_SAVE", + "MSG_UNREGISTER_ENTITY", + "MSG_UNREGISTER_SERVICE", "REGISTRY", "decode_json", "decode_json_dict", diff --git a/homeassistant/components/sandbox/protocol.py b/homeassistant/components/sandbox/protocol.py deleted file mode 100644 index 1fed5c3ab7ff..000000000000 --- a/homeassistant/components/sandbox/protocol.py +++ /dev/null @@ -1,143 +0,0 @@ -"""Wire-protocol message-type constants. - -The integration and the sandbox runtime exchange typed protobuf messages -over the :class:`Channel`. Each message type is namespaced ``sandbox/…``; -this module holds the type-string constants. Both sides share the same -names — kept here on the HA side and mirrored verbatim in -:mod:`hass_client.protocol` so neither has to import the other. - -The wire is protobuf (codec :class:`~.codec_protobuf.ProtobufCodec`, which a -:class:`~.channel.Channel` now requires explicitly): each ``type`` maps to a -request/result proto message pair in :mod:`.messages` (the `REGISTRY`), -generated from ``sandbox/proto/sandbox.proto``. The payload shapes described -below are the *logical* contract for each call — they are carried as those -typed proto messages, not free-form dicts (only genuinely dynamic fields, e.g. -``service_data`` / state attributes / serialized voluptuous schemas, cross -as orjson-encoded JSON ``bytes``). A registry-free line-oriented JSON codec lives -in the test helpers as the channel-core test/debug wire. - -Main → Sandbox calls: - -* ``sandbox/entry_setup`` — push a serialised :class:`ConfigEntry` into - the sandbox, asking it to load the owning integration and run - ``async_setup_entry``. Returns ``{"ok": bool, "reason": str | None}``. - Carries an ``integration_source`` sub-message telling a stateless sandbox - where to fetch the integration code: ``{kind: "builtin"}`` (the bundled - ``homeassistant`` package provides it — a no-op) or ``{kind: "git", url, - ref, tag, domain, subdir}`` for custom (HACS) integrations. ``ref`` is an - exact commit sha (main pins tag→sha; see ``sources.py``); the sandbox - fetches the code before setup (see ``hass_client.sources``). -* ``sandbox/entry_unload`` — ask the sandbox to unload an entry by id. -* ``sandbox/call_service`` — generic service dispatch (shared with - the main→sandbox service mirroring path). Payload mirrors a - ``ServiceCall``: ``(domain, service, target, service_data, context, - return_response)``. Returns either ``None`` or a service-response dict. -* ``sandbox/entity_query`` — generic request/response RPC for the - server-side entity queries with no ``SupportsResponse`` service to ride - (media search, update release notes, vacuum segments, the WS-only calendar - event edits). Payload ``{sandbox_entity_id, method, args, context_id}``; - the sandbox resolves the entity, invokes ``method`` with ``args`` as kwargs, - and returns the serialised result wrapped as ``{"value": }``. - Ops that map to a ``SupportsResponse`` service use ``call_service`` instead. -* ``sandbox/get_translations`` — pull a sandboxed integration's frontend - translation strings. Payload ``{language, domains: [str]}`` (main batches - every owned custom domain of one group into a single request). Response - ``{language, strings: {domain: }}`` — the - un-flattened nesting a ``translations/.json`` holds, with ``title`` - pre-filled from the integration name (main has no ``Integration`` for a - custom domain, so it cannot run that fallback). Built-in domains never - cross the wire — main reads its byte-identical disk copy. - -Sandbox → Main calls: - -* ``sandbox/register_entity`` — sandbox tells main "I just added an - entity, here's its description". Main builds the proxy and replies - ``{"entity_id": }`` so the sandbox can route later - ``call_service`` requests back to the right local entity. Optional - ``device_info`` field: a JSON-flattened ``DeviceInfo`` dict - — sets become lists of two-element lists (``identifiers`` / - ``connections``), tuples become lists (``via_device``), and - ``entry_type`` is the enum's string value. When present, main calls - :func:`device_registry.async_get_or_create` so the sandbox's devices - surface in main's device_registry tied to the sandboxed entry. -* ``sandbox/unregister_entity`` — symmetric counterpart. -* ``sandbox/state_changed`` — push (no response). Carries the - marshalled state delta for one entity. -* ``sandbox/register_service`` — sandbox tells main "I just - registered a service, please mirror it". Main installs a thin handler - that forwards calls back over the shared ``sandbox/call_service`` - channel. -* ``sandbox/unregister_service`` — symmetric counterpart. -* ``sandbox/fire_event`` — push (no response). The sandbox - forwards each ``_*`` event so main listeners (notably - ``automation``) can react as if the integration ran locally. -* ``sandbox/store_load`` — sandbox-side ``Store.async_load`` - proxies to this RPC. Payload ``{"key": str}``; response is the wrapped - ``{"version", "minor_version", "key", "data"}`` dict the sandbox last - saved, or ``None`` if no data exists yet. The group is implicit from - the channel — each :class:`SandboxBridge` only ever serves one group. -* ``sandbox/store_save`` — sandbox-side ``Store`` flush. - Payload ``{"key": str, "data": dict}``; main writes the wrapped dict - to ``/.storage/sandbox//`` atomically. Response - is ``{"ok": True}``. -* ``sandbox/store_remove`` — sandbox-side - ``Store.async_remove``. Payload ``{"key": str}``; main unlinks the - file (if any). Response is ``{"ok": True}``. - -Main → Sandbox shutdown: - -* ``sandbox/shutdown`` — ask the runtime to unload its entries, dump - ``RestoreEntity`` state, fire ``EVENT_HOMEASSISTANT_FINAL_WRITE`` so any - pending Stores flush to main via the ``current_sandbox`` store bridge, - and exit cleanly. Response ``{"ok": True, "unloaded": int, "restored": - int}``. The runtime sets its shutdown event right after writing the - reply, so the subprocess exits 0 on its own — main only needs SIGTERM - if the round-trip times out. -""" - -from typing import Final - -# Handshake (Sandbox → Main): the runtime's first frame on the channel. -# Replaces the old ``sandbox:ready`` stdout text marker — the manager -# registers a handler for this push and treats its arrival as "running", -# so stdout carries nothing but channel frames. -MSG_READY: Final = "sandbox/ready" - -# Main → Sandbox -MSG_ENTRY_SETUP: Final = "sandbox/entry_setup" -MSG_ENTRY_UNLOAD: Final = "sandbox/entry_unload" -MSG_CALL_SERVICE: Final = "sandbox/call_service" -MSG_ENTITY_QUERY: Final = "sandbox/entity_query" -MSG_GET_TRANSLATIONS: Final = "sandbox/get_translations" -MSG_SHUTDOWN: Final = "sandbox/shutdown" - -# Sandbox → Main -MSG_REGISTER_ENTITY: Final = "sandbox/register_entity" -MSG_UNREGISTER_ENTITY: Final = "sandbox/unregister_entity" -MSG_STATE_CHANGED: Final = "sandbox/state_changed" -MSG_REGISTER_SERVICE: Final = "sandbox/register_service" -MSG_UNREGISTER_SERVICE: Final = "sandbox/unregister_service" -MSG_FIRE_EVENT: Final = "sandbox/fire_event" -MSG_STORE_LOAD: Final = "sandbox/store_load" -MSG_STORE_SAVE: Final = "sandbox/store_save" -MSG_STORE_REMOVE: Final = "sandbox/store_remove" - - -__all__ = [ - "MSG_CALL_SERVICE", - "MSG_ENTITY_QUERY", - "MSG_ENTRY_SETUP", - "MSG_ENTRY_UNLOAD", - "MSG_FIRE_EVENT", - "MSG_GET_TRANSLATIONS", - "MSG_READY", - "MSG_REGISTER_ENTITY", - "MSG_REGISTER_SERVICE", - "MSG_SHUTDOWN", - "MSG_STATE_CHANGED", - "MSG_STORE_LOAD", - "MSG_STORE_REMOVE", - "MSG_STORE_SAVE", - "MSG_UNREGISTER_ENTITY", - "MSG_UNREGISTER_SERVICE", -] diff --git a/homeassistant/components/sandbox/proxy_flow.py b/homeassistant/components/sandbox/proxy_flow.py index 78b14a06d31d..237c4efc3960 100644 --- a/homeassistant/components/sandbox/proxy_flow.py +++ b/homeassistant/components/sandbox/proxy_flow.py @@ -35,7 +35,14 @@ from homeassistant.data_entry_flow import FlowResultType from ._proto import sandbox_pb2 as pb from .channel import ChannelClosedError, ChannelRemoteError -from .messages import decode_json, decode_json_dict, encode_json +from .messages import ( + MSG_FLOW_ABORT, + MSG_FLOW_INIT, + MSG_FLOW_STEP, + decode_json, + decode_json_dict, + encode_json, +) from .schema_bridge import reconstruct_schema if TYPE_CHECKING: @@ -167,7 +174,7 @@ class SandboxFlowProxy(ConfigFlow): ) if user_input is not None: request.data = encode_json(_to_jsonable(user_input)) - result = await channel.call("sandbox/flow_init", request) + result = await channel.call(MSG_FLOW_INIT, request) self._sandbox_flow_id = ( result.flow_id if result.HasField("flow_id") else None ) @@ -182,7 +189,7 @@ class SandboxFlowProxy(ConfigFlow): step.user_input = encode_json({"next_step_id": step_id}) elif user_input is not None: step.user_input = encode_json(user_input) - result = await channel.call("sandbox/flow_step", step) + result = await channel.call(MSG_FLOW_STEP, step) except ChannelClosedError: self._terminated = True _LOGGER.warning( @@ -395,7 +402,7 @@ def _reconstruct_menu_options(items: list[Any]) -> list[str] | dict[str, str]: async def _safe_abort(channel: Any, flow_id: str, group: str, handler: str) -> None: """Fire ``flow_abort`` on the sandbox and swallow errors.""" try: - await channel.call("sandbox/flow_abort", pb.FlowAbort(flow_id=flow_id)) + await channel.call(MSG_FLOW_ABORT, pb.FlowAbort(flow_id=flow_id)) except (ChannelClosedError, ChannelRemoteError) as err: _LOGGER.debug("Sandbox %r flow_abort for %s failed: %s", group, handler, err) diff --git a/homeassistant/components/sandbox/router.py b/homeassistant/components/sandbox/router.py index c6e34592a4ad..a20d6bafe084 100644 --- a/homeassistant/components/sandbox/router.py +++ b/homeassistant/components/sandbox/router.py @@ -29,8 +29,7 @@ from ._proto import sandbox_pb2 as pb from .channel import ChannelClosedError, ChannelRemoteError from .classifier import SandboxAssignment, classify from .manager import SandboxManager -from .messages import encode_json -from .protocol import MSG_ENTRY_SETUP, MSG_ENTRY_UNLOAD +from .messages import MSG_ENTRY_SETUP, MSG_ENTRY_UNLOAD, encode_json from .proxy_flow import SandboxFlowProxy from .sources import SandboxSourceError, async_resolve_integration_source diff --git a/homeassistant/components/sandbox/translation.py b/homeassistant/components/sandbox/translation.py index 5c00ee1d4459..9b5e0274773c 100644 --- a/homeassistant/components/sandbox/translation.py +++ b/homeassistant/components/sandbox/translation.py @@ -33,8 +33,7 @@ from homeassistant.loader import IntegrationNotFound, async_get_integration from ._proto import sandbox_pb2 as pb from .channel import Channel, ChannelClosedError, ChannelRemoteError -from .messages import decode_json_dict -from .protocol import MSG_GET_TRANSLATIONS +from .messages import MSG_GET_TRANSLATIONS, decode_json_dict from .proxy_flow import SandboxFlowProxy if TYPE_CHECKING: diff --git a/homeassistant/config_entries.py b/homeassistant/config_entries.py index 98324a414397..75e0892f086e 100644 --- a/homeassistant/config_entries.py +++ b/homeassistant/config_entries.py @@ -277,7 +277,7 @@ STATE_KEYS = { "error_reason_translation_key", "error_reason_translation_placeholders", } -FROZEN_CONFIG_ENTRY_ATTRS = {"entry_id", "domain", *STATE_KEYS} +FROZEN_CONFIG_ENTRY_ATTRS = {"entry_id", "domain", "sandbox", *STATE_KEYS} UPDATE_ENTRY_CONFIG_ENTRY_ATTRS = { "unique_id", "title", @@ -287,7 +287,6 @@ UPDATE_ENTRY_CONFIG_ENTRY_ATTRS = { "pref_disable_polling", "minor_version", "version", - "sandbox", } @@ -2662,7 +2661,6 @@ class ConfigEntries: options: Mapping[str, Any] | UndefinedType = UNDEFINED, pref_disable_new_entities: bool | UndefinedType = UNDEFINED, pref_disable_polling: bool | UndefinedType = UNDEFINED, - sandbox: str | None | UndefinedType = UNDEFINED, title: str | UndefinedType = UNDEFINED, unique_id: str | None | UndefinedType = UNDEFINED, version: int | UndefinedType = UNDEFINED, @@ -2683,7 +2681,6 @@ class ConfigEntries: options=options, pref_disable_new_entities=pref_disable_new_entities, pref_disable_polling=pref_disable_polling, - sandbox=sandbox, title=title, unique_id=unique_id, version=version, @@ -2702,7 +2699,6 @@ class ConfigEntries: options: Mapping[str, Any] | UndefinedType = UNDEFINED, pref_disable_new_entities: bool | UndefinedType = UNDEFINED, pref_disable_polling: bool | UndefinedType = UNDEFINED, - sandbox: str | None | UndefinedType = UNDEFINED, subentries: dict[str, ConfigSubentry] | UndefinedType = UNDEFINED, title: str | UndefinedType = UNDEFINED, unique_id: str | None | UndefinedType = UNDEFINED, @@ -2753,7 +2749,6 @@ class ConfigEntries: ("minor_version", minor_version), ("pref_disable_new_entities", pref_disable_new_entities), ("pref_disable_polling", pref_disable_polling), - ("sandbox", sandbox), ("title", title), ("version", version), ): diff --git a/sandbox/ARCHITECTURE.md b/sandbox/ARCHITECTURE.md index 374cf427e093..f979bdba556a 100644 --- a/sandbox/ARCHITECTURE.md +++ b/sandbox/ARCHITECTURE.md @@ -461,10 +461,11 @@ actually built, deferred, and flagged forward. For a quick map: | Shutdown | `__init__.py` (`_on_stop`), `manager.py` | `sandbox/__init__.py` (`_run_graceful_shutdown`) | | Test infra | — | `testing/`, `run_compat.py` | -The wire-protocol constants live in two files that mirror each other verbatim: -`homeassistant/components/sandbox/protocol.py` and -`hass_client/hass_client/protocol.py` (along with the mirrored `channel.py` / -`codec_protobuf.py` / `messages.py`). +The wire-protocol `MSG_*` constants live in `messages.py` (alongside the proto +registry and the dynamic-payload JSON codec), one of the files mirrored +verbatim between `homeassistant/components/sandbox/` and +`hass_client/hass_client/` (with `channel.py` / `codec_protobuf.py` and the +checked-in `_proto/sandbox_pb2.py`/`.pyi` gencode). --- diff --git a/sandbox/hass_client/hass_client/channel.py b/sandbox/hass_client/hass_client/channel.py index 9f3a9488b4c3..2428a2fb7939 100644 --- a/sandbox/hass_client/hass_client/channel.py +++ b/sandbox/hass_client/hass_client/channel.py @@ -16,7 +16,7 @@ dispatch core: :class:`StreamTransport` length-prefixes each frame (4-byte big-endian length + body) over an :class:`asyncio.StreamReader` / :class:`asyncio.StreamWriter` pair (stdio, unix socket). A future - ``WebSocketTransport`` drops in via :meth:`Channel.from_transport` using + ``WebSocketTransport`` drops in via ``Channel(transport=...)`` using aiohttp's native binary framing. The :class:`Frame` shape mirrors the three message kinds that cross the @@ -309,8 +309,7 @@ class Channel: The common case passes a ``reader``/``writer`` pair, framed with :class:`StreamTransport` (length-prefixed). To run over a non-stream - transport (e.g. websockets), pass ``transport=`` instead — see - :meth:`from_transport`. + transport (e.g. websockets), pass ``transport=`` instead. ``codec`` is required — production passes :class:`~.codec_protobuf.ProtobufCodec`; a forgotten codec is a @@ -340,24 +339,6 @@ class Channel: self._inflight_sem = asyncio.Semaphore(max_inflight) self._max_queued = max_queued - @classmethod - def from_transport( - cls, - transport: Transport, - *, - codec: Codec, - name: str = "channel", - max_inflight: int = DEFAULT_MAX_INFLIGHT, - ) -> Channel: - """Build a channel over an arbitrary :class:`Transport`. - - This is the seam a future ``WebSocketTransport`` drops into — the - dispatch core is identical regardless of how frames reach the wire. - """ - return cls( - transport=transport, codec=codec, name=name, max_inflight=max_inflight - ) - @property def closed(self) -> bool: """Return True once the channel has been closed.""" @@ -400,7 +381,8 @@ class Channel: if self._closed: raise ChannelClosedError(f"channel {self._name!r} is closed") call_id = self._next_id - self._next_id += 1 + # Wrap within the uint32 wire field, skipping 0 (id 0 marks a push). + self._next_id = self._next_id % 0xFFFFFFFF + 1 future: asyncio.Future[Any] = asyncio.get_running_loop().create_future() self._pending[call_id] = future try: @@ -463,8 +445,7 @@ class Channel: ``timeout`` — any handler still running afterwards is left for ``close()`` to cancel. Does not itself cancel anything. """ - inflight = [task for task in self._inflight if task is not self._reader_task] - if inflight: + if inflight := list(self._inflight): await asyncio.wait(inflight, timeout=timeout) async def _write(self, frame: Frame) -> None: diff --git a/sandbox/hass_client/hass_client/codec_protobuf.py b/sandbox/hass_client/hass_client/codec_protobuf.py index eea043e09e76..80b77c518f46 100644 --- a/sandbox/hass_client/hass_client/codec_protobuf.py +++ b/sandbox/hass_client/hass_client/codec_protobuf.py @@ -74,9 +74,13 @@ class ProtobufCodec: def _serialize_body(body: Any, cls: type[Message] | None) -> bytes: - """Serialise a proto-message body; ``None`` becomes an empty message.""" + """Serialise a proto-message body; ``None`` becomes an empty message. + + An empty proto message serialises to zero bytes, so ``None`` maps to + ``b""`` whether or not the type has a registered class. + """ if body is None: - return cls().SerializeToString() if cls is not None else b"" + return b"" if isinstance(body, Message): return body.SerializeToString() raise TypeError( diff --git a/sandbox/hass_client/hass_client/entity_bridge.py b/sandbox/hass_client/hass_client/entity_bridge.py index 759733ed4f84..d8a8ebca51aa 100644 --- a/sandbox/hass_client/hass_client/entity_bridge.py +++ b/sandbox/hass_client/hass_client/entity_bridge.py @@ -38,8 +38,13 @@ from ._json import json_safe from ._proto import sandbox_pb2 as pb from .approved_domains import ApprovedDomains from .channel import Channel -from .messages import encode_json, make_entity_description -from .protocol import MSG_REGISTER_ENTITY, MSG_STATE_CHANGED, MSG_UNREGISTER_ENTITY +from .messages import ( + MSG_REGISTER_ENTITY, + MSG_STATE_CHANGED, + MSG_UNREGISTER_ENTITY, + encode_json, + make_entity_description, +) _LOGGER = logging.getLogger(__name__) diff --git a/sandbox/hass_client/hass_client/entry_runner.py b/sandbox/hass_client/hass_client/entry_runner.py index b0379357c8cf..b9d943572e37 100644 --- a/sandbox/hass_client/hass_client/entry_runner.py +++ b/sandbox/hass_client/hass_client/entry_runner.py @@ -1,7 +1,7 @@ """Sandbox-side entry runner — loads integrations + drives ``async_setup_entry``. The manager pushes a serialised :class:`ConfigEntry` via -``sandbox/entry_setup`` (see :mod:`hass_client.protocol`). The runner +``sandbox/entry_setup`` (see :mod:`hass_client.messages`). The runner rebuilds the entry on the sandbox's private :class:`HomeAssistant`, calls ``hass.config_entries.async_setup`` to load the owning integration, and reports back. Main holds the canonical entry; the sandbox copy is @@ -20,12 +20,13 @@ from homeassistant.helpers.entity_component import DATA_INSTANCES from ._proto import sandbox_pb2 as pb from .approved_domains import ApprovedDomains from .channel import Channel -from .messages import decode_json_dict, encode_json -from .protocol import ( +from .messages import ( MSG_CALL_SERVICE, MSG_ENTITY_QUERY, MSG_ENTRY_SETUP, MSG_ENTRY_UNLOAD, + decode_json_dict, + encode_json, ) from .sources import FetchPrimitive, SandboxSourceError, async_ensure_integration_source diff --git a/sandbox/hass_client/hass_client/event_mirror.py b/sandbox/hass_client/hass_client/event_mirror.py index 5e235002003d..067c052866ee 100644 --- a/sandbox/hass_client/hass_client/event_mirror.py +++ b/sandbox/hass_client/hass_client/event_mirror.py @@ -5,11 +5,13 @@ up to main via ``sandbox/fire_event``. Canonical examples: ``zha_event``, ``mqtt_message_received``, ``hue_event``, ``device_tracker_see``. The bus listener is installed via ``MATCH_ALL`` so we don't need to know -the integration's event names ahead of time, with a callback-decorated -event filter so the bus can short-circuit on a fast path before queuing -the listener. Untrusted (non-approved) event types are silently dropped -— they would never have been forwarded anyway and don't deserve a log -line per event. +the integration's event names ahead of time; the handler itself does the +cheap event-type checks and returns early for foreign events. (The bus's +``event_filter`` fast path can't host that check: a filter receives only +``event.data`` — never the event type — and the bus skips filtered +listeners entirely for events fired without data.) Untrusted +(non-approved) event types are silently dropped — they would never have +been forwarded anyway and don't deserve a log line per event. System events that already cross the bridge through dedicated channels (``EVENT_STATE_CHANGED``, ``EVENT_SERVICE_REGISTERED``, …) are @@ -43,8 +45,7 @@ from homeassistant.core import Event, HomeAssistant, callback from ._proto import sandbox_pb2 as pb from .approved_domains import ApprovedDomains from .channel import Channel -from .messages import encode_json -from .protocol import MSG_FIRE_EVENT +from .messages import MSG_FIRE_EVENT, encode_json _LOGGER = logging.getLogger(__name__) diff --git a/sandbox/hass_client/hass_client/flow_runner.py b/sandbox/hass_client/hass_client/flow_runner.py index 3c4fe41e692f..3ae9dd435791 100644 --- a/sandbox/hass_client/hass_client/flow_runner.py +++ b/sandbox/hass_client/hass_client/flow_runner.py @@ -53,7 +53,13 @@ from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo from ._proto import sandbox_pb2 as pb from .channel import Channel -from .messages import decode_json_dict, encode_json +from .messages import ( + MSG_FLOW_ABORT, + MSG_FLOW_INIT, + MSG_FLOW_STEP, + decode_json_dict, + encode_json, +) from .schema_bridge import serialize_schema _LOGGER = logging.getLogger(__name__) @@ -147,9 +153,9 @@ class FlowRunner: def register(self, channel: Channel) -> None: """Register the ``sandbox/flow_*`` handlers on ``channel``.""" - channel.register("sandbox/flow_init", self._handle_flow_init) - channel.register("sandbox/flow_step", self._handle_flow_step) - channel.register("sandbox/flow_abort", self._handle_flow_abort) + channel.register(MSG_FLOW_INIT, self._handle_flow_init) + channel.register(MSG_FLOW_STEP, self._handle_flow_step) + channel.register(MSG_FLOW_ABORT, self._handle_flow_abort) async def async_stop(self) -> None: """Tear down in-progress flows and stop the private hass. diff --git a/sandbox/hass_client/hass_client/messages.py b/sandbox/hass_client/hass_client/messages.py index 1fd4a9a7e6cf..066ab650604a 100644 --- a/sandbox/hass_client/hass_client/messages.py +++ b/sandbox/hass_client/hass_client/messages.py @@ -1,23 +1,115 @@ -"""Typed protobuf message registry + dynamic-payload JSON codec. +"""Wire-protocol constants, typed proto registry + dynamic-payload JSON codec. -This module is the codec's view of the wire: the ``type → (request_cls, -result_cls)`` registry plus the single encoder/decoder pair for the genuinely -dynamic payloads (service_data, target, state attributes, capabilities, the -wrapped Store envelope, flow ``data``/``errors``/``context``, the serialized -voluptuous schema). Those cross as orjson-encoded JSON in ``bytes`` fields: -measured ~13x faster than the ``google.protobuf.Struct`` fields they replaced, -with native number fidelity (Struct stored every number as a double) and one -coercer — :func:`encode_json` embeds HA's rich-type JSON encoding, so -producers never pre-coerce. +The integration and the sandbox runtime exchange typed protobuf messages over +the :class:`Channel`. Each message type is namespaced ``sandbox/…``; this +module holds the ``MSG_*`` type-string constants, the ``type → (request_cls, +result_cls)`` registry (the codec resolves it on both encode and decode), and +the single encoder/decoder pair for the genuinely dynamic payloads +(service_data, target, state attributes, capabilities, the wrapped Store +envelope, flow ``data``/``errors``/``context``, the serialized voluptuous +schema). Those cross as orjson-encoded JSON in ``bytes`` fields: measured ~13x +faster than the ``google.protobuf.Struct`` fields they replaced, with native +number fidelity (Struct stored every number as a double) and one coercer — +:func:`encode_json` embeds HA's rich-type JSON encoding, so producers never +pre-coerce. Mirrored verbatim across the no-cross-import boundary, exactly like -:mod:`channel` / :mod:`protocol`: the same file lives at -``hass_client.messages``. The relative ``._proto`` import resolves to each -side's own checked-in gencode, so the two copies are byte-identical — and -``sandbox/proto/check_mirror_drift.sh`` fails the build if they drift apart. +:mod:`channel`: the same file lives at ``hass_client.messages``. The relative +``._proto`` import resolves to each side's own checked-in gencode, so the two +copies are byte-identical — and ``sandbox/proto/check_mirror_drift.sh`` fails +the build if they drift apart. + +Each ``MSG_*`` type maps to a request/result proto message pair in +``REGISTRY``, generated from ``sandbox/proto/sandbox.proto``. The payload +shapes described below are the *logical* contract for each call — they are +carried as those typed proto messages, not free-form dicts. A registry-free +line-oriented JSON codec lives in the test helpers as the channel-core +test/debug wire. + +Main → Sandbox calls: + +* ``sandbox/entry_setup`` — push a serialised :class:`ConfigEntry` into + the sandbox, asking it to load the owning integration and run + ``async_setup_entry``. Returns ``{"ok": bool, "reason": str | None}``. + Carries an ``integration_source`` sub-message telling a stateless sandbox + where to fetch the integration code: ``{kind: "builtin"}`` (the bundled + ``homeassistant`` package provides it — a no-op) or ``{kind: "git", url, + ref, tag, domain, subdir}`` for custom (HACS) integrations. ``ref`` is an + exact commit sha (main pins tag→sha; see ``sources.py``); the sandbox + fetches the code before setup (see ``hass_client.sources``). +* ``sandbox/entry_unload`` — ask the sandbox to unload an entry by id. +* ``sandbox/call_service`` — generic service dispatch (shared with + the main→sandbox service mirroring path). Payload mirrors a + ``ServiceCall``: ``(domain, service, target, service_data, context, + return_response)``. Returns either ``None`` or a service-response dict. +* ``sandbox/entity_query`` — generic request/response RPC for the + server-side entity queries with no ``SupportsResponse`` service to ride + (media search, update release notes, vacuum segments, the WS-only calendar + event edits). Payload ``{sandbox_entity_id, method, args, context_id}``; + the sandbox resolves the entity, invokes ``method`` with ``args`` as kwargs, + and returns the serialised result wrapped as ``{"value": }``. + Ops that map to a ``SupportsResponse`` service use ``call_service`` instead. +* ``sandbox/get_translations`` — pull a sandboxed integration's frontend + translation strings. Payload ``{language, domains: [str]}`` (main batches + every owned custom domain of one group into a single request). Response + ``{language, strings: {domain: }}`` — the + un-flattened nesting a ``translations/.json`` holds, with ``title`` + pre-filled from the integration name (main has no ``Integration`` for a + custom domain, so it cannot run that fallback). Built-in domains never + cross the wire — main reads its byte-identical disk copy. +* ``sandbox/ping`` — liveness probe; the runtime echoes an empty result. +* ``sandbox/flow_init`` / ``sandbox/flow_step`` / ``sandbox/flow_abort`` — + config-flow forwarding: bootstrap a sandbox-side flow, drive one step, + tear a flow down. See ``proxy_flow`` (main) / ``flow_runner`` (sandbox). + +Sandbox → Main calls: + +* ``sandbox/register_entity`` — sandbox tells main "I just added an + entity, here's its description". Main builds the proxy and replies + ``{"entity_id": }`` so the sandbox can route later + ``call_service`` requests back to the right local entity. Optional + ``device_info`` field: a JSON-flattened ``DeviceInfo`` dict + — sets become lists of two-element lists (``identifiers`` / + ``connections``), tuples become lists (``via_device``), and + ``entry_type`` is the enum's string value. When present, main calls + :func:`device_registry.async_get_or_create` so the sandbox's devices + surface in main's device_registry tied to the sandboxed entry. +* ``sandbox/unregister_entity`` — symmetric counterpart. +* ``sandbox/state_changed`` — push (no response). Carries the + marshalled state delta for one entity. +* ``sandbox/register_service`` — sandbox tells main "I just + registered a service, please mirror it". Main installs a thin handler + that forwards calls back over the shared ``sandbox/call_service`` + channel. +* ``sandbox/unregister_service`` — symmetric counterpart. +* ``sandbox/fire_event`` — push (no response). The sandbox + forwards each ``_*`` event so main listeners (notably + ``automation``) can react as if the integration ran locally. +* ``sandbox/store_load`` — sandbox-side ``Store.async_load`` + proxies to this RPC. Payload ``{"key": str}``; response is the wrapped + ``{"version", "minor_version", "key", "data"}`` dict the sandbox last + saved, or ``None`` if no data exists yet. The group is implicit from + the channel — each :class:`SandboxBridge` only ever serves one group. +* ``sandbox/store_save`` — sandbox-side ``Store`` flush. + Payload ``{"key": str, "data": dict}``; main writes the wrapped dict + to ``/.storage/sandbox//`` atomically. Response + is ``{"ok": True}``. +* ``sandbox/store_remove`` — sandbox-side + ``Store.async_remove``. Payload ``{"key": str}``; main unlinks the + file (if any). Response is ``{"ok": True}``. + +Main → Sandbox shutdown: + +* ``sandbox/shutdown`` — ask the runtime to unload its entries, dump + ``RestoreEntity`` state, fire ``EVENT_HOMEASSISTANT_FINAL_WRITE`` so any + pending Stores flush to main via the ``current_sandbox`` store bridge, + and exit cleanly. Response ``{"ok": True, "unloaded": int, "restored": + int}``. The runtime sets its shutdown event right after writing the + reply, so the subprocess exits 0 on its own — main only needs SIGTERM + if the round-trip times out. """ -from typing import Any +from typing import Any, Final from google.protobuf.message import Message import orjson @@ -26,36 +118,65 @@ from homeassistant.helpers.json import json_encoder_default from ._proto import sandbox_pb2 as pb +# Handshake (Sandbox → Main): the runtime's first frame on the channel. +# Replaces the old ``sandbox:ready`` stdout text marker — the manager +# registers a handler for this push and treats its arrival as "running", +# so stdout carries nothing but channel frames. +MSG_READY: Final = "sandbox/ready" + +# Main → Sandbox +MSG_ENTRY_SETUP: Final = "sandbox/entry_setup" +MSG_ENTRY_UNLOAD: Final = "sandbox/entry_unload" +MSG_CALL_SERVICE: Final = "sandbox/call_service" +MSG_ENTITY_QUERY: Final = "sandbox/entity_query" +MSG_GET_TRANSLATIONS: Final = "sandbox/get_translations" +MSG_SHUTDOWN: Final = "sandbox/shutdown" +MSG_PING: Final = "sandbox/ping" +MSG_FLOW_INIT: Final = "sandbox/flow_init" +MSG_FLOW_STEP: Final = "sandbox/flow_step" +MSG_FLOW_ABORT: Final = "sandbox/flow_abort" + +# Sandbox → Main +MSG_REGISTER_ENTITY: Final = "sandbox/register_entity" +MSG_UNREGISTER_ENTITY: Final = "sandbox/unregister_entity" +MSG_STATE_CHANGED: Final = "sandbox/state_changed" +MSG_REGISTER_SERVICE: Final = "sandbox/register_service" +MSG_UNREGISTER_SERVICE: Final = "sandbox/unregister_service" +MSG_FIRE_EVENT: Final = "sandbox/fire_event" +MSG_STORE_LOAD: Final = "sandbox/store_load" +MSG_STORE_SAVE: Final = "sandbox/store_save" +MSG_STORE_REMOVE: Final = "sandbox/store_remove" + # Wire type → (request message class, result message class). The result class # is ``None`` for one-way pushes (ready / state_changed / fire_event). The # codec resolves these from ``frame.type`` on both encode and decode. REGISTRY: dict[str, tuple[type[Message], type[Message] | None]] = { # handshake (push) - "sandbox/ready": (pb.Ready, None), + MSG_READY: (pb.Ready, None), # main → sandbox - "sandbox/entry_setup": (pb.EntrySetup, pb.EntrySetupResult), - "sandbox/entry_unload": (pb.EntryUnload, pb.EntryUnloadResult), - "sandbox/call_service": (pb.CallService, pb.CallServiceResult), - "sandbox/entity_query": (pb.EntityQuery, pb.EntityQueryResult), - "sandbox/get_translations": (pb.GetTranslations, pb.GetTranslationsResult), - "sandbox/shutdown": (pb.Shutdown, pb.ShutdownResult), - "sandbox/ping": (pb.Ping, pb.PingResult), - "sandbox/flow_init": (pb.FlowInit, pb.FlowResult), - "sandbox/flow_step": (pb.FlowStep, pb.FlowResult), - "sandbox/flow_abort": (pb.FlowAbort, pb.FlowAbortResult), + MSG_ENTRY_SETUP: (pb.EntrySetup, pb.EntrySetupResult), + MSG_ENTRY_UNLOAD: (pb.EntryUnload, pb.EntryUnloadResult), + MSG_CALL_SERVICE: (pb.CallService, pb.CallServiceResult), + MSG_ENTITY_QUERY: (pb.EntityQuery, pb.EntityQueryResult), + MSG_GET_TRANSLATIONS: (pb.GetTranslations, pb.GetTranslationsResult), + MSG_SHUTDOWN: (pb.Shutdown, pb.ShutdownResult), + MSG_PING: (pb.Ping, pb.PingResult), + MSG_FLOW_INIT: (pb.FlowInit, pb.FlowResult), + MSG_FLOW_STEP: (pb.FlowStep, pb.FlowResult), + MSG_FLOW_ABORT: (pb.FlowAbort, pb.FlowAbortResult), # sandbox → main - "sandbox/register_entity": (pb.EntityDescription, pb.RegisterEntityResult), - "sandbox/unregister_entity": (pb.UnregisterEntity, pb.UnregisterEntityResult), - "sandbox/state_changed": (pb.StateChanged, None), - "sandbox/register_service": (pb.RegisterService, pb.RegisterServiceResult), - "sandbox/unregister_service": ( + MSG_REGISTER_ENTITY: (pb.EntityDescription, pb.RegisterEntityResult), + MSG_UNREGISTER_ENTITY: (pb.UnregisterEntity, pb.UnregisterEntityResult), + MSG_STATE_CHANGED: (pb.StateChanged, None), + MSG_REGISTER_SERVICE: (pb.RegisterService, pb.RegisterServiceResult), + MSG_UNREGISTER_SERVICE: ( pb.UnregisterService, pb.UnregisterServiceResult, ), - "sandbox/fire_event": (pb.FireEvent, None), - "sandbox/store_load": (pb.StoreLoad, pb.StoreLoadResult), - "sandbox/store_save": (pb.StoreSave, pb.StoreSaveResult), - "sandbox/store_remove": (pb.StoreRemove, pb.StoreRemoveResult), + MSG_FIRE_EVENT: (pb.FireEvent, None), + MSG_STORE_LOAD: (pb.StoreLoad, pb.StoreLoadResult), + MSG_STORE_SAVE: (pb.StoreSave, pb.StoreSaveResult), + MSG_STORE_REMOVE: (pb.StoreRemove, pb.StoreRemoveResult), } @@ -203,6 +324,26 @@ def make_entity_description( __all__ = [ + "MSG_CALL_SERVICE", + "MSG_ENTITY_QUERY", + "MSG_ENTRY_SETUP", + "MSG_ENTRY_UNLOAD", + "MSG_FIRE_EVENT", + "MSG_FLOW_ABORT", + "MSG_FLOW_INIT", + "MSG_FLOW_STEP", + "MSG_GET_TRANSLATIONS", + "MSG_PING", + "MSG_READY", + "MSG_REGISTER_ENTITY", + "MSG_REGISTER_SERVICE", + "MSG_SHUTDOWN", + "MSG_STATE_CHANGED", + "MSG_STORE_LOAD", + "MSG_STORE_REMOVE", + "MSG_STORE_SAVE", + "MSG_UNREGISTER_ENTITY", + "MSG_UNREGISTER_SERVICE", "REGISTRY", "decode_json", "decode_json_dict", diff --git a/sandbox/hass_client/hass_client/protocol.py b/sandbox/hass_client/hass_client/protocol.py deleted file mode 100644 index 1fed5c3ab7ff..000000000000 --- a/sandbox/hass_client/hass_client/protocol.py +++ /dev/null @@ -1,143 +0,0 @@ -"""Wire-protocol message-type constants. - -The integration and the sandbox runtime exchange typed protobuf messages -over the :class:`Channel`. Each message type is namespaced ``sandbox/…``; -this module holds the type-string constants. Both sides share the same -names — kept here on the HA side and mirrored verbatim in -:mod:`hass_client.protocol` so neither has to import the other. - -The wire is protobuf (codec :class:`~.codec_protobuf.ProtobufCodec`, which a -:class:`~.channel.Channel` now requires explicitly): each ``type`` maps to a -request/result proto message pair in :mod:`.messages` (the `REGISTRY`), -generated from ``sandbox/proto/sandbox.proto``. The payload shapes described -below are the *logical* contract for each call — they are carried as those -typed proto messages, not free-form dicts (only genuinely dynamic fields, e.g. -``service_data`` / state attributes / serialized voluptuous schemas, cross -as orjson-encoded JSON ``bytes``). A registry-free line-oriented JSON codec lives -in the test helpers as the channel-core test/debug wire. - -Main → Sandbox calls: - -* ``sandbox/entry_setup`` — push a serialised :class:`ConfigEntry` into - the sandbox, asking it to load the owning integration and run - ``async_setup_entry``. Returns ``{"ok": bool, "reason": str | None}``. - Carries an ``integration_source`` sub-message telling a stateless sandbox - where to fetch the integration code: ``{kind: "builtin"}`` (the bundled - ``homeassistant`` package provides it — a no-op) or ``{kind: "git", url, - ref, tag, domain, subdir}`` for custom (HACS) integrations. ``ref`` is an - exact commit sha (main pins tag→sha; see ``sources.py``); the sandbox - fetches the code before setup (see ``hass_client.sources``). -* ``sandbox/entry_unload`` — ask the sandbox to unload an entry by id. -* ``sandbox/call_service`` — generic service dispatch (shared with - the main→sandbox service mirroring path). Payload mirrors a - ``ServiceCall``: ``(domain, service, target, service_data, context, - return_response)``. Returns either ``None`` or a service-response dict. -* ``sandbox/entity_query`` — generic request/response RPC for the - server-side entity queries with no ``SupportsResponse`` service to ride - (media search, update release notes, vacuum segments, the WS-only calendar - event edits). Payload ``{sandbox_entity_id, method, args, context_id}``; - the sandbox resolves the entity, invokes ``method`` with ``args`` as kwargs, - and returns the serialised result wrapped as ``{"value": }``. - Ops that map to a ``SupportsResponse`` service use ``call_service`` instead. -* ``sandbox/get_translations`` — pull a sandboxed integration's frontend - translation strings. Payload ``{language, domains: [str]}`` (main batches - every owned custom domain of one group into a single request). Response - ``{language, strings: {domain: }}`` — the - un-flattened nesting a ``translations/.json`` holds, with ``title`` - pre-filled from the integration name (main has no ``Integration`` for a - custom domain, so it cannot run that fallback). Built-in domains never - cross the wire — main reads its byte-identical disk copy. - -Sandbox → Main calls: - -* ``sandbox/register_entity`` — sandbox tells main "I just added an - entity, here's its description". Main builds the proxy and replies - ``{"entity_id": }`` so the sandbox can route later - ``call_service`` requests back to the right local entity. Optional - ``device_info`` field: a JSON-flattened ``DeviceInfo`` dict - — sets become lists of two-element lists (``identifiers`` / - ``connections``), tuples become lists (``via_device``), and - ``entry_type`` is the enum's string value. When present, main calls - :func:`device_registry.async_get_or_create` so the sandbox's devices - surface in main's device_registry tied to the sandboxed entry. -* ``sandbox/unregister_entity`` — symmetric counterpart. -* ``sandbox/state_changed`` — push (no response). Carries the - marshalled state delta for one entity. -* ``sandbox/register_service`` — sandbox tells main "I just - registered a service, please mirror it". Main installs a thin handler - that forwards calls back over the shared ``sandbox/call_service`` - channel. -* ``sandbox/unregister_service`` — symmetric counterpart. -* ``sandbox/fire_event`` — push (no response). The sandbox - forwards each ``_*`` event so main listeners (notably - ``automation``) can react as if the integration ran locally. -* ``sandbox/store_load`` — sandbox-side ``Store.async_load`` - proxies to this RPC. Payload ``{"key": str}``; response is the wrapped - ``{"version", "minor_version", "key", "data"}`` dict the sandbox last - saved, or ``None`` if no data exists yet. The group is implicit from - the channel — each :class:`SandboxBridge` only ever serves one group. -* ``sandbox/store_save`` — sandbox-side ``Store`` flush. - Payload ``{"key": str, "data": dict}``; main writes the wrapped dict - to ``/.storage/sandbox//`` atomically. Response - is ``{"ok": True}``. -* ``sandbox/store_remove`` — sandbox-side - ``Store.async_remove``. Payload ``{"key": str}``; main unlinks the - file (if any). Response is ``{"ok": True}``. - -Main → Sandbox shutdown: - -* ``sandbox/shutdown`` — ask the runtime to unload its entries, dump - ``RestoreEntity`` state, fire ``EVENT_HOMEASSISTANT_FINAL_WRITE`` so any - pending Stores flush to main via the ``current_sandbox`` store bridge, - and exit cleanly. Response ``{"ok": True, "unloaded": int, "restored": - int}``. The runtime sets its shutdown event right after writing the - reply, so the subprocess exits 0 on its own — main only needs SIGTERM - if the round-trip times out. -""" - -from typing import Final - -# Handshake (Sandbox → Main): the runtime's first frame on the channel. -# Replaces the old ``sandbox:ready`` stdout text marker — the manager -# registers a handler for this push and treats its arrival as "running", -# so stdout carries nothing but channel frames. -MSG_READY: Final = "sandbox/ready" - -# Main → Sandbox -MSG_ENTRY_SETUP: Final = "sandbox/entry_setup" -MSG_ENTRY_UNLOAD: Final = "sandbox/entry_unload" -MSG_CALL_SERVICE: Final = "sandbox/call_service" -MSG_ENTITY_QUERY: Final = "sandbox/entity_query" -MSG_GET_TRANSLATIONS: Final = "sandbox/get_translations" -MSG_SHUTDOWN: Final = "sandbox/shutdown" - -# Sandbox → Main -MSG_REGISTER_ENTITY: Final = "sandbox/register_entity" -MSG_UNREGISTER_ENTITY: Final = "sandbox/unregister_entity" -MSG_STATE_CHANGED: Final = "sandbox/state_changed" -MSG_REGISTER_SERVICE: Final = "sandbox/register_service" -MSG_UNREGISTER_SERVICE: Final = "sandbox/unregister_service" -MSG_FIRE_EVENT: Final = "sandbox/fire_event" -MSG_STORE_LOAD: Final = "sandbox/store_load" -MSG_STORE_SAVE: Final = "sandbox/store_save" -MSG_STORE_REMOVE: Final = "sandbox/store_remove" - - -__all__ = [ - "MSG_CALL_SERVICE", - "MSG_ENTITY_QUERY", - "MSG_ENTRY_SETUP", - "MSG_ENTRY_UNLOAD", - "MSG_FIRE_EVENT", - "MSG_GET_TRANSLATIONS", - "MSG_READY", - "MSG_REGISTER_ENTITY", - "MSG_REGISTER_SERVICE", - "MSG_SHUTDOWN", - "MSG_STATE_CHANGED", - "MSG_STORE_LOAD", - "MSG_STORE_REMOVE", - "MSG_STORE_SAVE", - "MSG_UNREGISTER_ENTITY", - "MSG_UNREGISTER_SERVICE", -] diff --git a/sandbox/hass_client/hass_client/sandbox/__init__.py b/sandbox/hass_client/hass_client/sandbox/__init__.py index 0e86f14cf48c..85a071687251 100644 --- a/sandbox/hass_client/hass_client/sandbox/__init__.py +++ b/sandbox/hass_client/hass_client/sandbox/__init__.py @@ -38,8 +38,13 @@ from hass_client.entity_bridge import EntityBridge from hass_client.entry_runner import EntryRunner from hass_client.event_mirror import EventMirror from hass_client.flow_runner import FlowRunner -from hass_client.messages import encode_json -from hass_client.protocol import MSG_GET_TRANSLATIONS, MSG_READY, MSG_SHUTDOWN +from hass_client.messages import ( + MSG_GET_TRANSLATIONS, + MSG_PING, + MSG_READY, + MSG_SHUTDOWN, + encode_json, +) from hass_client.sandbox_bridge import ChannelSandboxBridge from hass_client.service_mirror import ServiceMirror from homeassistant.const import EVENT_HOMEASSISTANT_FINAL_WRITE @@ -192,7 +197,7 @@ class SandboxRuntime: # Ready, so an `entry_setup` arriving in the gap between Ready and # handler registration used to hit `ChannelUnknownType` -> # SETUP_ERROR. Registering first removes that race entirely. - self._channel.register("sandbox/ping", _handle_ping) + self._channel.register(MSG_PING, _handle_ping) self._channel.register(MSG_SHUTDOWN, self._handle_shutdown) self._channel.register( MSG_GET_TRANSLATIONS, self._handle_get_translations diff --git a/sandbox/hass_client/hass_client/sandbox_bridge.py b/sandbox/hass_client/hass_client/sandbox_bridge.py index 8301f0bc6438..ba7f89db10ac 100644 --- a/sandbox/hass_client/hass_client/sandbox_bridge.py +++ b/sandbox/hass_client/hass_client/sandbox_bridge.py @@ -21,8 +21,7 @@ from homeassistant.util.json import SerializationError from ._proto import sandbox_pb2 as pb from .channel import Channel, ChannelClosedError, ChannelRemoteError -from .messages import decode_json_dict -from .protocol import MSG_STORE_LOAD, MSG_STORE_REMOVE, MSG_STORE_SAVE +from .messages import MSG_STORE_LOAD, MSG_STORE_REMOVE, MSG_STORE_SAVE, decode_json_dict _LOGGER = logging.getLogger(__name__) diff --git a/sandbox/hass_client/hass_client/service_mirror.py b/sandbox/hass_client/hass_client/service_mirror.py index 89acd2305666..14b5a7c7b298 100644 --- a/sandbox/hass_client/hass_client/service_mirror.py +++ b/sandbox/hass_client/hass_client/service_mirror.py @@ -24,13 +24,12 @@ from homeassistant.const import ( EVENT_SERVICE_REGISTERED, EVENT_SERVICE_REMOVED, ) -from homeassistant.core import Event, HomeAssistant, callback +from homeassistant.core import Event, HomeAssistant, Service, callback from ._proto import sandbox_pb2 as pb from .approved_domains import ApprovedDomains from .channel import Channel -from .messages import encode_json -from .protocol import MSG_REGISTER_SERVICE, MSG_UNREGISTER_SERVICE +from .messages import MSG_REGISTER_SERVICE, MSG_UNREGISTER_SERVICE, encode_json from .schema_bridge import serialize_schema _LOGGER = logging.getLogger(__name__) @@ -124,13 +123,19 @@ class ServiceMirror: key = (domain.lower(), service.lower()) if key in self._mirrored: return - supports_response = _supports_response(self.hass, domain, service) + # One registry lookup feeds both metadata fields. Best-effort: the + # service may not be visible yet (a race with the + # EVENT_SERVICE_REGISTERED listener) — main treats the metadata as + # authoritative and the sandbox's own handler still validates. + service_obj = self.hass.services.async_services_for_domain(domain).get( + service.lower() + ) msg = pb.RegisterService( domain=domain, service=service, - supports_response=supports_response, + supports_response=_supports_response(service_obj), ) - schema = _service_schema(self.hass, domain, service) + schema = serialize_schema(service_obj.schema) if service_obj else None if schema: msg.schema = encode_json(schema) self._mirrored.add(key) @@ -182,36 +187,14 @@ class ServiceMirror: ) -def _service_schema( - hass: HomeAssistant, domain: str, service: str -) -> list[dict[str, Any]] | None: - """Serialise the registered service's voluptuous schema for the wire. - - Returns ``None`` when the service registers with no schema (very - common), when the schema doesn't survive voluptuous_serialize, or - when the lookup races and the service isn't visible yet — in every - case main falls back to ``schema=None`` and the sandbox's own - handler still validates. - """ - services = hass.services.async_services_for_domain(domain) - service_obj = services.get(service.lower()) - if service_obj is None: - return None - return serialize_schema(service_obj.schema) - - -def _supports_response(hass: HomeAssistant, domain: str, service: str) -> str: - """Best-effort lookup of the service's ``supports_response`` value. +def _supports_response(service_obj: Service | None) -> str: + """Extract the service's ``supports_response`` value for the wire. Returns the lowercase string value (``"none"`` / ``"only"`` / ``"optional"``) since that's what main needs to pass back to - :meth:`hass.services.async_register`. Falls back to ``"none"`` if - the service isn't actually registered yet (a race with the - ``EVENT_SERVICE_REGISTERED`` listener) — the lookup is best-effort - and main treats the metadata as authoritative. + :meth:`hass.services.async_register`. Falls back to ``"none"`` for a + service that isn't actually registered yet. """ - services = hass.services.async_services_for_domain(domain) - service_obj = services.get(service.lower()) if service_obj is None: return "none" value = getattr(service_obj.supports_response, "value", None) diff --git a/sandbox/hass_client/hass_client/sources.py b/sandbox/hass_client/hass_client/sources.py index e07c963b3c3f..53c8c1c15f7c 100644 --- a/sandbox/hass_client/hass_client/sources.py +++ b/sandbox/hass_client/hass_client/sources.py @@ -8,8 +8,11 @@ descriptor on ``entry_setup`` that the sandbox fetches into :meth:`hass_client.entry_runner.EntryRunner._handle_entry_setup`). The fetch uses GitHub's codeload tarball for the exact commit sha (no ``git`` -binary dependency, matching HACS). A process-lifetime cache keyed by -``(url, ref)`` means multiple entries sourced from the same repo download once. +binary dependency, matching HACS). Concurrent fetches of the same ``(url, +ref)`` share one in-flight download (single-flight); different repos download +in parallel. Nothing pins tarball bytes past the extract — the extracted tree +under ``custom_components`` is the artifact, so a later same-repo fetch for a +different subdir simply re-downloads. The download primitive is injectable so tests substitute a local fixture for the real network fetch — no test ever hits GitHub. @@ -17,6 +20,7 @@ the real network fetch — no test ever hits GitHub. import asyncio from collections.abc import Awaitable, Callable +from functools import partial import io import logging from pathlib import Path @@ -26,11 +30,20 @@ from ._proto import sandbox_pb2 as pb _LOGGER = logging.getLogger(__name__) -# url, ref -> downloaded tarball bytes. Process-lifetime only (honours -# "stateless": nothing survives a process restart). Guarded by _CACHE_LOCK so -# concurrent entries from the same repo download exactly once. -_TARBALL_CACHE: dict[tuple[str, str], bytes] = {} -_CACHE_LOCK = asyncio.Lock() +# Single-flight downloads keyed by (url, ref): concurrent fetches of the same +# repo await one shared task; entries are dropped when the download finishes, +# so tarball bytes are never pinned for the process lifetime. _COMPLETED +# remembers which keys already downloaded once (log signal only — the extract +# for a new subdir must re-download regardless). +_INFLIGHT: dict[tuple[str, str], asyncio.Task[bytes]] = {} +_COMPLETED: set[tuple[str, str]] = set() + + +def _on_fetch_done(key: tuple[str, str], task: asyncio.Task[bytes]) -> None: + """Drop a finished download from the single-flight map.""" + _INFLIGHT.pop(key, None) + if not task.cancelled() and task.exception() is None: + _COMPLETED.add(key) # (repo url, exact sha) -> tarball bytes. FetchPrimitive = Callable[[str, str], Awaitable[bytes]] @@ -82,18 +95,20 @@ async def async_ensure_integration_source( fetcher = fetch if fetch is not None else _default_fetch key = (source.url, source.ref) - async with _CACHE_LOCK: - tarball = _TARBALL_CACHE.get(key) - if tarball is None: - _LOGGER.info( - "sandbox: fetching %s from %s@%s (%s)", - domain, - source.url, - source.ref, - source.tag or "no tag", - ) - tarball = await fetcher(source.url, source.ref) - _TARBALL_CACHE[key] = tarball + task = _INFLIGHT.get(key) + if task is None: + _LOGGER.info( + "sandbox: fetching %s from %s@%s (%s)%s", + domain, + source.url, + source.ref, + source.tag or "no tag", + " — re-download for a new subdir" if key in _COMPLETED else "", + ) + task = asyncio.get_running_loop().create_task(fetcher(source.url, source.ref)) + _INFLIGHT[key] = task + task.add_done_callback(partial(_on_fetch_done, key)) + tarball = await task await asyncio.get_running_loop().run_in_executor( None, _extract_subdir, tarball, subdir, dest diff --git a/sandbox/hass_client/hass_client/testing/pytest_plugin.py b/sandbox/hass_client/hass_client/testing/pytest_plugin.py index 8a5db92b0f93..3ca497a8b39f 100644 --- a/sandbox/hass_client/hass_client/testing/pytest_plugin.py +++ b/sandbox/hass_client/hass_client/testing/pytest_plugin.py @@ -115,7 +115,7 @@ class InProcessSandbox: the private hass's timers onto the shared test loop. """ if self.channel is not None and not self.channel.closed: - from hass_client.protocol import MSG_SHUTDOWN # noqa: PLC0415 + from hass_client.messages import MSG_SHUTDOWN # noqa: PLC0415 with contextlib.suppress(Exception): await asyncio.wait_for( @@ -164,7 +164,7 @@ class _InProcessSandboxProcess: """Best-effort: issue a shutdown call so the runtime exits cleanly.""" # Lazy import: testing package must not pull the HA integration # tree at import time. - from homeassistant.components.sandbox.protocol import ( # noqa: PLC0415 + from homeassistant.components.sandbox.messages import ( # noqa: PLC0415 MSG_SHUTDOWN, ) @@ -242,7 +242,6 @@ async def async_setup_inprocess_sandbox( # Mirror what the integration's ``_on_channel_ready`` does when the # real ``SandboxProcess`` opens its channel — register the bridge. - data.channels[group] = mgr_channel data.bridges[group] = async_create_bridge(hass, group=group, channel=mgr_channel) mgr_channel.start() diff --git a/sandbox/hass_client/tests/test_sandbox_runtime.py b/sandbox/hass_client/tests/test_sandbox_runtime.py index 7287a61e273e..2333be9db163 100644 --- a/sandbox/hass_client/tests/test_sandbox_runtime.py +++ b/sandbox/hass_client/tests/test_sandbox_runtime.py @@ -11,7 +11,7 @@ import asyncio from hass_client._proto import sandbox_pb2 as pb from hass_client.channel import Channel, ChannelRemoteError from hass_client.codec_protobuf import ProtobufCodec -from hass_client.protocol import MSG_READY +from hass_client.messages import MSG_READY from hass_client.sandbox import SandboxRuntime from hass_client.sandbox.__main__ import _build_parser import pytest diff --git a/sandbox/hass_client/tests/test_shutdown.py b/sandbox/hass_client/tests/test_shutdown.py index 236fbb07c465..b5b48bfc5c8c 100644 --- a/sandbox/hass_client/tests/test_shutdown.py +++ b/sandbox/hass_client/tests/test_shutdown.py @@ -19,8 +19,12 @@ from typing import Any from hass_client._proto import sandbox_pb2 as pb from hass_client.channel import Channel from hass_client.codec_protobuf import ProtobufCodec -from hass_client.messages import decode_json_dict -from hass_client.protocol import MSG_SHUTDOWN, MSG_STORE_LOAD, MSG_STORE_SAVE +from hass_client.messages import ( + MSG_SHUTDOWN, + MSG_STORE_LOAD, + MSG_STORE_SAVE, + decode_json_dict, +) from hass_client.sandbox import SandboxRuntime import pytest diff --git a/sandbox/hass_client/tests/test_sources.py b/sandbox/hass_client/tests/test_sources.py index fff1612837fa..fde216d0e795 100644 --- a/sandbox/hass_client/tests/test_sources.py +++ b/sandbox/hass_client/tests/test_sources.py @@ -3,6 +3,7 @@ All fetches use a local in-memory tarball fixture — no test hits the network. """ +import asyncio from collections.abc import Iterator import io from pathlib import Path @@ -15,11 +16,13 @@ import pytest @pytest.fixture(autouse=True) -def _clear_tarball_cache() -> Iterator[None]: - """Reset the process-lifetime tarball cache between tests for isolation.""" - sources_module._TARBALL_CACHE.clear() # noqa: SLF001 +def _clear_fetch_state() -> Iterator[None]: + """Reset the single-flight download state between tests for isolation.""" + sources_module._INFLIGHT.clear() # noqa: SLF001 + sources_module._COMPLETED.clear() # noqa: SLF001 yield - sources_module._TARBALL_CACHE.clear() # noqa: SLF001 + sources_module._INFLIGHT.clear() # noqa: SLF001 + sources_module._COMPLETED.clear() # noqa: SLF001 def _make_tarball( @@ -108,8 +111,50 @@ async def test_git_source_extracts_into_config_dir(tmp_path: Path) -> None: assert not (tmp_path / "README.md").exists() -async def test_second_call_same_ref_hits_cache(tmp_path: Path) -> None: - """Two entries from the same (url, ref) download once.""" +async def test_concurrent_same_ref_shares_one_download(tmp_path: Path) -> None: + """Concurrent entries from the same (url, ref) share one in-flight fetch.""" + tarball = _make_tarball( + top="my_custom-aaaa", + files={ + "custom_components/foo/manifest.json": "{}", + "custom_components/bar/manifest.json": "{}", + }, + ) + calls = 0 + + async def _fetch(url: str, ref: str) -> bytes: + nonlocal calls + calls += 1 + await asyncio.sleep(0) + return tarball + + source_foo = pb.IntegrationSource( + kind="git", + url="https://github.com/owner/repo", + ref="z" * 40, + domain="foo", + subdir="custom_components/foo", + ) + source_bar = pb.IntegrationSource( + kind="git", + url="https://github.com/owner/repo", + ref="z" * 40, + domain="bar", + subdir="custom_components/bar", + ) + + await asyncio.gather( + async_ensure_integration_source(str(tmp_path), source_foo, fetch=_fetch), + async_ensure_integration_source(str(tmp_path), source_bar, fetch=_fetch), + ) + + assert calls == 1 + assert (tmp_path / "custom_components" / "foo" / "manifest.json").exists() + assert (tmp_path / "custom_components" / "bar" / "manifest.json").exists() + + +async def test_sequential_new_subdir_redownloads(tmp_path: Path) -> None: + """A finished download is not pinned — a later new-subdir fetch re-downloads.""" tarball = _make_tarball( top="my_custom-aaaa", files={ @@ -142,7 +187,7 @@ async def test_second_call_same_ref_hits_cache(tmp_path: Path) -> None: await async_ensure_integration_source(str(tmp_path), source_foo, fetch=_fetch) await async_ensure_integration_source(str(tmp_path), source_bar, fetch=_fetch) - assert calls == 1 + assert calls == 2 assert (tmp_path / "custom_components" / "foo" / "manifest.json").exists() assert (tmp_path / "custom_components" / "bar" / "manifest.json").exists() diff --git a/sandbox/proto/check_mirror_drift.sh b/sandbox/proto/check_mirror_drift.sh index 8ecdd515de61..35a08b6f3adf 100755 --- a/sandbox/proto/check_mirror_drift.sh +++ b/sandbox/proto/check_mirror_drift.sh @@ -1,7 +1,8 @@ #!/usr/bin/env bash # Drift guard for the hand-mirrored sandbox wire modules. # -# channel.py, codec_protobuf.py and messages.py are maintained as byte-identical +# channel.py, codec_protobuf.py, messages.py and the checked-in protobuf +# gencode (_proto/sandbox_pb2.py + .pyi) are maintained as byte-identical # copies in two places: # # homeassistant/components/sandbox/ (HA Core integration side) @@ -12,9 +13,10 @@ # ``homeassistant.components.*``. This guard fails if any pair diverges, so the # "edit both copies" rule is enforced instead of trusted. # -# Unlike the proto gencode guard (check_drift.sh) this is a plain ``diff`` with -# no external tooling, so it is wired as a regular every-commit prek hook that -# fires whenever a mirrored file changes. +# Unlike the proto regeneration guard (check_drift.sh, which re-runs protoc to +# compare against sandbox.proto) this is a plain ``diff`` with no external +# tooling, so it is wired as a regular every-commit prek hook that fires +# whenever a mirrored file changes. set -euo pipefail @@ -24,7 +26,13 @@ cd "${REPO_ROOT}" HA_DIR="homeassistant/components/sandbox" CLIENT_DIR="sandbox/hass_client/hass_client" -MIRRORED_FILES=(channel.py codec_protobuf.py messages.py) +MIRRORED_FILES=( + channel.py + codec_protobuf.py + messages.py + _proto/sandbox_pb2.py + _proto/sandbox_pb2.pyi +) status=0 for file in "${MIRRORED_FILES[@]}"; do diff --git a/tests/components/sandbox/test_channel.py b/tests/components/sandbox/test_channel.py index e14c2f8d1126..c24cdf133ea0 100644 --- a/tests/components/sandbox/test_channel.py +++ b/tests/components/sandbox/test_channel.py @@ -21,7 +21,7 @@ class _QueueTransport: """In-memory :class:`Transport` backed by a pair of queues. Stands in for a non-stream transport (the seam a future - ``WebSocketTransport`` uses) so :meth:`Channel.from_transport` is + ``WebSocketTransport`` uses) so ``Channel(transport=...)`` is exercised without any reader/writer pipe. """ @@ -47,16 +47,12 @@ class _QueueTransport: return None -async def test_from_transport_round_trips() -> None: +async def test_transport_channel_round_trips() -> None: """A channel built over an arbitrary Transport dispatches normally.""" q1: asyncio.Queue[bytes | None] = asyncio.Queue() q2: asyncio.Queue[bytes | None] = asyncio.Queue() - channel_a = Channel.from_transport( - _QueueTransport(q1, q2), name="a", codec=JsonCodec() - ) - channel_b = Channel.from_transport( - _QueueTransport(q2, q1), name="b", codec=JsonCodec() - ) + channel_a = Channel(transport=_QueueTransport(q1, q2), name="a", codec=JsonCodec()) + channel_b = Channel(transport=_QueueTransport(q2, q1), name="b", codec=JsonCodec()) channel_a.start() channel_b.start() @@ -396,7 +392,7 @@ async def test_close_after_eof_still_closes_transport() -> None: inflight exactly once — otherwise the byte channel leaks every restart. """ transport = _ObservableTransport() - channel = Channel.from_transport(transport, name="eof", codec=JsonCodec()) + channel = Channel(transport=transport, name="eof", codec=JsonCodec()) started = asyncio.Event() cancelled = asyncio.Event() diff --git a/tests/components/sandbox/test_device_registry.py b/tests/components/sandbox/test_device_registry.py index 5b1dfe4d64ca..87a382dfac3a 100644 --- a/tests/components/sandbox/test_device_registry.py +++ b/tests/components/sandbox/test_device_registry.py @@ -103,10 +103,10 @@ async def test_register_entity_creates_device_entry( assert er_entry.device_id == device.id -async def test_register_entity_propagates_device_id_to_proxy( +async def test_register_entity_wires_proxy_device_entry( hass: HomeAssistant, entry: ConfigEntry ) -> None: - """The proxy entity reports the freshly-created device_id.""" + """The proxy entity is linked to the freshly-created DeviceEntry.""" bridge, main_channel, sandbox_channel = await _wire(hass) payload = _register_payload( entry, @@ -123,16 +123,16 @@ async def test_register_entity_propagates_device_id_to_proxy( await sandbox_channel.close() proxy = bridge._entities["light.kitchen"] - assert proxy.description.device_id is not None - device = dr.async_get(hass).async_get(proxy.description.device_id) + device = dr.async_get(hass).async_get_device( + identifiers={("sandboxed_hue", "bulb-002")} + ) assert device is not None - assert ("sandboxed_hue", "bulb-002") in device.identifiers - # The framework also wired entity.device_entry through async_add_entities. + # The framework wired entity.device_entry through async_add_entities. assert proxy.device_entry is not None - assert proxy.device_entry.id == proxy.description.device_id + assert proxy.device_entry.id == device.id -async def test_register_entity_without_device_info_leaves_device_id_unset( +async def test_register_entity_without_device_info_creates_no_device( hass: HomeAssistant, entry: ConfigEntry ) -> None: """Backwards compatibility: no device_info in payload → no device registered.""" @@ -147,7 +147,6 @@ async def test_register_entity_without_device_info_leaves_device_id_unset( proxy = bridge._entities["light.kitchen"] assert proxy.description.device_info is None - assert proxy.description.device_id is None # No device created against this entry. assert not any( entry.entry_id in d.config_entries for d in dr.async_get(hass).devices.values() diff --git a/tests/components/sandbox/test_init.py b/tests/components/sandbox/test_init.py index f366f719077b..0072755bd036 100644 --- a/tests/components/sandbox/test_init.py +++ b/tests/components/sandbox/test_init.py @@ -18,5 +18,4 @@ async def test_setup_installs_manager_router_and_hook( assert isinstance(data.manager, SandboxManager) assert isinstance(data.router, SandboxFlowRouter) assert hass.config_entries.router is data.router - assert data.channels == {} assert data.bridges == {} diff --git a/tests/components/sandbox/test_testing_plugins.py b/tests/components/sandbox/test_testing_plugins.py index dc9c0b90536b..c508765309fa 100644 --- a/tests/components/sandbox/test_testing_plugins.py +++ b/tests/components/sandbox/test_testing_plugins.py @@ -58,7 +58,6 @@ async def test_inprocess_plugin_wires_manager_and_bridge( """The plugin installs a bridge for the group and parks a fake process.""" data = hass.data[DATA_SANDBOX] assert in_process_sandbox.group == DEFAULT_GROUP - assert DEFAULT_GROUP in data.channels assert DEFAULT_GROUP in data.bridges manager = data.manager assert manager is not None @@ -78,7 +77,7 @@ async def test_inprocess_plugin_round_trips_ping( runtime's handler and returns the same payload a subprocess would. """ data = hass.data[DATA_SANDBOX] - channel = data.channels[DEFAULT_GROUP] + channel = data.bridges[DEFAULT_GROUP].channel result = await asyncio.wait_for(channel.call("sandbox/ping", None), timeout=2.0) assert result.pong == "sandbox" diff --git a/tests/components/sandbox/test_translation.py b/tests/components/sandbox/test_translation.py index d762549bf8ef..49e5e2027295 100644 --- a/tests/components/sandbox/test_translation.py +++ b/tests/components/sandbox/test_translation.py @@ -16,8 +16,7 @@ import pytest from homeassistant.components.sandbox import SandboxData from homeassistant.components.sandbox._proto import sandbox_pb2 as pb from homeassistant.components.sandbox.channel import Channel -from homeassistant.components.sandbox.messages import encode_json -from homeassistant.components.sandbox.protocol import MSG_GET_TRANSLATIONS +from homeassistant.components.sandbox.messages import MSG_GET_TRANSLATIONS, encode_json from homeassistant.components.sandbox.proxy_flow import SandboxFlowProxy from homeassistant.components.sandbox.router import SandboxFlowRouter from homeassistant.components.sandbox.translation import SandboxTranslationProvider diff --git a/tests/test_config_entries.py b/tests/test_config_entries.py index 41ca67e092af..1c1ae9e5b5eb 100644 --- a/tests/test_config_entries.py +++ b/tests/test_config_entries.py @@ -9699,34 +9699,8 @@ async def test_sandbox_absent_from_storage_loads_as_none( assert entries[0].sandbox is None -async def test_async_update_entry_sets_sandbox(hass: HomeAssistant) -> None: - """``async_update_entry(entry, sandbox=...)`` mutates and persists the field.""" - entry = MockConfigEntry(domain="test") - entry.add_to_hass(hass) - assert entry.sandbox is None - - changed = hass.config_entries.async_update_entry(entry, sandbox="built-in") - assert changed is True - assert entry.sandbox == "built-in" - - # Idempotent update returns False. - changed = hass.config_entries.async_update_entry(entry, sandbox="built-in") - assert changed is False - - # Storage cache is refreshed so the new value lands on disk. - stored = json_loads(json_dumps(entry.as_storage_fragment)) - assert stored["sandbox"] == "built-in" - - # And it can be cleared back to None. - changed = hass.config_entries.async_update_entry(entry, sandbox=None) - assert changed is True - assert entry.sandbox is None - stored = json_loads(json_dumps(entry.as_storage_fragment)) - assert "sandbox" not in stored - - def test_sandbox_cannot_be_set_directly() -> None: - """``entry.sandbox = ...`` is rejected — must go through update_entry.""" + """``entry.sandbox = ...`` is rejected — the field is frozen after init.""" entry = MockConfigEntry(domain="test") with pytest.raises(AttributeError): entry.sandbox = "built-in"