Merge pull request #69533 from NousResearch/bb/skin-live-broadcast

fix(themes): live skin sync reaches every surface — WS fan-out + missed-activation recovery

apps/desktop/src/themes/backend-sync.test.ts

@@ -39,15 +39,41 @@ describe('ingestBackendSkin', () => {
     expect($pendingSkinApply.get()).toBe('forest')
   })
 
-  it('seeds on connect so the first matching poll is a no-op, but a change applies', () => {
+  it('seed does not paint, but a later same-name skin.changed applies (missed-activation recovery)', () => {
+    // Connect while display.skin is already neon: seed records the baseline
+    // without painting (never stomp the persisted desktop theme on connect).
     ingestBackendSkin(skin('neon'), { apply: false }) // gateway.ready seed
-    ingestBackendSkin(skin('neon'), { apply: true }) // post-turn poll, unchanged
     expect($pendingSkinApply.get()).toBeNull()
 
+    // The activation event was missed (skin set while disconnected / backend
+    // restarted). Hermes re-affirms it — `hermes config set display.skin neon`
+    // or a `hermes skin set` recolor. That explicit event must repaint even
+    // though the name matches the seed.
+    ingestBackendSkin(skin('neon'), { apply: true })
+    expect($pendingSkinApply.get()).toBe('neon')
+
+    // Once applied, a repeat same-name event is a no-op again...
+    $pendingSkinApply.set(null)
+    ingestBackendSkin(skin('neon'), { apply: true })
+    expect($pendingSkinApply.get()).toBeNull()
+
+    // ...and a genuine switch still applies.
     ingestBackendSkin(skin('forest'), { apply: true }) // Hermes authored a new skin
     expect($pendingSkinApply.get()).toBe('forest')
   })
 
+  it('a reconnect re-seed after a real apply does not downgrade the applied baseline', () => {
+    ingestBackendSkin(skin('neon'), { apply: true }) // applied for real
+    $pendingSkinApply.set(null)
+
+    ingestBackendSkin(skin('neon'), { apply: false }) // reconnect: gateway.ready re-seed
+    ingestBackendSkin(skin('neon'), { apply: true }) // repeat event (e.g. in-place recolor)
+
+    // Already painted once — the repeat must not re-apply (protects a manual
+    // desktop-side theme switch from being snapped back after a reconnect).
+    expect($pendingSkinApply.get()).toBeNull()
+  })
+
   it('never registers default in the backend store (desktop keeps its own palette)', () => {
     ingestBackendSkin(skin('default'), { apply: true })
 

apps/desktop/src/themes/backend-sync.ts

@@ -29,10 +29,13 @@ export const $backendThemes = atom<Record<string, DesktopTheme>>({})
 /** One-shot skin name the ThemeProvider should switch to (it clears this). */
 export const $pendingSkinApply = atom<string | null>(null)
 
-// The last skin name we drove onto the desktop. Guards two things: re-applying
-// the same skin every post-turn poll, and snapping back after a manual switch —
-// only a CHANGE from this value applies. `default` is the "no opinion" sentinel.
-let lastSynced: string | null = null
+// Last skin name synced from the backend + whether it was ever APPLIED (vs
+// merely seeded at connect). Once applied, only a name change applies again —
+// no re-apply on repeat events, no snap-back after a manual desktop switch.
+// A `skin.changed` matching a seed-only baseline still applies: the seed
+// records without painting, so if the activation event was missed (backend
+// restart / disconnected), an explicit re-affirm must repaint, not no-op.
+let lastSynced: { applied: boolean; name: string } | null = null
 
 /** Test-only: reset the module's apply guard + registry between cases. */
 export function __resetBackendSkinSync(): void {
@@ -75,14 +78,17 @@ export function ingestBackendSkin(skin: HermesSkin | undefined | null, { apply }
   }
 
   if (!apply) {
-    // Connect-time seed: record the baseline so a later poll is a no-op.
-    lastSynced = name
+    // Connect-time seed: record without painting. A reconnect re-seed keeps an
+    // earlier real apply's flag so repeat events can't override a manual switch.
+    if (lastSynced?.name !== name || !lastSynced.applied) {
+      lastSynced = { applied: false, name }
+    }
 
     return
   }
 
-  if (name !== lastSynced) {
-    lastSynced = name
+  if (name !== lastSynced?.name || !lastSynced.applied) {
+    lastSynced = { applied: true, name }
     $pendingSkinApply.set(name)
   }
 }

hermes_cli/config.py

@@ -8986,6 +8986,19 @@ def set_config_value(key: str, value: str, force: bool = False):
     if env_var and key != "terminal.cwd":
         save_env_value(env_var, _terminal_env_value(value))
 
+    # Setting display.skin is an explicit "apply NOW" — bump the skin file's
+    # mtime so the gateway watcher's (name, mtime) signature moves even when the
+    # name is unchanged (re-affirming the active skin after a surface missed the
+    # original activation). Built-ins have no file; a name switch already moves
+    # their signature.
+    if key == "display.skin" and isinstance(value, str) and value:
+        try:
+            skin_file = get_hermes_home() / "skins" / f"{value}.yaml"
+            if skin_file.exists():
+                skin_file.touch()
+        except Exception:
+            pass  # best-effort: the config write above already succeeded
+
     # Mask the echoed value when the (possibly nested) key is credential-shaped
     # — e.g. `hermes config set model.api_key cfut_...` routes to config.yaml
     # (lowercase, so it misses the .env api_keys list above) and would otherwise

tests/hermes_cli/test_set_config_value.py

@@ -630,3 +630,49 @@ class TestValidateConfigKey:
         from hermes_cli.config import _validate_config_key
         is_known, suggestion = _validate_config_key("agent._max_turns")
         assert not is_known, "Sub-key typo under a known top-level key must still be flagged"
+
+
+# ---------------------------------------------------------------------------
+# display.skin → touch the skin file (live re-affirm broadcast)
+# ---------------------------------------------------------------------------
+
+class TestDisplaySkinTouch:
+    """Setting display.skin must bump the named skin file's mtime.
+
+    The gateway's skin watcher broadcasts ``skin.changed`` on a signature move
+    of (active name, skin-file mtime). Re-affirming the already-configured skin
+    (`hermes config set display.skin X` while it is already X — the recovery
+    path when a surface missed the original activation) moves NEITHER part, so
+    without the touch the explicit apply is invisible to every live surface.
+    """
+
+    def test_reaffirming_same_skin_moves_the_watcher_signature(self, _isolated_hermes_home):
+        import os as _os
+        skins = _isolated_hermes_home / "skins"
+        skins.mkdir()
+        skin_file = skins / "synthwave.yaml"
+        skin_file.write_text("name: synthwave\ncolors:\n  background: '#1a1030'\n")
+        # Age the file so an mtime bump is unambiguous even on coarse clocks.
+        _os.utime(skin_file, (1_000_000_000, 1_000_000_000))
+
+        set_config_value("display.skin", "synthwave")
+        first = skin_file.stat().st_mtime
+        assert first > 1_000_000_000
+
+        _os.utime(skin_file, (1_000_000_000, 1_000_000_000))
+        set_config_value("display.skin", "synthwave")  # same name, re-affirmed
+        assert skin_file.stat().st_mtime > 1_000_000_000
+
+    def test_builtin_or_missing_skin_file_is_fine(self, _isolated_hermes_home):
+        """Built-ins have no user file — the set must still succeed cleanly."""
+        set_config_value("display.skin", "mono")
+        assert "skin: mono" in _read_config(_isolated_hermes_home)
+
+    def test_touch_preserves_skin_file_contents(self, _isolated_hermes_home):
+        skins = _isolated_hermes_home / "skins"
+        skins.mkdir()
+        body = "name: neon\ncolors:\n  ui_accent: '#ff33aa'\n"
+        (skins / "neon.yaml").write_text(body)
+
+        set_config_value("display.skin", "neon")
+        assert (skins / "neon.yaml").read_text() == body

tests/test_tui_gateway_ws.py

@@ -127,6 +127,28 @@ def test_ws_disconnect_preserves_and_repoints_reconnectable_session(monkeypatch)
         server._sessions.clear()
 
 
+def test_ws_connection_registers_then_disconnect_unregisters_live_transport(monkeypatch):
+    """A connected client must be tracked in the live-transport registry so a
+    session-less global broadcast (skin.changed from the background watcher)
+    reaches it, and dropped on disconnect so no stale write targets a dead peer.
+    This is the WS half of the cross-surface live-theme fix."""
+    server._sessions.clear()
+    server._live_transports.clear()
+    seen = {}
+    try:
+        _run_disconnect(
+            monkeypatch,
+            lambda t: seen.__setitem__("registered", t in server._live_transports),
+        )
+        # Seeded at receive_text time — i.e. after gateway.ready registered it.
+        assert seen["registered"] is True
+        # handle_ws's finally must have unregistered it.
+        assert not server._live_transports
+    finally:
+        server._sessions.clear()
+        server._live_transports.clear()
+
+
 def test_ws_write_loop_stall_does_not_latch_transport(monkeypatch):
     """A write that times out because the event loop is stalled (GIL-heavy
     agent turn) must NOT latch the transport closed — the frame is already

tests/tui_gateway/test_protocol.py

@@ -41,6 +41,7 @@ def server():
         mod._sessions.clear()
         mod._pending.clear()
         mod._answers.clear()
+        mod._live_transports.clear()
 
 
 @pytest.fixture()
@@ -2161,3 +2162,118 @@ def test_broadcast_skin_if_changed_on_any_signature_move(server, monkeypatch):
         server._broadcast_skin_if_changed()
 
     assert [ev for ev, _ in emitted] == ["skin.changed"] * 3
+
+
+# ── global-event broadcast (session-less events reach every WS client) ──
+
+
+class _RecordingTransport:
+    """Minimal Transport stand-in that records the frames written to it."""
+
+    def __init__(self) -> None:
+        self.frames: list[dict] = []
+
+    def write(self, obj: dict) -> bool:
+        self.frames.append(obj)
+        return True
+
+    def close(self) -> None:
+        pass
+
+
+def test_broadcast_global_event_reaches_registered_transports_from_background_thread(capture):
+    """The core of the bug: a session-less event emitted from a thread with no
+    contextvar-bound transport (the skin watcher) must still reach connected WS
+    clients. Before the registry it fell through to stdout and the desktop GUI
+    never repainted."""
+    server, buf = capture
+    a, b = _RecordingTransport(), _RecordingTransport()
+    server.register_live_transport(a)
+    server.register_live_transport(b)
+
+    # Emit from a background thread — no request context, no contextvar binding,
+    # exactly like the skin watcher loop.
+    t = threading.Thread(
+        target=server._broadcast_global_event,
+        args=("skin.changed", {"name": "synthwave", "colors": {"background": "#1a1030"}}),
+    )
+    t.start()
+    t.join(timeout=2)
+
+    for transport in (a, b):
+        assert len(transport.frames) == 1
+        params = transport.frames[0]["params"]
+        assert params["type"] == "skin.changed"
+        assert params["session_id"] == ""  # session-less: a surface-global announce
+        assert params["payload"]["name"] == "synthwave"
+
+    # It must NOT have leaked onto stdout when live transports exist.
+    assert buf.getvalue() == ""
+
+
+def test_broadcast_global_event_falls_back_to_stdio_without_transports(capture):
+    """With no client registered (the stdio TUI path, whose stdio transport is
+    tee'd to the dashboard WS publisher, and tests), broadcasting still emits via
+    write_json so that surface is unchanged."""
+    server, buf = capture
+    assert not server._live_transports
+
+    server._broadcast_global_event("skin.changed", {"name": "midnight"})
+
+    frame = json.loads(buf.getvalue())
+    assert frame["params"]["type"] == "skin.changed"
+    assert frame["params"]["session_id"] == ""
+    assert frame["params"]["payload"]["name"] == "midnight"
+
+
+def test_unregister_live_transport_stops_delivery(capture):
+    """A disconnected peer (unregistered in the ws finally block) receives nothing
+    — and a stale write is never attempted against its closed socket."""
+    server, buf = capture
+    a = _RecordingTransport()
+    server.register_live_transport(a)
+    server.unregister_live_transport(a)
+
+    server._broadcast_global_event("skin.changed", {"name": "x"})
+
+    assert a.frames == []
+    # No live transports left → fell back to stdio.
+    assert json.loads(buf.getvalue())["params"]["type"] == "skin.changed"
+
+
+def test_broadcast_global_event_survives_a_wedged_peer(capture):
+    """One broken transport must never starve the others (or the watcher thread)."""
+    server, _buf = capture
+
+    class _Boom(_RecordingTransport):
+        def write(self, obj):
+            raise RuntimeError("peer gone")
+
+    boom, good = _Boom(), _RecordingTransport()
+    server.register_live_transport(boom)
+    server.register_live_transport(good)
+
+    server._broadcast_global_event("skin.changed", {"name": "x"})
+
+    assert len(good.frames) == 1  # the healthy peer still got it
+
+
+def test_skin_change_broadcasts_to_every_connected_client(server, monkeypatch):
+    """End-to-end intent: a skin move repaints ALL connected surfaces, not just
+    the one that triggered it — the whole point of the cross-surface theme SDK."""
+    desktop, dashboard = _RecordingTransport(), _RecordingTransport()
+    server.register_live_transport(desktop)
+    server.register_live_transport(dashboard)
+
+    sigs = iter([("default", 1.0), ("synthwave", 2.0)])
+    monkeypatch.setattr(server, "_last_skin_sig", None, raising=False)
+    monkeypatch.setattr(server, "_skin_sig", lambda: next(sigs))
+    monkeypatch.setattr(server, "resolve_skin", lambda: {"name": "synthwave", "colors": {}})
+
+    server._broadcast_skin_if_changed()  # first move → default
+    server._broadcast_skin_if_changed()  # second move → synthwave
+
+    for transport in (desktop, dashboard):
+        types = [f["params"]["type"] for f in transport.frames]
+        assert types == ["skin.changed", "skin.changed"]
+        assert transport.frames[-1]["params"]["payload"]["name"] == "synthwave"

tui_gateway/server.py

@@ -1212,11 +1212,61 @@ def write_json(obj: dict) -> bool:
     return (current_transport() or _stdio_transport).write(obj)
 
 
-def _emit(event: str, sid: str, payload: dict | None = None):
-    params = {"type": event, "session_id": sid}
+def _event_frame(event: str, sid: str, payload: dict | None = None) -> dict:
+    params: dict = {"type": event, "session_id": sid}
     if payload is not None:
         params["payload"] = payload
-    write_json({"jsonrpc": "2.0", "method": "event", "params": params})
+    return {"jsonrpc": "2.0", "method": "event", "params": params}
+
+
+def _emit(event: str, sid: str, payload: dict | None = None):
+    write_json(_event_frame(event, sid, payload))
+
+
+# Live client transports, one per connected WS peer (maintained by tui_gateway.ws).
+# A session-less event from a background thread has neither a session transport
+# nor a contextvar binding, so write_json would drop it on stdio — this registry
+# is how such events reach WS clients at all. See _broadcast_global_event.
+_live_transports: set[Transport] = set()
+_live_transports_lock = threading.Lock()
+
+
+def register_live_transport(transport: Transport | None) -> None:
+    """Track a connected client transport for global broadcasts. Idempotent."""
+    if transport is None:
+        return
+    with _live_transports_lock:
+        _live_transports.add(transport)
+
+
+def unregister_live_transport(transport: Transport | None) -> None:
+    """Stop tracking a transport (call on disconnect). Idempotent."""
+    with _live_transports_lock:
+        _live_transports.discard(transport)
+
+
+def _broadcast_global_event(event: str, payload: dict | None = None) -> None:
+    """Fan a session-less, surface-global event (``skin.changed``) to every
+    connected client. Emitters like the skin watcher run on background threads
+    where ``write_json``'s ladder bottoms out at stdio and WS peers never see
+    the frame. No registered transports (stdio TUI, tests) → plain ``_emit``,
+    which that path already tees where it needs to go.
+    """
+    with _live_transports_lock:
+        targets = list(_live_transports)
+
+    if not targets:
+        _emit(event, "", payload)
+        return
+
+    frame = _event_frame(event, "", payload)
+    for transport in targets:
+        try:
+            transport.write(frame)
+        except Exception:
+            # One wedged peer must not stall the rest; disconnect teardown
+            # unregisters it.
+            logger.debug("global-event broadcast write failed type=%s", event, exc_info=True)
 
 
 _compute_host_supervisor = None
@@ -2477,7 +2527,7 @@ def _broadcast_skin_if_changed() -> None:
         return
     _last_skin_sig = sig
     try:
-        _emit("skin.changed", "", resolve_skin())
+        _broadcast_global_event("skin.changed", resolve_skin())
     except Exception:
         pass
 
@@ -12224,9 +12274,10 @@ def _(rid, params: dict) -> dict:
                 _write_config_key(f"display.{key}", value)
                 nv = value
                 if key == "skin":
-                    _emit("skin.changed", "", resolve_skin())
-                    # Keep the reconcile baseline in sync so the per-tool check
-                    # doesn't re-broadcast the skin the /skin RPC just applied.
+                    # Every connected surface repaints, not just the RPC's
+                    # client; then sync the watcher baseline so the poll loop
+                    # doesn't re-broadcast the skin this RPC just applied.
+                    _broadcast_global_event("skin.changed", resolve_skin())
                     _note_skin_broadcast()
             resp = {"key": key, "value": nv}
             if key == "personality":

tui_gateway/ws.py

@@ -329,6 +329,9 @@ async def handle_ws(ws: Any) -> None:
         if ready_ok:
             # Live-apply skins Hermes activates mid-conversation.
             server._ensure_skin_watcher()
+            # Track this peer for session-less global broadcasts (skin.changed
+            # from the background watcher) — write_json can't route those.
+            server.register_live_transport(transport)
         if not ready_ok:
             disconnect_reason = "ready_send_failed"
             send_failures += 1
@@ -429,6 +432,7 @@ async def handle_ws(ws: Any) -> None:
         reaped_sessions = 0
         detached_sessions = 0
         if transport is not None:
+            server.unregister_live_transport(transport)
             transport.close()
 
             # Reap sessions this transport owned (close_on_disconnect sidecar