starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-24 07:25:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

sandbox: fold protocol.py into messages.py; delete dead code; client simplifications

Browse Source 
C1 — the wire-type constants lived in protocol.py, hand-mirrored but
excluded from the drift guard and missing four types (ping + the three
flow ops appeared as string literals in six files). All MSG_* constants
now live in the guarded, mirrored messages.py, REGISTRY is keyed by
them, production literals are gone, and both protocol.py copies are
deleted. The drift guard (script + pre-commit files pattern) now also
byte-diffs the checked-in _proto gencode pair.

C5 — dead code, each re-verified before deletion: the unused
async_update_entry(sandbox=...) plumbing (the field is now in
FROZEN_CONFIG_ENTRY_ATTRS — stronger immutability, smaller core diff);
SandboxEntityDescription.device_id (write-only);
_parse_supports_response's unreachable branches;
_translate_remote_error's duplicate branches; SandboxData.channels (no
production reader — consumers use bridges[group].channel);
drain_inflight's dead reader-task filter; Channel.from_transport
(test-only duplicate of __init__(transport=...)); _serialize_body's
None branch; uint32 wrap for _next_id on a months-lived channel;
defensive getattr on always-present dataclass fields.

E6 — sources.py: the global fetch lock serialized downloads of
different repos and _TARBALL_CACHE pinned whole tarballs for process
lifetime; per-key single-flight tasks + a completed-key set replace
both. service_mirror does one services_for_domain lookup per
registration instead of two. event_mirror's docstring described an
event_filter fast path that cannot exist (core filters receive only
event.data, and data-less events skip filtered listeners) — the
docstring now tells the truth instead.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QCotUYum6AoisyrxshoiJJ
This commit is contained in:
Paulus Schoutsen
2026-07-07 16:59:14 -04:00
co-authored by Claude Fable 5
parent 50c6845f28
commit 0d075d1303
36 changed files with 584 additions and 590 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.pre-commit-config.yaml
+3 -2
View File
@@ -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
homeassistant/components/sandbox/__init__.py
-3
View File
@@ -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()
homeassistant/components/sandbox/bridge.py
+10 -18
View File
@@ -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}")
homeassistant/components/sandbox/channel.py
+5 -24
View File
@@ -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:
homeassistant/components/sandbox/codec_protobuf.py
+6 -2
View File
@@ -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(
homeassistant/components/sandbox/manager.py
+1 -1
View File
@@ -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__)
homeassistant/components/sandbox/messages.py
+176 -35
View File
@@ -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": <return>}``.
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: <raw strings.json dict>}}`` — the
un-flattened nesting a ``translations/<lang>.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": <main-side 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 ``<owned_domain>_*`` 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 ``<config>/.storage/sandbox/<group>/<key>`` 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",
homeassistant/components/sandbox/protocol.py
-143
View File
@@ -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": <return>}``.
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: <raw strings.json dict>}}`` — the
un-flattened nesting a ``translations/<lang>.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": <main-side 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 ``<owned_domain>_*`` 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 ``<config>/.storage/sandbox/<group>/<key>`` 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",
]
homeassistant/components/sandbox/proxy_flow.py
+11 -4
View File
@@ -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)
homeassistant/components/sandbox/router.py
+1 -2
View File
@@ -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
homeassistant/components/sandbox/translation.py
+1 -2
View File
@@ -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:
homeassistant/config_entries.py
+1 -6
View File
@@ -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),
):
sandbox/ARCHITECTURE.md
+5 -4
View File
@@ -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).
---
sandbox/hass_client/hass_client/channel.py
+5 -24
View File
@@ -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:
sandbox/hass_client/hass_client/codec_protobuf.py
+6 -2
View File
@@ -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(
sandbox/hass_client/hass_client/entity_bridge.py
+7 -2
View File
@@ -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__)
sandbox/hass_client/hass_client/entry_runner.py
+4 -3
View File
@@ -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
sandbox/hass_client/hass_client/event_mirror.py
+8 -7
View File
@@ -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__)
sandbox/hass_client/hass_client/flow_runner.py
+10 -4
View File
@@ -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.
sandbox/hass_client/hass_client/messages.py
+176 -35
View File
@@ -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": <return>}``.
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: <raw strings.json dict>}}`` — the
un-flattened nesting a ``translations/<lang>.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": <main-side 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 ``<owned_domain>_*`` 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 ``<config>/.storage/sandbox/<group>/<key>`` 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",
sandbox/hass_client/hass_client/protocol.py
-143
View File
@@ -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": <return>}``.
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: <raw strings.json dict>}}`` — the
un-flattened nesting a ``translations/<lang>.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": <main-side 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 ``<owned_domain>_*`` 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 ``<config>/.storage/sandbox/<group>/<key>`` 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",
]
sandbox/hass_client/hass_client/sandbox/__init__.py
+8 -3
View File
@@ -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
sandbox/hass_client/hass_client/sandbox_bridge.py
+1 -2
View File
@@ -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__)
sandbox/hass_client/hass_client/service_mirror.py
+15 -32
View File
@@ -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)
sandbox/hass_client/hass_client/sources.py
+34 -19
View File
@@ -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
sandbox/hass_client/hass_client/testing/pytest_plugin.py
+2 -3
View File
@@ -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()
sandbox/hass_client/tests/test_sandbox_runtime.py
+1 -1
View File
@@ -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
sandbox/hass_client/tests/test_shutdown.py
+6 -2
View File
@@ -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
sandbox/hass_client/tests/test_sources.py
+52 -7
View File
@@ -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()
sandbox/proto/check_mirror_drift.sh
+13 -5
View File
@@ -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/<file> (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
tests/components/sandbox/test_channel.py
+5 -9
View File
@@ -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()
tests/components/sandbox/test_device_registry.py
+8 -9
View File
@@ -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()
tests/components/sandbox/test_init.py
-1
View File
@@ -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 == {}
tests/components/sandbox/test_testing_plugins.py
+1 -2
View File
@@ -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"
tests/components/sandbox/test_translation.py
+1 -2
View File
@@ -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
tests/test_config_entries.py
+1 -27
View File
@@ -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"