feat(ipc): control socket stage 3 - a state stream replaces polled cache keys (#735)

Adds state.get / state.subscribe to the display's control socket (StateHub in src/ipc/server.py). The web interface holds one subscription per process (web_interface/display_state.py) and reads current-status, on-demand status, plugin runtime and /health's display_loop from it, falling back to the cache keys and heartbeat file. While the socket serves readers, display_current_state and plugin_runtime_snapshot are written less often (about 1.5 instead of 5 cache writes a minute for 15 s screens).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/display_controller.py

@@ -55,7 +55,7 @@ from src.ipc.contract import (
     PluginReloadArgs,
     PluginReloadResult,
 )
-from src.ipc.server import ControlServer, QueuedCommand, start_control_server
+from src.ipc.server import ControlServer, QueuedCommand, StateHub, start_control_server
 from src.vegas_mode.render_pipeline import SYNC_SEND_INTERVAL
 
 # Get logger with consistent configuration
@@ -65,6 +65,13 @@ logger = get_logger(__name__)
 # treats display_current_state older than 120 s as unknown.
 CURRENT_STATE_REFRESH_SECONDS = 30
 
+# While the control socket serves the web interface's state readers
+# (StateHub.readers_active), display_current_state is only their fallback:
+# it is then rewritten at this interval and on a change of the flags, not on
+# every mode change. Below the readers' 120 s max_age, so the fallback copy
+# never reads as unknown.
+CURRENT_STATE_RELAXED_REFRESH_SECONDS = 60
+
 # How long startup will wait for plugins to fetch their first data before
 # showing anything. Each plugin's update blocks for up to the executor's 30s
 # timeout and they run one after another, so the uncapped total is the sum of
@@ -1430,18 +1437,60 @@ class DisplayController:
             return None
         return max(0.0, expires_at - time.time())
 
+    #: The control socket's state stream (src/ipc/server.StateHub), while the
+    #: socket is served. Class-level default for controllers built without
+    #: __init__ (tests) and for a display with no socket.
+    _state_hub: Optional[StateHub] = None
+
+    def _current_mode_state(self) -> Dict[str, Any]:
+        """What display_current_state and the socket's ``display`` section hold."""
+        return {
+            'mode': self.current_display_mode,
+            'plugin_id': self.mode_to_plugin_id.get(self.current_display_mode),
+            'mode_index': self.current_mode_index,
+            'total_modes': len(self.available_modes),
+            'on_demand_active': self.on_demand_active,
+            'is_display_active': self.is_display_active,
+            'last_updated': time.time(),
+        }
+
+    def _push_live_state(self, display_state: Optional[Dict[str, Any]] = None) -> None:
+        """Hand the current mode and the brightness to the control socket's
+        state stream. In memory, no disk: the hub only bumps its version (and
+        wakes subscribers) when something other than ``last_updated`` changed.
+
+        Called on every pass of the publish points below, so
+        ``display.last_updated`` doubles as the render thread's proof of life
+        for the socket's readers, as the cache key's max_age does today.
+        """
+        hub = self._state_hub
+        if hub is None:
+            return
+        try:
+            hub.publish('display', display_state or self._current_mode_state(),
+                        volatile=('last_updated',))
+            hub.publish('brightness', {
+                'brightness': getattr(self, '_normal_brightness', None),
+                'panel_brightness': getattr(self, 'current_brightness', None),
+                'dimmed': bool(getattr(self, 'is_dimmed', False)),
+            })
+        except Exception as err:  # pylint: disable=broad-except
+            logger.debug("Could not publish the display state to the control socket: %s",
+                         err, exc_info=True)
+
+    def _state_readers_on_socket(self) -> bool:
+        """Is the control socket serving the web interface's state readers?"""
+        hub = self._state_hub
+        try:
+            return bool(hub is not None and hub.readers_active())
+        except Exception:  # pylint: disable=broad-except
+            return False
+
     def _publish_current_mode_state(self) -> None:
         """Publish the currently active display mode/plugin to cache for the web UI."""
         try:
-            state = {
-                'mode': self.current_display_mode,
-                'plugin_id': self.mode_to_plugin_id.get(self.current_display_mode),
-                'mode_index': self.current_mode_index,
-                'total_modes': len(self.available_modes),
-                'on_demand_active': self.on_demand_active,
-                'is_display_active': self.is_display_active,
-                'last_updated': time.time(),
-            }
+            state = self._current_mode_state()
+            self._push_live_state(state)
             self.cache_manager.set('display_current_state', state)
             self._last_published_mode = self.current_display_mode
             self._last_published_flags = self._current_state_flags()
@@ -1467,11 +1516,23 @@ class DisplayController:
         priority, a single enabled plugin -- has to be republished or the UI
         reports it as unknown. Otherwise this writes only on a change, not on
         every render tick.
+
+        While the control socket serves the web interface's state readers,
+        the socket's in-memory copy is updated on every call and the cache
+        key is only their fallback: a mode change alone is then written at
+        the relaxed refresh (CURRENT_STATE_RELAXED_REFRESH_SECONDS), still
+        inside the readers' max_age. The flags are still written at once.
+        When the socket stops serving them, the next call writes a changed
+        mode again.
         """
-        if (self.current_display_mode != self._last_published_mode
+        relaxed = self._state_readers_on_socket()
+        refresh = CURRENT_STATE_RELAXED_REFRESH_SECONDS if relaxed else CURRENT_STATE_REFRESH_SECONDS
+        if ((not relaxed and self.current_display_mode != self._last_published_mode)
                 or self._current_state_flags() != getattr(self, '_last_published_flags', None)
-                or time.monotonic() - self._last_published_at >= CURRENT_STATE_REFRESH_SECONDS):
-            self._publish_current_mode_state()
+                or time.monotonic() - self._last_published_at >= refresh):
+            self._publish_current_mode_state()   # pushes to the socket as well
+        else:
+            self._push_live_state()
 
     def _on_demand_state(self) -> Dict[str, Any]:
         """The on-demand state as published to the cache and the control socket."""
@@ -1494,6 +1555,11 @@ class DisplayController:
         """Publish current on-demand state to cache for external consumers."""
         try:
             state = self._on_demand_state()
+            hub = self._state_hub
+            if hub is not None:
+                # In memory, first: a subscriber hears the outcome of an
+                # on-demand command even if the cache write below fails.
+                hub.publish('on_demand', state, volatile=('last_updated', 'remaining'))
             self.cache_manager.set('display_on_demand_state', state)
         except (OSError, RuntimeError, ValueError, TypeError) as err:
             logger.error("Failed to publish on-demand state: %s", err, exc_info=True)
@@ -1738,12 +1804,31 @@ class DisplayController:
         """
         if self._control_server is not None:
             return
+        hub = StateHub(loop_probe=display_watchdog.watchdog.liveness)
         try:
             self._control_server = start_control_server(
                 status_provider=self._control_status,
-                cache_dir=getattr(self.cache_manager, 'cache_dir', None))
+                cache_dir=getattr(self.cache_manager, 'cache_dir', None),
+                state_hub=hub)
         except Exception:  # pylint: disable=broad-except
             logger.exception("Control socket not started; using the file mailbox only")
+        if self._control_server is not None:
+            self._start_state_stream(hub)
+
+    def _start_state_stream(self, hub: StateHub) -> None:
+        """Start publishing to the socket's state stream (``state.get`` and
+        ``state.subscribe``): everything a reader would see, now, then on
+        every publish. Never raises; without it readers use the cache keys."""
+        try:
+            self._state_hub = hub
+            self._push_live_state()
+            hub.publish('on_demand', self._on_demand_state(),
+                        volatile=('last_updated', 'remaining'))
+            publisher = getattr(self, '_plugin_runtime_publisher', None)
+            if publisher is not None:
+                publisher.attach_hub(hub)
+        except Exception:  # pylint: disable=broad-except
+            logger.exception("Control socket state stream not started; readers use the cache")
 
     def _control_status(self) -> Dict[str, Any]:
         """The socket's on_demand.status answer. Runs on the socket's thread: reads only."""
@@ -4532,6 +4617,7 @@ class DisplayController:
             except Exception as e:
                 logger.warning("Error closing the control socket: %s", e)
             self._control_server = None
+            self._state_hub = None
         # Stop the async update worker first so no in-flight update() call
         # is still touching display/cache-backed resources while they're
         # torn down below.

src/display_watchdog.py

@@ -216,6 +216,23 @@ class RenderWatchdog:
     def armed(self) -> bool:
         return self._armed
 
+    def liveness(self) -> Dict[str, Any]:
+        """The heartbeat, in memory: what the control socket's state stream
+        reports as ``loop``.
+
+        ``heartbeat_age_seconds`` is the age of the render thread's last beat,
+        the beat that writes the heartbeat file, so it ages at the same rate
+        and is judged by the same ``HEARTBEAT_STALE_SECONDS``. None until the
+        loop has drawn its first frame. Any thread may call this: it only
+        reads two attributes.
+        """
+        last = self._last_beat
+        age = None
+        if self._armed and last is not None:
+            age = max(self._clock() - last, 0.0)
+        return {'heartbeat_age_seconds': age, 'armed': self._armed,
+                'stale_after': HEARTBEAT_STALE_SECONDS}
+
     def _on_render_thread(self) -> bool:
         return self._render_thread is not None and threading.get_ident() == self._render_thread
 

src/ipc/client.py

@@ -10,20 +10,24 @@ blocks for longer than ``timeout`` in total.
 from __future__ import annotations
 
 import socket
+import threading
 import time
 import uuid
-from typing import Any, Dict, List, Mapping, Optional, Sequence
+from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence
 
 from src.ipc.contract import (
     AWAIT_SECONDS,
     MAX_MESSAGE_BYTES,
     PROTOCOL_VERSION,
+    SUBSCRIBE_KEEPALIVE_SECONDS,
     SUPPORTED_VERSIONS,
     Command,
     FrameReader,
     ProtocolError,
     Request,
     Response,
+    StateEvent,
+    StateEventKind,
     client_socket_paths,
     decode_message,
     encode_message,
@@ -233,3 +237,227 @@ def hello(client: str = 'web', *, timeout: float = DEFAULT_TIMEOUT_SECONDS,
     """Version negotiation: the result's ``version`` is the one both sides speak."""
     return request(Command.HELLO, {'versions': list(SUPPORTED_VERSIONS), 'client': client},
                    timeout=timeout, paths=paths)
+
+
+# -- the state stream (stage 3) ---------------------------------------------------------
+
+def state_get(since: Optional[int] = None, epoch: Optional[str] = None, *,
+              timeout: float = DEFAULT_TIMEOUT_SECONDS,
+              paths: Optional[Sequence[str]] = None) -> Dict[str, Any]:
+    """The display's state now, as a :class:`~src.ipc.contract.StateSnapshot`.
+
+    With ``since``/``epoch`` from an earlier answer, an unchanged state comes
+    back in the short ``changed: false`` form. Raises :class:`ControlError`
+    (``unknown_command`` from a display older than stage 3).
+    """
+    args: Dict[str, Any] = {}
+    if since is not None:
+        args['since'] = since
+    if epoch is not None:
+        args['epoch'] = epoch
+    return request(Command.STATE_GET, args, timeout=timeout, paths=paths)
+
+
+def snapshot_age(snapshot: Mapping[str, Any], now_mono: Optional[float] = None) -> float:
+    """Seconds since ``snapshot`` arrived: ``received_mono`` (set by
+    :meth:`StateSubscription.latest`) to now; 0 for a one-shot answer."""
+    received = snapshot.get('received_mono')
+    if isinstance(received, (int, float)) and not isinstance(received, bool):
+        now_mono = time.monotonic() if now_mono is None else now_mono
+        return max(now_mono - float(received), 0.0)
+    return 0.0
+
+
+def snapshot_loop_age(snapshot: Mapping[str, Any],
+                      now_mono: Optional[float] = None) -> Optional[float]:
+    """The render loop's heartbeat age now, from a state snapshot: the age the
+    display measured when it answered, plus the time since the answer
+    arrived. None when the display has no beat to report yet."""
+    loop = snapshot.get('loop')
+    if not isinstance(loop, dict):
+        state = snapshot.get('state')
+        loop = state.get('loop') if isinstance(state, dict) else None
+    age = loop.get('heartbeat_age_seconds') if isinstance(loop, dict) else None
+    if not isinstance(age, (int, float)) or isinstance(age, bool):
+        return None
+    return max(float(age), 0.0) + snapshot_age(snapshot, now_mono)
+
+
+#: A subscription that has heard nothing for this long is not trusted: the
+#: display sends a tick at least every SUBSCRIBE_KEEPALIVE_SECONDS.
+SUBSCRIPTION_SILENCE_SECONDS = 3 * SUBSCRIBE_KEEPALIVE_SECONDS
+
+#: Reconnect backoff: the first retry, and the cap. A display that does not
+#: know state.subscribe (stage 2 or older) is retried at the cap.
+_RECONNECT_MIN_SECONDS = 1.0
+_RECONNECT_MAX_SECONDS = 30.0
+
+#: Failures that another try soon will not fix.
+_SLOW_RETRY_REASONS = frozenset({'unknown_command', 'unsupported_version', 'disabled',
+                                 'unsupported'})
+
+
+class StateSubscription:
+    """One ``state.subscribe`` connection, held on a daemon thread.
+
+    Keeps the latest snapshot the display pushed, so a reader answers from
+    memory (:meth:`latest`). Reconnects with a backoff when the display goes
+    away. Never raises into the caller: :meth:`latest` is None whenever the
+    copy cannot be vouched for (not connected, or silent for longer than
+    ``silence``), and the caller falls back.
+    """
+
+    def __init__(self, paths: Optional[Sequence[str]] = None, *,
+                 silence: float = SUBSCRIPTION_SILENCE_SECONDS,
+                 connect_timeout: float = DEFAULT_TIMEOUT_SECONDS,
+                 clock: Callable[[], float] = time.monotonic):
+        self._paths = list(paths) if paths is not None else None
+        self._silence = silence
+        self._connect_timeout = connect_timeout
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._snapshot: Optional[Dict[str, Any]] = None
+        self._received: Optional[float] = None
+        self._connected = False
+        self._stop = threading.Event()
+        self._sock: Optional[socket.socket] = None
+        self._thread: Optional[threading.Thread] = None
+        #: The reason the last connection ended (a ControlError reason).
+        self.last_error: Optional[str] = None
+        #: Full snapshots received: the subscribe answer and each state event.
+        self.snapshots = 0
+
+    # -- the reader's side ---------------------------------------------------
+
+    @property
+    def connected(self) -> bool:
+        return self._connected
+
+    def latest(self) -> Optional[Dict[str, Any]]:
+        """A copy of the latest snapshot, with ``received_mono`` (this
+        process's monotonic clock when it arrived); None when not trusted."""
+        with self._lock:
+            if not self._connected or self._snapshot is None or self._received is None:
+                return None
+            if self._clock() - self._received > self._silence:
+                return None
+            snap = dict(self._snapshot)
+            snap['received_mono'] = self._received
+            return snap
+
+    # -- lifecycle -----------------------------------------------------------
+
+    def start(self) -> 'StateSubscription':
+        if self._thread is None or not self._thread.is_alive():
+            self._stop.clear()
+            self._thread = threading.Thread(target=self._run, name='ledmatrix-state-feed',
+                                            daemon=True)
+            self._thread.start()
+        return self
+
+    def stop(self, timeout: float = 2.0) -> None:
+        self._stop.set()
+        sock = self._sock
+        if sock is not None:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+        thread = self._thread
+        if thread is not None and thread is not threading.current_thread():
+            thread.join(timeout)
+        self._thread = None
+
+    # -- the feed thread -----------------------------------------------------
+
+    def _run(self) -> None:
+        backoff = _RECONNECT_MIN_SECONDS
+        while not self._stop.is_set():
+            try:
+                self._follow()
+                backoff = _RECONNECT_MIN_SECONDS
+            except ControlError as e:
+                self.last_error = e.reason
+                if e.reason in _SLOW_RETRY_REASONS:
+                    backoff = _RECONNECT_MAX_SECONDS
+            except Exception as e:  # pylint: disable=broad-except
+                self.last_error = type(e).__name__
+            finally:
+                with self._lock:
+                    self._connected = False
+                sock, self._sock = self._sock, None
+                if sock is not None:
+                    try:
+                        sock.close()
+                    except OSError:
+                        pass
+            if self._stop.wait(backoff):
+                return
+            backoff = min(backoff * 2, _RECONNECT_MAX_SECONDS)
+
+    def _follow(self) -> None:
+        """Subscribe, then read events until the connection ends. Raises ControlError."""
+        if not socket_supported():
+            raise ControlError('unsupported', 'no Unix sockets on this platform')
+        candidates = list(self._paths) if self._paths is not None else client_socket_paths()
+        if not candidates:
+            raise ControlError('disabled', 'the control socket is turned off')
+        request_id = str(uuid.uuid4())
+        payload = encode_message(Request(id=request_id, cmd=Command.STATE_SUBSCRIBE,
+                                         args={}).to_dict())
+        sock = _connect(candidates, time.monotonic() + self._connect_timeout)
+        self._sock = sock
+        try:
+            sock.settimeout(self._connect_timeout)
+            sock.sendall(payload)
+            # A read waits for the next event; the display sends one at least
+            # every keepalive, so this much silence means it is gone.
+            sock.settimeout(self._silence)
+            reader = FrameReader(MAX_MESSAGE_BYTES)
+            first = True
+            while not self._stop.is_set():
+                data = sock.recv(65536)
+                if not data:
+                    raise ControlError('closed', 'the display closed the connection')
+                for line in reader.feed(data):
+                    obj = decode_message(line)
+                    if first:
+                        response = Response.from_dict(obj)
+                        if not response.ok:
+                            error = response.error
+                            raise ControlError(error.code if error else 'bad_response',
+                                               error.message if error else '')
+                        self._store(dict(response.result or {}), full=True)
+                        first = False
+                        continue
+                    event = StateEvent.from_dict(obj)
+                    self._store(event.result, full=event.event == StateEventKind.STATE)
+        except socket.timeout:
+            raise ControlError('timeout', 'the display went quiet') from None
+        except ProtocolError as e:
+            raise ControlError('bad_response', e.message) from None
+        except OSError as e:
+            if self._stop.is_set():
+                return
+            raise ControlError('closed', str(e)) from None
+
+    def _store(self, result: Dict[str, Any], full: bool) -> None:
+        now = self._clock()
+        with self._lock:
+            if full and isinstance(result.get('state'), dict):
+                self._snapshot = result
+                self.snapshots += 1
+            elif (self._snapshot is not None
+                  and result.get('epoch') == self._snapshot.get('epoch')):
+                # A tick: nothing changed but the render loop's liveness.
+                snap = dict(self._snapshot)
+                loop = result.get('loop')
+                if isinstance(loop, dict):
+                    snap['state'] = dict(snap.get('state') or {}, loop=loop)
+                    snap['loop'] = loop
+                snap['served_at'] = result.get('served_at', snap.get('served_at'))
+                self._snapshot = snap
+            else:
+                return  # a tick before any state, or from another epoch
+            self._received = now
+            self._connected = True

src/ipc/contract.py

@@ -30,6 +30,11 @@ later the state stream). A few commands (:data:`AWAITED_COMMANDS`) are
 answered only once the render thread has applied them, or with ``pending``
 when it has not within :data:`AWAIT_SECONDS`.
 
+``state.subscribe`` is the one exception to "one response per request": its
+response is followed, on the same connection, by :class:`StateEvent` lines
+the display pushes until either side hangs up. Events carry ``event``
+instead of ``ok``.
+
 New commands are added within a protocol version: a display that does not
 know one answers ``unknown_command``, the client falls back, and ``hello``
 lists the commands a display knows. The version changes only when the
@@ -142,11 +147,14 @@ class Command:
     ON_DEMAND_STATUS = 'on_demand.status'
     BRIGHTNESS_SET = 'brightness.set'
     PLUGIN_RELOAD = 'plugin.reload'
+    STATE_GET = 'state.get'
+    STATE_SUBSCRIBE = 'state.subscribe'
 
 
 #: Every command version 1 defines, in the order ``hello`` reports them.
-#: ``brightness.set`` and ``plugin.reload`` came in stage 2, within version 1
-#: (see the module docstring on adding commands).
+#: ``brightness.set`` and ``plugin.reload`` came in stage 2, and ``state.get``
+#: and ``state.subscribe`` in stage 3, all within version 1 (see the module
+#: docstring on adding commands).
 COMMANDS: Tuple[str, ...] = (
     Command.HELLO,
     Command.PING,
@@ -155,6 +163,8 @@ COMMANDS: Tuple[str, ...] = (
     Command.ON_DEMAND_STATUS,
     Command.BRIGHTNESS_SET,
     Command.PLUGIN_RELOAD,
+    Command.STATE_GET,
+    Command.STATE_SUBSCRIBE,
 )
 
 #: Commands that are queued for the render thread.
@@ -173,6 +183,25 @@ AWAIT_SECONDS: Dict[str, float] = {
 }
 AWAITED_COMMANDS = frozenset(AWAIT_SECONDS)
 
+#: The state stream (stage 3). ``state.subscribe`` turns its connection into
+#: a one-way stream of :class:`StateEvent` lines. Subscribers have their own
+#: bound, separate from the short request connections, so they can never
+#: take the slots a command needs.
+MAX_SUBSCRIBERS = 4
+
+#: A subscriber hears from the display at least this often: a ``state``
+#: event when something changed, else a ``tick`` carrying the render loop's
+#: liveness. A client that has heard nothing for a few of these treats its
+#: copy as unknown.
+SUBSCRIBE_KEEPALIVE_SECONDS = 5.0
+
+#: The shape of the ``state`` object in a state snapshot. Bumped only when a
+#: field changes meaning; new fields are added within a schema.
+STATE_SCHEMA = 1
+
+#: The sections of a state snapshot, in the order they are documented.
+STATE_SECTIONS: Tuple[str, ...] = ('display', 'on_demand', 'brightness', 'plugins', 'loop')
+
 #: Brightness, in percent, as the display's hardware setting takes it.
 MIN_BRIGHTNESS = 0
 MAX_BRIGHTNESS = 100
@@ -477,8 +506,65 @@ class PluginReloadArgs:
         return cls(plugin_id=plugin_id)
 
 
+def _optional_version(args: Mapping[str, Any], key: str) -> Optional[int]:
+    value = args.get(key)
+    if value is None:
+        return None
+    if not _is_int(value) or value < 0:
+        raise ProtocolError(ErrorCode.INVALID_ARGS, f'{key} must be a non-negative integer')
+    return value
+
+
+def _optional_epoch(args: Mapping[str, Any]) -> Optional[str]:
+    value = args.get('epoch')
+    if value is None or value == '':
+        return None
+    if not _valid_id(value):
+        raise ProtocolError(ErrorCode.INVALID_ARGS,
+                            f'epoch must be a printable string of 1-{MAX_ID_LENGTH} characters')
+    return str(value)
+
+
+@dataclass(frozen=True)
+class StateGetArgs:
+    """``state.get``: the display's state, as a versioned snapshot.
+
+    With ``since`` and the ``epoch`` it came from, the answer is only
+    ``{changed: false, version, epoch, served_at, loop}`` while the state is
+    still at that version, so a poller that already has it is sent no state.
+    """
+    since: Optional[int] = None
+    epoch: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {'since': self.since, 'epoch': self.epoch}
+
+    @classmethod
+    def from_dict(cls, args: Mapping[str, Any]) -> 'StateGetArgs':
+        return cls(since=_optional_version(args, 'since'), epoch=_optional_epoch(args))
+
+
+@dataclass(frozen=True)
+class StateSubscribeArgs:
+    """``state.subscribe``: the snapshot now, then a push stream of changes.
+
+    The response is the snapshot ``state.get`` returns. After it the
+    connection carries only :class:`StateEvent` lines from the display: a
+    ``state`` event whenever the state changes (always the latest version,
+    so a reader that falls behind skips versions instead of queueing them),
+    and a ``tick`` at least every :data:`SUBSCRIBE_KEEPALIVE_SECONDS`.
+    """
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {}
+
+    @classmethod
+    def from_dict(cls, args: Mapping[str, Any]) -> 'StateSubscribeArgs':
+        return cls()
+
+
 CommandArgs = Union[HelloArgs, OnDemandStartArgs, OnDemandStopArgs, NoArgs,
-                    BrightnessSetArgs, PluginReloadArgs]
+                    BrightnessSetArgs, PluginReloadArgs, StateGetArgs, StateSubscribeArgs]
 
 #: The arguments of a command that goes on the render thread's queue.
 QueuedArgs = Union[OnDemandStartArgs, OnDemandStopArgs, BrightnessSetArgs, PluginReloadArgs]
@@ -491,6 +577,8 @@ _ARG_TYPES: Dict[str, Any] = {
     Command.ON_DEMAND_STATUS: NoArgs,
     Command.BRIGHTNESS_SET: BrightnessSetArgs,
     Command.PLUGIN_RELOAD: PluginReloadArgs,
+    Command.STATE_GET: StateGetArgs,
+    Command.STATE_SUBSCRIBE: StateSubscribeArgs,
 }
 
 
@@ -563,6 +651,100 @@ class PluginReloadResult(TypedDict):
     modes: List[str]
 
 
+class LoopState(TypedDict):
+    """``loop``: is the render loop still going round?
+
+    ``heartbeat_age_seconds`` is the age of the render thread's last beat,
+    measured in memory by the display when it answered -- the same beat that
+    writes ``display-heartbeat.json``. None until the loop has drawn its first
+    frame. At ``stale_after`` or more the loop is stalled: the threshold
+    ``/api/v3/health`` uses.
+    """
+    heartbeat_age_seconds: Optional[float]
+    armed: bool
+    stale_after: float
+
+
+class StateSnapshot(TypedDict, total=False):
+    """The answer to ``state.get`` and ``state.subscribe``, and the
+    ``result`` of a ``state`` event.
+
+    ``version`` counts changes to the state within one ``epoch`` (one run of
+    the display process): a reader that sees a new epoch starts over.
+    ``changed`` is False only for a ``state.get`` whose ``since`` is still
+    current, and then ``state`` is absent. ``served_at`` is the display's
+    wall clock when it answered. ``loop`` is measured at that moment, so it
+    is also inside ``state``.
+
+    ``state`` holds the sections in :data:`STATE_SECTIONS`:
+
+    * ``display``: what ``display_current_state`` holds (mode, plugin_id,
+      mode_index, total_modes, on_demand_active, is_display_active,
+      last_updated);
+    * ``on_demand``: what ``display_on_demand_state`` holds;
+    * ``brightness``: ``{brightness, panel_brightness, dimmed}``;
+    * ``plugins``: the plugin runtime snapshot (``plugin_runtime_snapshot``),
+      or None when there is none (or it was too large to send);
+    * ``loop``: :class:`LoopState`.
+
+    A section the display has not published yet is None.
+    """
+    schema: int
+    version: int
+    epoch: str
+    pid: int
+    served_at: float
+    changed: bool
+    state: Dict[str, Any]
+    loop: LoopState
+
+
+class StateEventKind:
+    STATE = 'state'   # result: a full StateSnapshot, the latest version
+    TICK = 'tick'     # result: {version, epoch, pid, served_at, loop}; nothing changed
+
+
+@dataclass(frozen=True)
+class StateEvent:
+    """One message the display pushes to a subscriber.
+
+    ``{"v": 1, "id": "<the subscribe request's id>", "event": "state" | "tick",
+    "result": {...}}``. It has no ``ok``, which is how a reader tells it from
+    a response.
+    """
+    id: str
+    event: str
+    result: Dict[str, Any]
+    v: int = PROTOCOL_VERSION
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {'v': self.v, 'id': self.id, 'event': self.event, 'result': dict(self.result)}
+
+    @classmethod
+    def from_dict(cls, obj: Any) -> 'StateEvent':
+        """Validate an event. Raises :class:`ProtocolError` (BAD_REQUEST)."""
+        if not isinstance(obj, dict):
+            raise ProtocolError(ErrorCode.BAD_REQUEST, 'an event must be a JSON object')
+        version = obj.get('v')
+        if not _is_int(version):
+            raise ProtocolError(ErrorCode.BAD_REQUEST, 'v must be an integer')
+        raw_id = obj.get('id')
+        if not isinstance(raw_id, str):
+            raise ProtocolError(ErrorCode.BAD_REQUEST, 'id must be a string')
+        event = obj.get('event')
+        if event not in (StateEventKind.STATE, StateEventKind.TICK):
+            raise ProtocolError(ErrorCode.BAD_REQUEST, 'event must be "state" or "tick"')
+        result = obj.get('result')
+        if not isinstance(result, dict):
+            raise ProtocolError(ErrorCode.BAD_REQUEST, 'result must be a JSON object')
+        return cls(id=raw_id, event=event, result=result, v=version)
+
+
+def is_event(obj: Any) -> bool:
+    """Whether a decoded message is a pushed event rather than a response."""
+    return isinstance(obj, dict) and 'event' in obj and 'ok' not in obj
+
+
 def negotiate_version(client_versions: Tuple[int, ...]) -> Optional[int]:
     """The highest version both sides speak, or None."""
     common = set(client_versions) & set(SUPPORTED_VERSIONS)

src/ipc/server.py

@@ -14,6 +14,13 @@ every kind of screen. An awaited command (``brightness.set``,
 ``plugin.reload``) carries a :class:`CommandOutcome` that the render thread
 fills in; its connection thread waits for that, bounded, before answering.
 
+The state stream (stage 3): the display publishes what it is doing into a
+:class:`StateHub`, in memory, and ``state.get`` / ``state.subscribe`` read
+it. A subscriber's connection gives back its request slot, takes one of
+:data:`~src.ipc.contract.MAX_SUBSCRIBERS`, and is pushed the latest version
+on every change plus a keepalive tick, from its own thread: publishing never
+waits for a reader, and a reader that stops reading is dropped.
+
 Robustness rules, because this runs inside the display process:
 
 * every connection has its own daemon thread, at most :data:`MAX_CLIENTS` at
@@ -37,6 +44,7 @@ server checks them again: root, its own user, or a member of that group.
 
 from __future__ import annotations
 
+import json
 import logging
 import os
 import queue
@@ -45,8 +53,9 @@ import stat
 import struct
 import threading
 import time
+import uuid
 from dataclasses import dataclass, field
-from typing import Any, Callable, Dict, FrozenSet, List, Mapping, Optional
+from typing import Any, Callable, Dict, FrozenSet, Iterable, List, Mapping, Optional, Tuple
 
 from src.ipc.contract import (
     AWAIT_SECONDS,
@@ -55,8 +64,12 @@ from src.ipc.contract import (
     DEFAULT_SOCKET_DIR,
     DEFAULT_SOCKET_PATH,
     MAX_MESSAGE_BYTES,
+    MAX_SUBSCRIBERS,
     PROTOCOL_VERSION,
     QUEUED_COMMANDS,
+    STATE_SCHEMA,
+    STATE_SECTIONS,
+    SUBSCRIBE_KEEPALIVE_SECONDS,
     SUPPORTED_VERSIONS,
     AckResult,
     BrightnessSetArgs,
@@ -72,6 +85,9 @@ from src.ipc.contract import (
     QueuedArgs,
     Request,
     Response,
+    StateEvent,
+    StateEventKind,
+    StateGetArgs,
     configured_socket_path,
     decode_message,
     dev_socket_path,
@@ -181,6 +197,214 @@ class QueuedCommand:
             self.outcome.fail(code, message)
 
 
+# -- the state stream (stage 3) ----------------------------------------------------------
+
+#: How long a ``state.get`` keeps the display counting its readers as served
+#: over the socket (:meth:`StateHub.readers_active`). A subscriber counts for
+#: as long as it is connected.
+READER_WINDOW_SECONDS = 60.0
+
+#: Room kept for the envelope (``v``, ``id``, ``event``) around a snapshot,
+#: within MAX_MESSAGE_BYTES.
+_ENVELOPE_ROOM = 512
+
+_MISSING = object()
+
+LoopProbe = Callable[[], Mapping[str, Any]]
+
+
+def _fingerprint(value: Optional[Mapping[str, Any]], volatile: Iterable[str]) -> Any:
+    """What a section's version is judged on: the value minus its volatile keys
+    (timestamps that move on every publish without anything changing)."""
+    if value is None:
+        return None
+    skip = frozenset(volatile)
+    return {k: v for k, v in value.items() if k not in skip} if skip else dict(value)
+
+
+def _unknown_loop() -> Dict[str, Any]:
+    return {'heartbeat_age_seconds': None, 'armed': False, 'stale_after': None}
+
+
+def fit_snapshot(snapshot: Dict[str, Any]) -> Dict[str, Any]:
+    """``snapshot``, or a copy without the plugin runtime section when the
+    message would be over MAX_MESSAGE_BYTES (hundreds of plugins). The
+    reader then falls back to the cache for that section only; ``truncated``
+    says which was left out."""
+    state = snapshot.get('state')
+    if not isinstance(state, dict) or state.get('plugins') is None:
+        return snapshot
+    try:
+        size = len(json.dumps(snapshot, separators=(',', ':'), ensure_ascii=True,
+                              allow_nan=False))
+    except (TypeError, ValueError):
+        size = MAX_MESSAGE_BYTES
+    if size <= MAX_MESSAGE_BYTES - _ENVELOPE_ROOM:
+        return snapshot
+    logger.warning("State snapshot is %d bytes; sending it without the plugin runtime "
+                   "section", size)
+    trimmed = dict(snapshot)
+    trimmed['state'] = dict(state, plugins=None)
+    trimmed['truncated'] = ['plugins']
+    return trimmed
+
+
+class StateHub:
+    """The display's live state, in memory, for ``state.get`` and ``state.subscribe``.
+
+    Writers publish whole sections (:meth:`publish`): the render thread
+    publishes ``display``, ``on_demand`` and ``brightness``, and the plugin
+    runtime publisher's thread publishes ``plugins``. Each section has one
+    writer. ``loop`` is not published: it is measured when a reader asks
+    (``loop_probe``), so it keeps ageing while the render thread is stuck.
+
+    The version goes up when a section's value changes, ignoring the keys
+    the publisher names as volatile (timestamps). Publishing never blocks on
+    a reader: the lock is held only to swap a dict reference and compare it,
+    and every socket write happens on the reader's own thread, outside it.
+    A reader that is slow gets the latest version when it next asks, not
+    every version in between.
+    """
+
+    def __init__(self, loop_probe: Optional[LoopProbe] = None, *,
+                 clock: Callable[[], float] = time.monotonic,
+                 wall_clock: Callable[[], float] = time.time,
+                 epoch: Optional[str] = None, pid: Optional[int] = None,
+                 reader_window: float = READER_WINDOW_SECONDS):
+        self._cond = threading.Condition(threading.Lock())
+        self._sections: Dict[str, Optional[Dict[str, Any]]] = {}
+        self._fingerprints: Dict[str, Any] = {}
+        self._version = 0
+        self.epoch = epoch or uuid.uuid4().hex[:16]
+        self.pid = os.getpid() if pid is None else pid
+        self._loop_probe = loop_probe
+        self._clock = clock
+        self._wall_clock = wall_clock
+        self._reader_window = reader_window
+        self._last_read: Optional[float] = None
+        self._subscribers = 0
+
+    @property
+    def version(self) -> int:
+        return self._version
+
+    @property
+    def subscribers(self) -> int:
+        return self._subscribers
+
+    # -- writers -------------------------------------------------------------
+
+    def publish(self, section: str, value: Optional[Mapping[str, Any]],
+                volatile: Iterable[str] = ()) -> bool:
+        """Store a section's latest value; True when that is a new version.
+
+        The value is copied (one level), so the caller may reuse its dict.
+        """
+        stored = None if value is None else dict(value)
+        fingerprint = _fingerprint(stored, volatile)
+        with self._cond:
+            self._sections[section] = stored
+            if self._fingerprints.get(section, _MISSING) == fingerprint:
+                return False
+            self._fingerprints[section] = fingerprint
+            self._version += 1
+            self._cond.notify_all()
+        return True
+
+    def wake(self) -> None:
+        """Wake every waiting reader (the server is closing)."""
+        with self._cond:
+            self._cond.notify_all()
+
+    # -- readers -------------------------------------------------------------
+
+    def loop(self) -> Dict[str, Any]:
+        """The render loop's liveness now. Never raises."""
+        if self._loop_probe is None:
+            return _unknown_loop()
+        try:
+            return dict(self._loop_probe())
+        except Exception:  # pylint: disable=broad-except
+            logger.debug("Render loop liveness probe failed", exc_info=True)
+            return _unknown_loop()
+
+    def snapshot(self, since: Optional[int] = None,
+                 epoch: Optional[str] = None) -> Dict[str, Any]:
+        """The :class:`~src.ipc.contract.StateSnapshot` now.
+
+        ``since`` with this hub's ``epoch``, still the current version, gives
+        the short ``changed: false`` form.
+        """
+        with self._cond:
+            version = self._version
+            sections = dict(self._sections)
+        loop = self.loop()
+        result: Dict[str, Any] = {
+            'schema': STATE_SCHEMA,
+            'version': version,
+            'epoch': self.epoch,
+            'pid': self.pid,
+            'served_at': self._wall_clock(),
+            'loop': loop,
+        }
+        if since is not None and epoch == self.epoch and since == version:
+            result['changed'] = False
+            return result
+        state: Dict[str, Any] = {name: sections.get(name) for name in STATE_SECTIONS
+                                 if name != 'loop'}
+        state['loop'] = loop
+        result['changed'] = True
+        result['state'] = state
+        return result
+
+    def wait_for_change(self, version: int, timeout: float,
+                        stop: Optional[threading.Event] = None) -> bool:
+        """Block up to ``timeout`` for a version other than ``version``."""
+        with self._cond:
+            self._cond.wait_for(
+                lambda: self._version != version or (stop is not None and stop.is_set()),
+                timeout)
+            return self._version != version
+
+    # -- who is reading ------------------------------------------------------
+
+    def note_read(self) -> None:
+        self._last_read = self._clock()
+
+    def subscriber_joined(self) -> None:
+        with self._cond:
+            self._subscribers += 1
+
+    def subscriber_left(self) -> None:
+        with self._cond:
+            self._subscribers = max(0, self._subscribers - 1)
+            self._last_read = self._clock()
+
+    def readers_active(self) -> bool:
+        """Is the socket serving state readers? A subscriber is connected, or a
+        ``state.get`` came within the reader window. The display uses this to
+        write the cache copies of the same state less often."""
+        if self._subscribers > 0:
+            return True
+        last = self._last_read
+        return last is not None and self._clock() - last < self._reader_window
+
+
+class _Slot:
+    """A connection slot, released once (a subscriber gives its back early)."""
+
+    def __init__(self, semaphore: threading.BoundedSemaphore):
+        self._semaphore = semaphore
+        self._held = True
+        self._lock = threading.Lock()
+
+    def release(self) -> None:
+        with self._lock:
+            if self._held:
+                self._held = False
+                self._semaphore.release()
+
+
 # -- peer credentials ------------------------------------------------------------------
 
 @dataclass(frozen=True)
@@ -315,8 +539,14 @@ class ControlServer:
                  message_timeout: float = MESSAGE_TIMEOUT_SECONDS,
                  idle_timeout: float = IDLE_TIMEOUT_SECONDS,
                  check_peer: bool = True,
-                 await_seconds: Optional[Mapping[str, float]] = None):
+                 await_seconds: Optional[Mapping[str, float]] = None,
+                 state_hub: Optional[StateHub] = None,
+                 max_subscribers: int = MAX_SUBSCRIBERS,
+                 keepalive: float = SUBSCRIBE_KEEPALIVE_SECONDS):
         self.path = path
+        self.state_hub = state_hub
+        self._subscriber_slots = threading.BoundedSemaphore(max_subscribers)
+        self._keepalive = keepalive
         self._await_seconds: Dict[str, float] = dict(AWAIT_SECONDS)
         if await_seconds:
             self._await_seconds.update(await_seconds)
@@ -378,6 +608,8 @@ class ControlServer:
         """Stop accepting and remove the socket file (only if it is still ours)."""
         self._stopping.set()
         self._close_socket()
+        if self.state_hub is not None:
+            self.state_hub.wake()   # subscribers see _stopping and hang up
         thread = self._thread
         if thread is not None and thread is not threading.current_thread():
             thread.join(timeout=2.0)
@@ -549,6 +781,7 @@ class ControlServer:
 
     def _serve(self, conn: socket.socket) -> None:
         """One connection: authenticate, then answer requests until it ends."""
+        slot = _Slot(self._slots)
         try:
             conn.settimeout(self._io_timeout)
             peer = peer_credentials(conn)
@@ -557,7 +790,7 @@ class ControlServer:
                                "this user or group %s", peer.pid, peer.uid, peer.gid, self._group)
                 self._send(conn, Response.failure(None, ErrorCode.FORBIDDEN, 'not permitted'))
                 return
-            self._read_requests(conn, peer)
+            self._read_requests(conn, peer, slot)
         except Exception:  # pylint: disable=broad-except
             logger.exception("Control socket connection failed")
         finally:
@@ -565,7 +798,7 @@ class ControlServer:
                 conn.close()
             except OSError:
                 pass
-            self._slots.release()
+            slot.release()
 
     def _peer_ok(self, peer: PeerCredentials) -> bool:
         groups = None
@@ -573,7 +806,8 @@ class ControlServer:
             groups = process_groups(peer.pid)
         return peer_allowed(peer, self._own_uid, self._group, groups)
 
-    def _read_requests(self, conn: socket.socket, peer: Optional[PeerCredentials]) -> None:
+    def _read_requests(self, conn: socket.socket, peer: Optional[PeerCredentials],
+                       slot: Optional[_Slot] = None) -> None:
         reader = FrameReader(MAX_MESSAGE_BYTES)
         idle_since = time.monotonic()
         message_started: Optional[float] = None
@@ -598,7 +832,13 @@ class ControlServer:
                 self._send(conn, Response.failure(None, e.code, e.message))
                 return  # can't find the next message boundary: hang up
             for line in lines:
-                if not self._send(conn, self.handle_line(line, peer)):
+                response, cmd = self._handle(line, peer)
+                if cmd == Command.STATE_SUBSCRIBE and response.ok:
+                    # The connection becomes a one-way stream; anything the
+                    # client sent after the subscribe is ignored.
+                    self._subscribe(conn, response, slot)
+                    return
+                if not self._send(conn, response):
                     return
             if reader.pending:
                 if message_started is None or lines:
@@ -624,20 +864,91 @@ class ControlServer:
     # -- requests --------------------------------------------------------------------
 
     def handle_line(self, line: bytes, peer: Optional[PeerCredentials] = None) -> Response:
-        """Answer one request line. Never raises."""
+        """Answer one request line. Never raises.
+
+        A ``state.subscribe`` answered here gets its snapshot only; the
+        stream that follows needs a connection (``_read_requests``).
+        """
+        return self._handle(line, peer)[0]
+
+    def _handle(self, line: bytes,
+                peer: Optional[PeerCredentials]) -> Tuple[Response, Optional[str]]:
+        """The response to one line, and the command it answered (when known)."""
         request_id: Optional[str] = None
+        cmd: Optional[str] = None
         try:
             obj = decode_message(line)
             raw_id = obj.get('id')
             request_id = raw_id if isinstance(raw_id, str) and len(raw_id) <= 128 else None
             request = Request.from_dict(obj)
             request_id = request.id
-            return self._dispatch(request, peer)
+            cmd = request.cmd
+            return self._dispatch(request, peer), cmd
         except ProtocolError as e:
-            return Response.failure(e.request_id or request_id, e.code, e.message)
+            return Response.failure(e.request_id or request_id, e.code, e.message), cmd
         except Exception:  # pylint: disable=broad-except
             logger.exception("Control socket handler failed")
-            return Response.failure(request_id, ErrorCode.INTERNAL, 'internal error')
+            return Response.failure(request_id, ErrorCode.INTERNAL, 'internal error'), cmd
+
+    # -- the state stream ----------------------------------------------------------
+
+    def _subscribe(self, conn: socket.socket, response: Response,
+                   slot: Optional[_Slot]) -> None:
+        """Answer a ``state.subscribe`` and push state events until it ends.
+
+        Subscribers have their own bound (MAX_SUBSCRIBERS) and give their
+        request slot back, so a few browsers watching never use up the slots
+        commands need. Everything here runs on this connection's thread: a
+        reader that does not keep up only stalls its own sends, and one that
+        stops reading for a whole IO timeout is dropped. The render thread
+        only ever publishes into the hub.
+        """
+        hub = self.state_hub
+        if hub is None or not self._subscriber_slots.acquire(blocking=False):
+            self._send(conn, Response.failure(response.id, ErrorCode.BUSY,
+                                              'too many state subscribers', v=response.v))
+            return
+        if slot is not None:
+            slot.release()
+        hub.subscriber_joined()
+        try:
+            if not self._send(conn, response):
+                return
+            result = response.result or {}
+            version = result.get('version', -1)
+            sub_id = response.id or ''
+            logger.debug("Control socket: state subscriber joined at version %s", version)
+            while not self._stopping.is_set():
+                hub.wait_for_change(version, self._keepalive, self._stopping)
+                if self._stopping.is_set():
+                    return
+                snap = hub.snapshot(since=version, epoch=hub.epoch)
+                if snap.get('changed'):
+                    version = snap['version']
+                    event = StateEvent(sub_id, StateEventKind.STATE, fit_snapshot(snap),
+                                       v=response.v)
+                else:
+                    event = StateEvent(sub_id, StateEventKind.TICK, snap, v=response.v)
+                if not self._send_event(conn, event):
+                    return
+        finally:
+            hub.subscriber_left()
+            self._subscriber_slots.release()
+
+    def _send_event(self, conn: socket.socket, event: StateEvent) -> bool:
+        try:
+            data = encode_message(event.to_dict())
+        except ProtocolError as e:
+            logger.error("Control socket state event not sent: %s", e.message)
+            return False
+        try:
+            conn.sendall(data)
+            return True
+        except socket.timeout:
+            logger.info("Control socket: dropping a state subscriber that stopped reading")
+            return False
+        except OSError:
+            return False
 
     def _dispatch(self, request: Request, peer: Optional[PeerCredentials]) -> Response:
         if request.cmd == Command.HELLO:
@@ -678,6 +989,18 @@ class ControlServer:
                                         v=request.v)
             return Response.success(request.id, self._status_provider(), v=request.v)
 
+        if request.cmd in (Command.STATE_GET, Command.STATE_SUBSCRIBE):
+            hub = self.state_hub
+            if hub is None:
+                return Response.failure(request.id, ErrorCode.INTERNAL, 'no state available',
+                                        v=request.v)
+            if isinstance(args, StateGetArgs):
+                hub.note_read()
+                snap = hub.snapshot(since=args.since, epoch=args.epoch)
+            else:
+                snap = hub.snapshot()
+            return Response.success(request.id, fit_snapshot(snap), v=request.v)
+
         if request.cmd in QUEUED_COMMANDS and isinstance(args, (
                 OnDemandStartArgs, OnDemandStopArgs, BrightnessSetArgs, PluginReloadArgs)):
             awaited = request.cmd in AWAITED_COMMANDS
@@ -728,22 +1051,25 @@ class ControlServer:
 
 def start_control_server(status_provider: Optional[StatusProvider] = None,
                          cache_dir: Optional[str] = None,
-                         environ: Optional[Mapping[str, str]] = None) -> Optional[ControlServer]:
+                         environ: Optional[Mapping[str, str]] = None,
+                         state_hub: Optional[StateHub] = None) -> Optional[ControlServer]:
     """Start the display's control socket, or return None when it can't run.
 
     None covers Windows, ``LEDMATRIX_CONTROL_SOCKET=off`` and any failure to
-    bind; in every case the web interface falls back to the file mailbox.
+    bind; in every case the web interface falls back to the file mailbox
+    and to the cache keys the display still writes.
     """
     path = server_socket_path(environ)
     if path is None:
         logger.debug("Control socket disabled or unsupported here; using the file mailbox only")
         return None
-    server = ControlServer(path, status_provider, resolve_socket_group(cache_dir))
+    server = ControlServer(path, status_provider, resolve_socket_group(cache_dir),
+                           state_hub=state_hub)
     return server if server.start() else None
 
 
 __all__ = [
-    'CommandOutcome', 'ControlServer', 'PeerCredentials', 'QueuedCommand', 'StatusProvider',
-    'peer_allowed', 'peer_credentials', 'process_groups', 'resolve_socket_group',
-    'server_socket_path', 'start_control_server', 'PROTOCOL_VERSION',
+    'CommandOutcome', 'ControlServer', 'PeerCredentials', 'QueuedCommand', 'StateHub',
+    'StatusProvider', 'fit_snapshot', 'peer_allowed', 'peer_credentials', 'process_groups',
+    'resolve_socket_group', 'server_socket_path', 'start_control_server', 'PROTOCOL_VERSION',
 ]

src/plugin_system/plugin_runtime.py

@@ -36,6 +36,13 @@ tmpfs. A missing heartbeat (dev server, emulator, Windows, a display still
 starting up) or one from another process (a display restarted after a
 watchdog kill) says nothing, and the snapshot is judged on its own.
 
+The control socket. Where the display serves its state stream (stage 3,
+docs/IPC_CONTROL_SOCKET.md), every tick also hands the snapshot to it, in
+memory, and the web interface reads it there first
+(``view_from_socket_state``, judged by the same rules). While the socket
+serves those readers, the cache copy is their fallback and an unchanged
+snapshot is rewritten every ``RELAXED_REFRESH_INTERVAL`` instead.
+
 A dead publisher. systemd removes the heartbeat's directory when the
 service stops, so after a watchdog kill there is no heartbeat to go stale.
 The reader then asks whether the snapshot's ``pid`` still exists (POSIX
@@ -48,7 +55,7 @@ import math
 import os
 import threading
 import time
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any, Callable, Dict, Optional
 
 from src import display_watchdog
@@ -75,6 +82,15 @@ TICK_INTERVAL = 5.0
 #: A snapshot older than this is stale: three missed refreshes.
 STALE_AFTER = 3 * REFRESH_INTERVAL
 
+#: The refresh while the control socket serves the web interface's readers
+#: (``StateHub.readers_active``). The cache copy is then only their fallback,
+#: so an unchanged snapshot is rewritten half as often; the snapshot says so
+#: in its own ``refresh_interval`` and ``stale_after``.
+RELAXED_REFRESH_INTERVAL = 2 * REFRESH_INTERVAL
+
+#: The control socket's section for this snapshot (``state.plugins``).
+STATE_SECTION = "plugins"
+
 #: Bounds on a published ``stale_after``, so a corrupt value can make a
 #: reader neither trust a dead display for hours nor distrust a live one.
 _STALE_AFTER_MIN = 30.0
@@ -140,7 +156,8 @@ def summarize_error(error_info: Optional[Dict[str, Any]]) -> Optional[Dict[str,
 
 def build_runtime_snapshot(state_manager: Any, *, started_at: float,
                            now: Optional[float] = None,
-                           running: bool = True) -> Dict[str, Any]:
+                           running: bool = True,
+                           refresh_interval: float = REFRESH_INTERVAL) -> Dict[str, Any]:
     """The snapshot for ``state_manager`` (a plugin_state.PluginStateManager).
 
     A stopped snapshot (``running=False``) lists no plugins: nothing is
@@ -162,8 +179,8 @@ def build_runtime_snapshot(state_manager: Any, *, started_at: float,
         "running": running,
         "published_at": time.time() if now is None else now,
         "started_at": started_at,
-        "refresh_interval": REFRESH_INTERVAL,
-        "stale_after": STALE_AFTER,
+        "refresh_interval": refresh_interval,
+        "stale_after": 3 * refresh_interval,
         "pid": os.getpid(),
         "plugins": plugins,
     }
@@ -197,30 +214,87 @@ class PluginRuntimePublisher:
         self._tick_lock = threading.Lock()
         self._stop = threading.Event()
         self._thread: Optional[threading.Thread] = None
+        # The control socket's state stream (src/ipc/server.StateHub), when
+        # the display serves one: every tick also hands it the snapshot, in
+        # memory, and the cache refresh relaxes while it has readers.
+        self._hub: Any = None
+        self._hub_change: Optional[int] = None
+        self._hub_snapshot: Optional[Dict[str, Any]] = None
+        self.relaxed_refresh_interval = RELAXED_REFRESH_INTERVAL
 
-    def _write(self, running: bool) -> None:
-        snapshot = build_runtime_snapshot(self.state_manager, started_at=self.started_at,
-                                          now=self._wall_clock(), running=running)
+    def attach_hub(self, hub: Any) -> None:
+        """Also publish to the control socket's state hub, starting now."""
+        with self._tick_lock:
+            self._hub = hub
+            self._hub_change = None
+            self._hub_snapshot = None
+            try:
+                self._push_to_hub(self.state_manager.change_count)
+            except Exception as err:  # never let reporting break the display
+                logger.debug("Could not publish the plugin runtime state: %s", err,
+                             exc_info=True)
+
+    def _push_to_hub(self, change: int) -> None:
+        """The snapshot to the state hub: rebuilt when the state machine
+        changed, otherwise the last one with a new ``published_at``, which
+        the hub does not count as a new version. In memory, every tick, so
+        the socket's copy is never more than a tick old."""
+        hub = self._hub
+        if hub is None:
+            return
+        now = self._wall_clock()
+        if self._hub_snapshot is None or change != self._hub_change:
+            snapshot = build_runtime_snapshot(self.state_manager, started_at=self.started_at,
+                                              now=now)
+        else:
+            snapshot = dict(self._hub_snapshot, published_at=now)
+        hub.publish(STATE_SECTION, snapshot, volatile=("published_at",))
+        self._hub_snapshot = snapshot
+        self._hub_change = change
+
+    def _cache_refresh_interval(self) -> float:
+        """The cache refresh: relaxed while the socket serves the readers."""
+        hub = self._hub
+        try:
+            if hub is not None and hub.readers_active():
+                return self.relaxed_refresh_interval
+        except Exception:  # pylint: disable=broad-except
+            # The normal interval is the safe answer: it only writes more.
+            logger.debug("State hub readers_active() failed; using the normal refresh", exc_info=True)
+        return self.refresh_interval
+
+    def _write(self, running: bool, refresh_interval: Optional[float] = None) -> None:
+        snapshot = build_runtime_snapshot(
+            self.state_manager, started_at=self.started_at, now=self._wall_clock(),
+            running=running,
+            refresh_interval=self.refresh_interval if refresh_interval is None
+            else refresh_interval)
         self.cache_manager.set(PLUGIN_RUNTIME_KEY, snapshot)
 
     def tick(self) -> bool:
         """Publish if something changed (throttled) or the refresh is due.
-        True if a snapshot was written."""
+        True if a snapshot was written to the cache."""
         with self._tick_lock:
             try:
                 change = self.state_manager.change_count
+                try:
+                    self._push_to_hub(change)
+                except Exception as err:  # the cache copy still goes out below
+                    logger.debug("Could not publish the plugin runtime state: %s", err,
+                                 exc_info=True)
                 now = self._clock()
+                refresh = self._cache_refresh_interval()
                 since = None if self._last_attempt is None else now - self._last_attempt
                 if since is not None:
                     if change == self._published_change:
-                        if since < self.refresh_interval:
+                        if since < refresh:
                             return False
                     elif since < self.min_interval:
                         return False
                 # Stamp the attempt before writing: a cache that keeps failing
                 # is retried at the throttled rate, not on every tick.
                 self._last_attempt = now
-                self._write(running=True)
+                self._write(running=True, refresh_interval=refresh)
                 self._published_change = change
                 return True
             except Exception as err:  # never let reporting break the display
@@ -317,6 +391,9 @@ class PluginRuntimeView:
     plugins: Dict[str, Dict[str, Any]] = field(default_factory=dict)
     #: Age of the render loop's heartbeat, when it was taken into account.
     heartbeat_age_seconds: Optional[float] = None
+    #: Where the snapshot came from: ``cache`` (the shared cache file and the
+    #: heartbeat file) or ``socket`` (the control socket's state stream).
+    source: str = "cache"
 
     @property
     def live(self) -> bool:
@@ -348,6 +425,7 @@ class PluginRuntimeView:
             "stale_after": self.stale_after,
             "heartbeat_age_seconds": (None if self.heartbeat_age_seconds is None
                                       else round(self.heartbeat_age_seconds, 1)),
+            "source": self.source,
         }
 
 
@@ -445,6 +523,36 @@ def view_from_snapshot(snapshot: Any, now: Optional[float] = None,
     )
 
 
+def view_from_socket_state(snapshot: Any, now: Optional[float] = None,
+                           now_mono: Optional[float] = None) -> Optional[PluginRuntimeView]:
+    """Judge the ``plugins`` section of a control-socket state snapshot by
+    the same rules as the cache copy; None when it has none (an older
+    display, or a snapshot too large to carry it), so the caller reads the
+    cache instead.
+
+    The display measured its render loop's heartbeat age when it answered
+    (``state.loop``); that is the heartbeat here, aged by the time since the
+    answer arrived. A live snapshot with a stalled loop is ``stalled``, and a
+    snapshot older than its ``stale_after`` (the publisher thread stopped)
+    is ``stale``, exactly as for the cache. The display answered, so its
+    process is alive: there is no pid check.
+    """
+    from src.ipc.client import snapshot_loop_age  # stdlib-only module
+    if not isinstance(snapshot, dict):
+        return None
+    state = snapshot.get("state")
+    plugins = state.get(STATE_SECTION) if isinstance(state, dict) else None
+    if not isinstance(plugins, dict):
+        return None
+    now_mono = time.monotonic() if now_mono is None else now_mono
+    beat_age = snapshot_loop_age(snapshot, now_mono=now_mono)
+    heartbeat = None
+    if beat_age is not None:
+        heartbeat = {"pid": plugins.get("pid"), "mono": now_mono - beat_age}
+    view = view_from_snapshot(plugins, now=now, heartbeat=heartbeat, now_mono=now_mono)
+    return replace(view, source="socket")
+
+
 def read_plugin_runtime(cache_manager: Any, now: Optional[float] = None,
                         heartbeat_path: Optional[str] = None) -> PluginRuntimeView:
     """The display's latest snapshot, judged for staleness and against the