sandbox: inline push dispatch + single-writer coalescing state queue

Channel (both mirrors): register_push_inline() runs a sync push handler
directly in the read loop — no task, no semaphore, nothing queued (so it
also bypasses the max_queued shed). Per-entity in-order delivery becomes
a guarantee instead of an accident of FIFO task scheduling. Calls stay
on the async task path (they must write replies); the overload shed now
logs dropped pushes instead of silently returning. Main's
_handle_state_changed and _handle_fire_event have no awaits — both are
now sync callbacks registered inline. Measured push throughput 105k ->
156k msg/s over a socketpair (and ~55k -> 156k cumulative with the
orjson wire change).

EntityBridge: the task-per-state-change model (asyncio.create_task per
EVENT_STATE_CHANGED) is replaced by a per-entity latest-state slot
drained by one writer task. Rapid bursts coalesce to the newest state
(test: 5 events -> 1 push), removal/register races are ordered by
construction (a _writing marker keeps a removal-during-register from
being misclassified), and the last update in a burst can no longer be
shed into a permanently stale proxy. Deletes the _removed_while_pending
/ _state_differs reconciliation machinery this replaces (deferred
plan-review-simplification Phase 6, now justified by measurements).

Also (plan §E5): entities whose _describe fails are skipped stickily
(cleared on their entity-registry update) instead of re-attempting a
register on every state write, and the unregister path pops _last_hash.

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

homeassistant/components/sandbox/bridge.py

@@ -241,10 +241,13 @@ class SandboxBridge:
 
         channel.register(MSG_REGISTER_ENTITY, self._handle_register_entity)
         channel.register(MSG_UNREGISTER_ENTITY, self._handle_unregister_entity)
-        channel.register(MSG_STATE_CHANGED, self._handle_state_changed)
+        # State pushes + event re-fires are pure sync work (dict updates +
+        # a bus fire) — dispatch them inline in the read loop so per-entity
+        # ordering is guaranteed and no task is spawned per frame.
+        channel.register_push_inline(MSG_STATE_CHANGED, self._handle_state_changed)
         channel.register(MSG_REGISTER_SERVICE, self._handle_register_service)
         channel.register(MSG_UNREGISTER_SERVICE, self._handle_unregister_service)
-        channel.register(MSG_FIRE_EVENT, self._handle_fire_event)
+        channel.register_push_inline(MSG_FIRE_EVENT, self._handle_fire_event)
         channel.register(MSG_STORE_LOAD, self._handle_store_load)
         channel.register(MSG_STORE_SAVE, self._handle_store_save)
         channel.register(MSG_STORE_REMOVE, self._handle_store_remove)
@@ -569,7 +572,8 @@ class SandboxBridge:
             await component.async_remove_entity(entity_id)
         return pb.UnregisterEntityResult(ok=True)
 
-    async def _handle_state_changed(self, msg: pb.StateChanged) -> None:
+    @callback
+    def _handle_state_changed(self, msg: pb.StateChanged) -> None:
         proxy = self._entities.get(msg.sandbox_entity_id)
         if proxy is None:
             return
@@ -714,7 +718,8 @@ class SandboxBridge:
             event_type.startswith(f"{domain}_") for domain in self._owned_domains()
         )
 
-    async def _handle_fire_event(self, msg: pb.FireEvent) -> None:
+    @callback
+    def _handle_fire_event(self, msg: pb.FireEvent) -> None:
         """Re-fire a sandbox-side event on main's bus.
 
         The sandbox tags every push with ``event_type`` + ``event_data`` and,

homeassistant/components/sandbox/channel.py

@@ -33,12 +33,24 @@ class runs in the HA Core integration and inside the sandbox subprocess
 sync by the protocol shape rather than a shared import — the integration
 must not depend on ``hass_client``).
 
-Inbound calls and pushes are dispatched in their own tasks so a handler
-that itself issues :meth:`Channel.call` does not block the reader — the
-reply for the nested call has to come back through the same reader. A
-bounded semaphore caps how many handlers can run concurrently; the N+1th
-inbound message queues at the semaphore (not at the reader) until a slot
-frees up.
+Inbound dispatch has two modes:
+
+* **Task dispatch** (:meth:`Channel.register`, async handlers) — calls and
+  pushes run in their own tasks so a handler that itself issues
+  :meth:`Channel.call` does not block the reader — the reply for the
+  nested call has to come back through the same reader. A bounded
+  semaphore caps how many handlers can run concurrently; the N+1th
+  inbound message queues at the semaphore (not at the reader) until a
+  slot frees up.
+* **Inline push dispatch** (:meth:`Channel.register_push_inline`, sync
+  handlers) — the handler runs directly in the read loop, no task, no
+  semaphore. For pure-synchronous work (state pushes are dict updates)
+  this removes a task spawn per frame and makes in-order delivery a
+  *guarantee* instead of an accident of FIFO task scheduling. Inline
+  handlers queue nothing, so they also bypass the ``max_queued`` shed.
+  Calls can never be inline (they must write replies): the call branch
+  consults only the async handlers, so a call to an inline-only type is
+  rejected with ``ChannelUnknownType``.
 """
 
 # This module is hand-mirrored: a byte-identical copy lives at both
@@ -63,6 +75,7 @@ import voluptuous as vol
 _LOGGER = logging.getLogger(__name__)
 
 Handler = Callable[[Any], Awaitable[Any]]
+SyncHandler = Callable[[Any], None]
 
 DEFAULT_MAX_INFLIGHT = 16
 
@@ -318,6 +331,7 @@ class Channel:
         self._next_id = 1
         self._pending: dict[int, asyncio.Future[Any]] = {}
         self._handlers: dict[str, Handler] = {}
+        self._inline_push_handlers: dict[str, SyncHandler] = {}
         self._reader_task: asyncio.Task[None] | None = None
         self._closed: bool = False
         self._close_done: bool = False
@@ -353,6 +367,19 @@ class Channel:
         """Register an async handler for inbound calls of this type."""
         self._handlers[msg_type] = handler
 
+    def register_push_inline(self, msg_type: str, handler: SyncHandler) -> None:
+        """Register a sync handler run inline in the read loop for pushes.
+
+        The handler executes directly in ``_dispatch`` — no task, no
+        semaphore — so it must be purely synchronous (no awaits, no
+        blocking I/O). Frames of this type are delivered strictly in wire
+        order and never shed under the ``max_queued`` cap (they queue
+        nothing). Only PUSH frames dispatch here: the call branch consults
+        only :meth:`register` handlers, so a CALL of this type errors back
+        with ``ChannelUnknownType``.
+        """
+        self._inline_push_handlers[msg_type] = handler
+
     def start(self) -> None:
         """Begin reading messages off the wire."""
         if self._reader_task is not None:
@@ -490,7 +517,11 @@ class Channel:
                     task.cancel()
 
     def _dispatch(self, frame: Frame) -> None:
-        """Route an inbound frame; non-blocking — handlers run in tasks."""
+        """Route an inbound frame.
+
+        Non-blocking: async handlers run in tasks; inline push handlers are
+        pure-sync and run right here.
+        """
         if frame.kind is FrameKind.RESPONSE:
             # Response to a call we sent out — set the future inline; no I/O.
             future = self._pending.get(frame.id)
@@ -508,10 +539,27 @@ class Channel:
                 )
             return
 
-        # Backpressure: responses are handled inline above and never shed.
-        # Bound the inbound handler tasks (each pins a decoded payload) so a
-        # frame-flood throttles here instead of growing memory without bound —
-        # reject calls with an error frame, silently drop pushes.
+        if (
+            frame.kind is FrameKind.PUSH
+            and (inline := self._inline_push_handlers.get(frame.type)) is not None
+        ):
+            # Inline push: run synchronously in the read loop — no task, no
+            # semaphore, nothing queued, so the max_queued shed below does
+            # not apply. Guarantees in-order delivery per frame type.
+            try:
+                inline(frame.payload)
+            except Exception:
+                _LOGGER.exception(
+                    "Channel %s: push handler for %s raised",
+                    self._name,
+                    frame.type,
+                )
+            return
+
+        # Backpressure: responses and inline pushes are handled above and
+        # never shed. Bound the inbound handler tasks (each pins a decoded
+        # payload) so a frame-flood throttles here instead of growing memory
+        # without bound — reject calls with an error frame, drop pushes.
         if len(self._inflight) >= self._max_queued:
             if frame.kind is FrameKind.CALL:
                 self._spawn_handler(
@@ -524,6 +572,12 @@ class Channel:
                         )
                     )
                 )
+                return
+            _LOGGER.warning(
+                "Channel %s: overloaded, dropping push frame %s",
+                self._name,
+                frame.type,
+            )
             return
 
         handler = self._handlers.get(frame.type)
@@ -632,6 +686,7 @@ __all__ = [
     "FrameTooLargeError",
     "Handler",
     "StreamTransport",
+    "SyncHandler",
     "Transport",
     "error_data_for",
 ]

sandbox/hass_client/hass_client/channel.py

@@ -33,12 +33,24 @@ class runs in the HA Core integration and inside the sandbox subprocess
 sync by the protocol shape rather than a shared import — the integration
 must not depend on ``hass_client``).
 
-Inbound calls and pushes are dispatched in their own tasks so a handler
-that itself issues :meth:`Channel.call` does not block the reader — the
-reply for the nested call has to come back through the same reader. A
-bounded semaphore caps how many handlers can run concurrently; the N+1th
-inbound message queues at the semaphore (not at the reader) until a slot
-frees up.
+Inbound dispatch has two modes:
+
+* **Task dispatch** (:meth:`Channel.register`, async handlers) — calls and
+  pushes run in their own tasks so a handler that itself issues
+  :meth:`Channel.call` does not block the reader — the reply for the
+  nested call has to come back through the same reader. A bounded
+  semaphore caps how many handlers can run concurrently; the N+1th
+  inbound message queues at the semaphore (not at the reader) until a
+  slot frees up.
+* **Inline push dispatch** (:meth:`Channel.register_push_inline`, sync
+  handlers) — the handler runs directly in the read loop, no task, no
+  semaphore. For pure-synchronous work (state pushes are dict updates)
+  this removes a task spawn per frame and makes in-order delivery a
+  *guarantee* instead of an accident of FIFO task scheduling. Inline
+  handlers queue nothing, so they also bypass the ``max_queued`` shed.
+  Calls can never be inline (they must write replies): the call branch
+  consults only the async handlers, so a call to an inline-only type is
+  rejected with ``ChannelUnknownType``.
 """
 
 # This module is hand-mirrored: a byte-identical copy lives at both
@@ -63,6 +75,7 @@ import voluptuous as vol
 _LOGGER = logging.getLogger(__name__)
 
 Handler = Callable[[Any], Awaitable[Any]]
+SyncHandler = Callable[[Any], None]
 
 DEFAULT_MAX_INFLIGHT = 16
 
@@ -318,6 +331,7 @@ class Channel:
         self._next_id = 1
         self._pending: dict[int, asyncio.Future[Any]] = {}
         self._handlers: dict[str, Handler] = {}
+        self._inline_push_handlers: dict[str, SyncHandler] = {}
         self._reader_task: asyncio.Task[None] | None = None
         self._closed: bool = False
         self._close_done: bool = False
@@ -353,6 +367,19 @@ class Channel:
         """Register an async handler for inbound calls of this type."""
         self._handlers[msg_type] = handler
 
+    def register_push_inline(self, msg_type: str, handler: SyncHandler) -> None:
+        """Register a sync handler run inline in the read loop for pushes.
+
+        The handler executes directly in ``_dispatch`` — no task, no
+        semaphore — so it must be purely synchronous (no awaits, no
+        blocking I/O). Frames of this type are delivered strictly in wire
+        order and never shed under the ``max_queued`` cap (they queue
+        nothing). Only PUSH frames dispatch here: the call branch consults
+        only :meth:`register` handlers, so a CALL of this type errors back
+        with ``ChannelUnknownType``.
+        """
+        self._inline_push_handlers[msg_type] = handler
+
     def start(self) -> None:
         """Begin reading messages off the wire."""
         if self._reader_task is not None:
@@ -490,7 +517,11 @@ class Channel:
                     task.cancel()
 
     def _dispatch(self, frame: Frame) -> None:
-        """Route an inbound frame; non-blocking — handlers run in tasks."""
+        """Route an inbound frame.
+
+        Non-blocking: async handlers run in tasks; inline push handlers are
+        pure-sync and run right here.
+        """
         if frame.kind is FrameKind.RESPONSE:
             # Response to a call we sent out — set the future inline; no I/O.
             future = self._pending.get(frame.id)
@@ -508,10 +539,27 @@ class Channel:
                 )
             return
 
-        # Backpressure: responses are handled inline above and never shed.
-        # Bound the inbound handler tasks (each pins a decoded payload) so a
-        # frame-flood throttles here instead of growing memory without bound —
-        # reject calls with an error frame, silently drop pushes.
+        if (
+            frame.kind is FrameKind.PUSH
+            and (inline := self._inline_push_handlers.get(frame.type)) is not None
+        ):
+            # Inline push: run synchronously in the read loop — no task, no
+            # semaphore, nothing queued, so the max_queued shed below does
+            # not apply. Guarantees in-order delivery per frame type.
+            try:
+                inline(frame.payload)
+            except Exception:
+                _LOGGER.exception(
+                    "Channel %s: push handler for %s raised",
+                    self._name,
+                    frame.type,
+                )
+            return
+
+        # Backpressure: responses and inline pushes are handled above and
+        # never shed. Bound the inbound handler tasks (each pins a decoded
+        # payload) so a frame-flood throttles here instead of growing memory
+        # without bound — reject calls with an error frame, drop pushes.
         if len(self._inflight) >= self._max_queued:
             if frame.kind is FrameKind.CALL:
                 self._spawn_handler(
@@ -524,6 +572,12 @@ class Channel:
                         )
                     )
                 )
+                return
+            _LOGGER.warning(
+                "Channel %s: overloaded, dropping push frame %s",
+                self._name,
+                frame.type,
+            )
             return
 
         handler = self._handlers.get(frame.type)
@@ -632,6 +686,7 @@ __all__ = [
     "FrameTooLargeError",
     "Handler",
     "StreamTransport",
+    "SyncHandler",
     "Transport",
     "error_data_for",
 ]

sandbox/hass_client/hass_client/entity_bridge.py

@@ -5,6 +5,14 @@ The bridge listens for ``EVENT_STATE_CHANGED`` on the sandbox-private
 trigger a ``sandbox/register_entity`` call up to main; subsequent
 changes become ``sandbox/state_changed`` pushes.
 
+State flow is a per-entity latest-state slot drained by one writer task:
+the event listener only classifies and writes ``_pending[entity_id]``
+(latest write wins, so a burst for one entity coalesces to a single
+push), and the writer drains slots in insertion order, one RPC at a
+time. That makes per-entity ordering a structural guarantee (no
+task-per-event racing) and means overload coalesces instead of dropping
+the last update in a burst.
+
 We deliberately tag every event with the sandbox-side ``entry_id`` of
 the owning :class:`EntityPlatform` so main can route each proxy entity
 to the right :class:`ConfigEntry`. Entities that aren't owned by a
@@ -13,6 +21,7 @@ integration creates outside its own entry) are skipped with a debug log.
 """
 
 import asyncio
+import contextlib
 import json
 import logging
 from typing import Any
@@ -34,6 +43,9 @@ from .protocol import MSG_REGISTER_ENTITY, MSG_STATE_CHANGED, MSG_UNREGISTER_ENT
 
 _LOGGER = logging.getLogger(__name__)
 
+# Slot value marking a pending removal for an entity.
+_REMOVED = object()
+
 
 class EntityBridge:
     """Forwards sandbox-side entity lifecycle events up to main.
@@ -59,12 +71,21 @@ class EntityBridge:
         self.approved = approved if approved is not None else ApprovedDomains()
         self._channel: Channel | None = None
         self._registered: set[str] = set()
-        self._pending: set[str] = set()
-        # Entities whose removal arrived while their register RPC was still
-        # in flight. The register task flushes (unregisters) them once it
-        # completes — relying on `_registered` membership would miss the
-        # removal because the entity isn't registered yet at removal time.
-        self._removed_while_pending: set[str] = set()
+        # Per-entity latest-state slot, drained in insertion order by the
+        # writer task. Values are a State (latest write wins — a burst for
+        # one entity coalesces to a single push) or ``_REMOVED``.
+        self._pending: dict[str, Any] = {}
+        # Entity the writer is currently awaiting an RPC for. Its slot has
+        # been popped, so ``_pending``/``_registered`` membership alone would
+        # misclassify a removal arriving mid-RPC as never-seen.
+        self._writing: str | None = None
+        self._wake = asyncio.Event()
+        self._writer_task: asyncio.Task[None] | None = None
+        # Entities whose ``_describe`` failed (no live entity object / no
+        # owning entry). Ignored on every state write instead of re-attempting
+        # a full register each time; cleared on an entity-registry update for
+        # that entity_id.
+        self._skipped: set[str] = set()
         # Domain each registered entity contributed to ApprovedDomains, so its
         # approval refcount can be released symmetrically on unregister.
         self._approved_domain: dict[str, str] = {}
@@ -79,6 +100,9 @@ class EntityBridge:
     def register(self, channel: Channel) -> None:
         """Subscribe to state + registry events and capture the channel."""
         self._channel = channel
+        self._writer_task = asyncio.create_task(
+            self._writer_loop(), name="sandbox:entity-bridge-writer"
+        )
         self._unsub_state = self.hass.bus.async_listen(
             EVENT_STATE_CHANGED, self._on_state_changed
         )
@@ -93,7 +117,7 @@ class EntityBridge:
         )
 
     async def async_stop(self) -> None:
-        """Detach the state + registry listeners."""
+        """Detach the listeners and stop the writer task."""
         for attr in (
             "_unsub_state",
             "_unsub_entity_registry",
@@ -103,50 +127,94 @@ class EntityBridge:
             if unsub is not None:
                 unsub()
                 setattr(self, attr, None)
+        if self._writer_task is not None:
+            self._writer_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._writer_task
+            self._writer_task = None
 
     @callback
     def _on_state_changed(self, event: Event[EventStateChangedData]) -> None:
+        """Classify the event into the entity's slot and wake the writer.
+
+        Pure sync bookkeeping — all RPC work happens in the writer task.
+        """
         if self._channel is None or self._channel.closed:
             return
         entity_id: str = event.data["entity_id"]
+        if entity_id in self._skipped:
+            return
         new_state = event.data.get("new_state")
 
         if new_state is None:
+            # Only meaningful for entities we track or are about to: a
+            # removal for a never-seen entity has nothing to unregister.
+            if (
+                entity_id not in self._registered
+                and entity_id not in self._pending
+                and entity_id != self._writing
+            ):
+                return
+            self._pending[entity_id] = _REMOVED
+        else:
+            # Latest write wins. This may also overwrite a pending _REMOVED:
+            # main's register is an upsert, so skipping the removal blip and
+            # jumping straight to the newer state is safe.
+            self._pending[entity_id] = new_state
+        self._wake.set()
+
+    async def _writer_loop(self) -> None:
+        """Drain ``_pending`` slots in insertion order, one RPC at a time.
+
+        The slot is popped *before* awaiting, so updates arriving during the
+        RPC land in a fresh slot and get a later round — including a state
+        change or removal racing an in-flight register. One failing RPC is
+        logged and must not kill the loop.
+        """
+        while True:
+            await self._wake.wait()
+            self._wake.clear()
+            while self._pending:
+                entity_id = next(iter(self._pending))
+                slot = self._pending.pop(entity_id)
+                self._writing = entity_id
+                try:
+                    await self._process_slot(entity_id, slot)
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    _LOGGER.exception(
+                        "EntityBridge: writer failed for %s", entity_id
+                    )
+                finally:
+                    self._writing = None
+
+    async def _process_slot(self, entity_id: str, slot: Any) -> None:
+        """Ship one drained slot to main."""
+        if slot is _REMOVED:
             if entity_id in self._registered:
                 self._registered.discard(entity_id)
+                self._last_hash.pop(entity_id, None)
                 self._release_approval(entity_id)
-                asyncio.create_task(  # noqa: RUF006
-                    self._push_unregister(entity_id),
-                    name=f"sandbox:unregister:{entity_id}",
-                )
-            elif entity_id in self._pending:
-                # Removed mid-register: the register task hasn't added it to
-                # _registered yet, so flag it and let that task unregister.
-                self._removed_while_pending.add(entity_id)
+                await self._push_unregister(entity_id)
             return
-
         if entity_id in self._registered:
-            asyncio.create_task(  # noqa: RUF006
-                self._push_state(entity_id, new_state),
-                name=f"sandbox:state:{entity_id}",
-            )
+            await self._push_state(entity_id, slot)
             return
-
-        if entity_id in self._pending:
-            return
-        self._pending.add(entity_id)
-        asyncio.create_task(  # noqa: RUF006
-            self._register_and_push(entity_id, new_state),
-            name=f"sandbox:register:{entity_id}",
-        )
+        await self._register(entity_id, slot)
+        # No post-register flush needed: any state change during the register
+        # RPC landed in a fresh _pending slot and gets its own round.
 
     @callback
     def _on_entity_registry_updated(self, event: Event[Any]) -> None:
         if self._channel is None or self._channel.closed:
             return
+        entity_id: str = event.data["entity_id"]
+        # A registry entry appearing/changing may make a previously
+        # undescribable entity describable — let its next state write retry.
+        self._skipped.discard(entity_id)
         if event.data.get("action") != "update":
             return
-        entity_id: str = event.data["entity_id"]
         if entity_id not in self._registered:
             return
         asyncio.create_task(  # noqa: RUF006
@@ -173,41 +241,6 @@ class EntityBridge:
                 name=f"sandbox:resend:{entity_id}",
             )
 
-    async def _register_and_push(self, entity_id: str, new_state: Any) -> None:
-        try:
-            await self._register(entity_id, new_state)
-        finally:
-            self._pending.discard(entity_id)
-
-        # While the register RPC was in flight, _on_state_changed dropped any
-        # further state_changed for this entity (it was neither registered nor
-        # re-queued). Reconcile that coalesced gap now.
-        #
-        # NOTE: this is the *correctness* fix. Plan 5 (simplification) builds a
-        # single-writer queue on top of the entity push path; when it lands it
-        # should subsume this flush into the queue's ordering guarantees.
-        if entity_id in self._removed_while_pending:
-            # A removal raced the register. Now that register has completed
-            # (and added the entity to _registered), unregister it so main
-            # doesn't keep a ghost proxy.
-            self._removed_while_pending.discard(entity_id)
-            if entity_id in self._registered:
-                self._registered.discard(entity_id)
-                self._last_hash.pop(entity_id, None)
-                self._release_approval(entity_id)
-                await self._push_unregister(entity_id)
-            return
-
-        if entity_id not in self._registered:
-            # Register failed (or was skipped); nothing to flush.
-            return
-
-        current = self.hass.states.get(entity_id)
-        if current is None:
-            return
-        if _state_differs(current, new_state):
-            await self._push_state(entity_id, current)
-
     def _describe(self, entity_id: str) -> dict[str, Any] | None:
         """Build the registry-shaped description for a live entity, or None."""
         domain = entity_id.split(".", 1)[0]
@@ -235,6 +268,10 @@ class EntityBridge:
             return
         payload = self._describe(entity_id)
         if payload is None:
+            # Sticky skip: without it every subsequent state write for this
+            # entity re-attempts the full describe + register. An
+            # entity-registry update for the entity clears the skip.
+            self._skipped.add(entity_id)
             return
         new_hash = _payload_hash(payload)
         initial_state = None
@@ -357,19 +394,6 @@ def _to_entity_description(
     )
 
 
-def _state_differs(current: Any, snapshot: Any) -> bool:
-    """Whether ``current`` state/attributes differ from the registered snapshot.
-
-    ``snapshot`` is the ``new_state`` captured when the register task was
-    created; ``current`` is the live state re-read after the register RPC
-    resolved. A difference means a state change was coalesced away during the
-    in-flight window and must be flushed.
-    """
-    snap_state = getattr(snapshot, "state", None)
-    snap_attrs = dict(snapshot.attributes) if hasattr(snapshot, "attributes") else {}
-    return current.state != snap_state or dict(current.attributes) != snap_attrs
-
-
 def _payload_hash(payload: dict[str, Any]) -> str:
     """Stable hash of a description payload's mirrored fields.
 

sandbox/hass_client/tests/test_entity_bridge.py

@@ -424,11 +424,12 @@ async def test_unregister_releases_domain_approval(
 async def test_state_update_during_register_is_flushed(
     channels: tuple[Channel, Channel], hass_with_demo_component
 ) -> None:
-    """A state change coalesced away while register is in flight is flushed.
+    """A state change arriving while register is in flight is pushed after.
 
-    The register RPC is held open; a second ``async_set`` lands in the state
-    machine but is dropped by the bridge (entity still pending). Once register
-    completes, the bridge re-reads the live state and pushes the gap.
+    The register RPC is held open; a second ``async_set`` lands in the
+    entity's pending slot (the single writer is busy awaiting the register, so
+    nothing is pushed yet). Once register completes, the writer drains the
+    slot and pushes the newer state.
     """
     main, sandbox = channels
     hass, component = hass_with_demo_component
@@ -483,13 +484,124 @@ async def test_state_update_during_register_is_flushed(
     await bridge.async_stop()
 
 
+async def test_state_burst_coalesces_to_single_push(
+    channels: tuple[Channel, Channel], hass_with_demo_component
+) -> None:
+    """A rapid burst of state changes coalesces to one push of the latest.
+
+    The writer task cannot run between the synchronous ``async_set`` calls,
+    so each write overwrites the entity's single pending slot; the writer
+    then ships one push carrying the final state instead of one per event.
+    """
+    main, sandbox = channels
+    hass, component = hass_with_demo_component
+
+    state_calls: list[pb.StateChanged] = []
+
+    async def _on_register(msg: pb.EntityDescription) -> pb.RegisterEntityResult:
+        return pb.RegisterEntityResult(entity_id="demo.lamp_main")
+
+    async def _on_state(msg: pb.StateChanged) -> None:
+        state_calls.append(msg)
+
+    main.register("sandbox/register_entity", _on_register)
+    main.register("sandbox/state_changed", _on_state)
+    main.start()
+    sandbox.start()
+
+    entity = _FakeEntity()
+    component._entities[entity.entity_id] = entity  # noqa: SLF001
+
+    bridge = EntityBridge(hass)
+    bridge.register(sandbox)
+    await _register_initial(bridge, hass, entity)
+
+    n_events = 5
+    for idx in range(n_events):
+        hass.states.async_set(entity.entity_id, f"level_{idx}", {"idx": idx})
+
+    for _ in range(50):
+        if state_calls:
+            break
+        await asyncio.sleep(0)
+    # Let everything settle so a non-coalescing bridge would have flushed
+    # every push before the count is asserted.
+    for _ in range(20):
+        await asyncio.sleep(0)
+
+    assert len(state_calls) == 1
+    assert state_calls[0].state == f"level_{n_events - 1}"
+    assert decode_json_dict(state_calls[0].attributes)["idx"] == n_events - 1
+
+    await bridge.async_stop()
+
+
+async def test_describe_failure_is_sticky_until_registry_update(
+    channels: tuple[Channel, Channel], hass_with_demo_component
+) -> None:
+    """An undescribable entity is skipped once, not re-attempted per write.
+
+    The first state write for an entity with no live entity object marks it
+    skipped; later writes never reach ``_describe`` again. An entity-registry
+    update for that entity clears the skip so the next write retries.
+    """
+    main, sandbox = channels
+    hass, component = hass_with_demo_component
+
+    register_calls: list[pb.EntityDescription] = []
+
+    async def _on_register(msg: pb.EntityDescription) -> pb.RegisterEntityResult:
+        register_calls.append(msg)
+        return pb.RegisterEntityResult(entity_id="demo.lamp_main")
+
+    main.register("sandbox/register_entity", _on_register)
+    main.start()
+    sandbox.start()
+
+    # No entity in the component: _describe returns None.
+    bridge = EntityBridge(hass)
+    bridge.register(sandbox)
+
+    hass.states.async_set("demo.lamp", "off", {})
+    for _ in range(20):
+        await asyncio.sleep(0)
+    assert "demo.lamp" in bridge._skipped  # noqa: SLF001
+
+    # Further writes are ignored — no new slot, no describe attempt.
+    hass.states.async_set("demo.lamp", "on", {})
+    for _ in range(20):
+        await asyncio.sleep(0)
+    assert register_calls == []
+
+    # The entity becomes describable and its registry entry updates.
+    entity = _FakeEntity()
+    component._entities[entity.entity_id] = entity  # noqa: SLF001
+    hass.bus.async_fire(
+        er.EVENT_ENTITY_REGISTRY_UPDATED,
+        {"action": "update", "entity_id": entity.entity_id, "changes": {}},
+    )
+    assert entity.entity_id not in bridge._skipped  # noqa: SLF001
+
+    # A *changed* state (async_set with the previous value fires no
+    # EVENT_STATE_CHANGED) now registers the entity.
+    hass.states.async_set(entity.entity_id, "dim", {})
+    for _ in range(50):
+        if register_calls:
+            break
+        await asyncio.sleep(0)
+    assert len(register_calls) == 1
+
+    await bridge.async_stop()
+
+
 async def test_removal_during_register_unregisters(
     channels: tuple[Channel, Channel], hass_with_demo_component
 ) -> None:
     """An entity removed while its register RPC is in flight is unregistered.
 
-    Without the removal-while-pending flag the removal is dropped (the entity
-    isn't in ``_registered`` yet), leaving a ghost proxy on main.
+    The removal lands as a ``_REMOVED`` slot (the writer's in-flight marker
+    keeps it from being classified as never-seen); once register completes,
+    the writer drains that slot and unregisters — no ghost proxy on main.
     """
     main, sandbox = channels
     hass, component = hass_with_demo_component

tests/components/sandbox/test_channel.py

@@ -197,6 +197,66 @@ async def test_push_message_is_one_way(channels: tuple) -> None:
     assert received == [{"hello": "world"}]
 
 
+async def test_inline_push_handler_runs_in_wire_order(channels: tuple) -> None:
+    """Inline push handlers run in the read loop, preserving wire order."""
+    channel_a, channel_b = channels
+    received: list[int] = []
+
+    def receive(payload: dict) -> None:
+        received.append(payload["idx"])
+
+    channel_b.register_push_inline("test/inline", receive)
+    for idx in range(100):
+        await channel_a.push("test/inline", {"idx": idx})
+
+    for _ in range(100):
+        if len(received) == 100:
+            break
+        await asyncio.sleep(0.01)
+    assert received == list(range(100))
+
+
+async def test_inline_push_handler_exception_keeps_reader_alive(
+    channels: tuple, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A raising inline handler is logged; the read loop keeps dispatching."""
+    channel_a, channel_b = channels
+    received: list[int] = []
+
+    def receive(payload: dict) -> None:
+        if payload["idx"] == 0:
+            raise ValueError("boom")
+        received.append(payload["idx"])
+
+    channel_b.register_push_inline("test/inline", receive)
+    await channel_a.push("test/inline", {"idx": 0})
+    await channel_a.push("test/inline", {"idx": 1})
+
+    for _ in range(100):
+        if received:
+            break
+        await asyncio.sleep(0.01)
+    assert received == [1]
+    assert "push handler for test/inline raised" in caplog.text
+
+    # Calls still round-trip on the surviving read loop.
+    async def echo(payload: dict) -> dict:
+        return {"echoed": payload["value"]}
+
+    channel_b.register("test/echo", echo)
+    assert await channel_a.call("test/echo", {"value": 5}) == {"echoed": 5}
+
+
+async def test_call_to_inline_only_type_is_unknown(channels: tuple) -> None:
+    """A CALL frame never dispatches inline — only async handlers serve calls."""
+    channel_a, channel_b = channels
+
+    channel_b.register_push_inline("test/inline", lambda payload: None)
+    with pytest.raises(ChannelRemoteError) as err:
+        await asyncio.wait_for(channel_a.call("test/inline", {}), timeout=2.0)
+    assert err.value.error_type == "ChannelUnknownType"
+
+
 async def test_handler_can_call_back_without_deadlock(channels: tuple) -> None:
     """A handler that issues channel.call mid-execution doesn't deadlock.
 
@@ -377,13 +437,16 @@ async def test_close_after_eof_still_closes_transport() -> None:
     assert transport.close_calls == 1
 
 
-async def test_read_backpressure_sheds_over_queued_cap() -> None:
+async def test_read_backpressure_sheds_over_queued_cap(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
     """A frame-flood is bounded: over the cap, calls are shed not queued.
 
     With a tiny ``max_queued`` and handlers that never return, the reader keeps
     draining the wire but stops growing handler tasks — once the cap is hit,
-    further calls come back as ``ChannelOverloaded`` instead of piling up
-    unbounded decoded payloads.
+    further calls come back as ``ChannelOverloaded``, task-dispatched pushes
+    are dropped with a warning, and inline pushes (which queue nothing) still
+    go through.
     """
     channel_a, channel_b = make_channel_pair(
         max_inflight_b=2, max_queued_b=3, use_json=True
@@ -420,6 +483,20 @@ async def test_read_backpressure_sheds_over_queued_cap() -> None:
 
         # The reader threw the excess away rather than growing the inflight set.
         assert len(channel_b._inflight) == 3
+
+        # A task-dispatched push is shed too — with a warning, not silently.
+        inline_received: list[dict] = []
+        channel_b.register_push_inline("test/inline", inline_received.append)
+        await channel_a.push("test/never", {"idx": 100})
+        # Inline pushes bypass the shed entirely (they queue nothing).
+        await channel_a.push("test/inline", {"idx": 101})
+        for _ in range(100):
+            if inline_received:
+                break
+            await asyncio.sleep(0.01)
+        assert inline_received == [{"idx": 101}]
+        assert "overloaded, dropping push frame test/never" in caplog.text
+        assert len(channel_b._inflight) == 3
     finally:
         release.set()
         for task in pending: