From 3378e528e7c529c791a3f310541245e685532e3b Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Tue, 28 Jul 2026 15:36:37 -0500 Subject: [PATCH 1/4] feat(gateway): generalize the skin watcher into a change watcher (pet/cron/sessions broadcasts) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gateway had exactly one global broadcast (skin.changed) fed by a 0.5s signature loop. Generalize that loop into a table of cheap on-disk signatures so the process broadcasts what it already knows: - pet.changed — active pet slug/sheet-revision/scale moved (/pet, hatch) - cron.changed — cron/jobs.json mtime moved (CRUD + scheduler tick) - sessions.changed — state.db/-wal mtime moved: the cross-process signal for messaging-gateway and cron-run writes (#58671), floored to one broadcast per 2s so a streaming turn's append burst coalesces gateway.ready now carries change_events:true so clients can demote their legacy polls to slow backstops while staying compatible with older backends. Groundwork for #73618. --- tui_gateway/entry.py | 6 +- tui_gateway/server.py | 132 ++++++++++++++++++++++++++++++++++++++++-- tui_gateway/ws.py | 5 +- 3 files changed, 136 insertions(+), 7 deletions(-) diff --git a/tui_gateway/entry.py b/tui_gateway/entry.py index 2b8ccee8838a..cfd714b258b7 100644 --- a/tui_gateway/entry.py +++ b/tui_gateway/entry.py @@ -443,7 +443,11 @@ def main(): if not write_json({ "jsonrpc": "2.0", "method": "event", - "params": {"type": "gateway.ready", "payload": {"skin": resolve_skin()}}, + "params": { + "type": "gateway.ready", + # change_events: see tui_gateway/ws.py — clients demote legacy polls. + "payload": {"skin": resolve_skin(), "change_events": True}, + }, }): _log_exit("startup write failed (broken stdout pipe before first event)") sys.exit(0) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index e35e528a02c6..b88d77b4fbe6 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -2957,14 +2957,135 @@ def _broadcast_skin_if_changed() -> None: pass +def _watcher_home() -> Path: + """Active profile home for the change watcher's signature probes.""" + override = get_hermes_home_override() + return Path(override if isinstance(override, str) and override else _hermes_home) + + +def _pet_sig() -> tuple: + """(slug, spritesheet revision, scale) of the active pet — ("off",) when none. + + Cheap by construction: config comes from the mtime-cached ``_load_cfg`` and + the sheet revision is one stat. Moves when ``/pet`` (de)activates a pet, the + hatch flow rebuilds a sheet, or the scale changes.""" + display = _load_cfg().get("display") or {} + pet_cfg = display.get("pet") if isinstance(display.get("pet"), dict) else {} + if not pet_cfg or not pet_cfg.get("enabled"): + return ("off",) + try: + enabled, pet, scale = _pet_active_selection() + if not enabled or pet is None or not pet.exists: + return ("off",) + return (pet.slug, _pet_sheet_revision(pet.spritesheet), scale) + except Exception: # noqa: BLE001 - cosmetic, never break the watcher + return ("off",) + + +def _pet_changed_payload() -> dict: + """``pet.info.meta``-shaped payload for ``pet.changed`` — enough for the + renderer to decide whether the heavy sprite payload needs a refetch.""" + try: + enabled, pet, scale = _pet_active_selection() + if not enabled or pet is None or not pet.exists: + return {"enabled": False} + return { + "enabled": True, + "slug": pet.slug, + "displayName": pet.display_name, + "scale": scale, + "spritesheetRevision": _pet_sheet_revision(pet.spritesheet), + } + except Exception: # noqa: BLE001 - cosmetic, never break the watcher + return {"enabled": False} + + +def _cron_sig(): + """mtime of the profile's cron/jobs.json — moves on create/edit/pause/ + remove AND on scheduler tick bookkeeping (last_run/next_run).""" + try: + return (_watcher_home() / "cron" / "jobs.json").stat().st_mtime_ns + except OSError: + return None + + +def _sessions_sig(): + """Newest mtime across state.db and its WAL — the cross-process change + signal. Messaging-gateway turns and cron runs are written by OTHER + processes that never touch this gateway's transports; the shared SQLite + file is the one thing they all move (#58671).""" + home = _watcher_home() + sig = None + for name in ("state.db", "state.db-wal"): + try: + mtime = (home / name).stat().st_mtime_ns + except OSError: + continue + sig = mtime if sig is None else max(sig, mtime) + return sig + + +# Watched change signals: event → (check interval, signature fn, payload fn). +# Signatures are stat/dict-lookup cheap, same bar as the skin watcher; the +# check interval keeps the pricier probes (pet resolves the active sheet off +# disk) off the 0.5s tick. +_CHANGE_WATCHES: dict[str, tuple[float, Any, Any]] = { + "pet.changed": (2.0, _pet_sig, _pet_changed_payload), + "cron.changed": (1.0, _cron_sig, lambda: {}), + "sessions.changed": (0.5, _sessions_sig, lambda: {}), +} + +# state.db moves on every message append during a streaming turn; the floor +# coalesces that burst to one broadcast per window (trailing edge included — +# a floored change keeps its old signature and re-fires next tick). +_CHANGE_BROADCAST_FLOOR_S = {"sessions.changed": 2.0} + +_change_sigs: dict[str, Any] = {} +_change_checked_at: dict[str, float] = {} +_change_broadcast_at: dict[str, float] = {} + + +def _broadcast_watched_changes(now: float | None = None) -> None: + """One pass over ``_CHANGE_WATCHES``: recompute due signatures, broadcast + the events whose signature moved. First sighting seeds silently so a + gateway boot never fires a spurious refresh storm.""" + now = time.monotonic() if now is None else now + for event, (interval, sig_fn, payload_fn) in _CHANGE_WATCHES.items(): + if now - _change_checked_at.get(event, -interval) < interval: + continue + _change_checked_at[event] = now + try: + sig = sig_fn() + except Exception: # noqa: BLE001 - a broken probe must not kill the loop + continue + if event not in _change_sigs: + _change_sigs[event] = sig + continue + if sig == _change_sigs[event]: + continue + floor = _CHANGE_BROADCAST_FLOOR_S.get(event, 0.0) + if floor and now - _change_broadcast_at.get(event, -floor) < floor: + # Floored: leave the old signature in place so the change re-fires + # once the window opens (the trailing edge of the burst). + continue + _change_sigs[event] = sig + _change_broadcast_at[event] = now + try: + _broadcast_global_event(event, payload_fn()) + except Exception: # noqa: BLE001 + pass + + _skin_watcher_started = False def _ensure_skin_watcher() -> None: - """Poll the config for skin changes and broadcast ``skin.changed`` — so a skin - Hermes activates (``hermes config set display.skin``) or recolors goes live on - every surface within ~half a second, on its own, with no tool-hook or slash - command in the loop. Idempotent; started at gateway.ready.""" + """Watch cheap on-disk signatures and broadcast change events — so a skin + Hermes activates, a pet ``/pet`` adopts, a cron the scheduler fires, or a + messaging turn another process writes goes live on every surface within a + couple seconds, on its own, with no client-side poll in the loop. + Idempotent; started at gateway.ready. (Named for its original skin-only + duty; it is the process's one change watcher.)""" global _skin_watcher_started if _skin_watcher_started: return @@ -2975,8 +3096,9 @@ def _ensure_skin_watcher() -> None: while True: time.sleep(0.5) _broadcast_skin_if_changed() + _broadcast_watched_changes() - threading.Thread(target=_loop, name="hermes-skin-watcher", daemon=True).start() + threading.Thread(target=_loop, name="hermes-change-watcher", daemon=True).start() def _resolve_model() -> str: diff --git a/tui_gateway/ws.py b/tui_gateway/ws.py index 795f85c13e74..df2dcda322b4 100644 --- a/tui_gateway/ws.py +++ b/tui_gateway/ws.py @@ -309,7 +309,10 @@ async def handle_ws(ws: Any) -> None: "method": "event", "params": { "type": "gateway.ready", - "payload": {"skin": server.resolve_skin()}, + # change_events: this backend broadcasts pet.changed / + # cron.changed / sessions.changed, so clients can demote + # their legacy polls to slow backstops. + "payload": {"skin": server.resolve_skin(), "change_events": True}, }, } ) From fe64cafa33b5631da7fb911b1803da86407b0ebd Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Tue, 28 Jul 2026 15:42:46 -0500 Subject: [PATCH 2/4] =?UTF-8?q?feat(desktop):=20event-driven=20live=20sync?= =?UTF-8?q?=20=E2=80=94=20change=20broadcasts=20replace=20the=20always-on?= =?UTF-8?q?=20polls?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One controller (store/live-sync.ts, the workspace-events twin for gateway state): gateway-event.ts routes pet.changed / cron.changed / sessions.changed into tick atoms, gated on the active profile like skin.changed; gateway.ready's change_events flag arms them; a gateway wipe resets the capability. Consumers subscribe to ticks and demote their polls to slow backstops (legacy cadence is kept verbatim against older backends): - session.active_list 1.5s → sessions.changed + 30s backstop - cron list 30s → cron.changed + 5min backstop - messaging lists 10s → trailing sessions.changed refresh + legacy poll only when events are unavailable - open messaging transcript 5s → sessions.changed + 30s backstop - pet.info 3s/15s → pet.changed, NO timer on event-capable backends (users with no pet used to poll hardest); enabled=false broadcasts clear the mascot with zero round-trips - cron runs page/peek 8s → cron.changed + 60s backstop Idle renderer traffic drops from ~89 req/min to the backstops (~5/min). Part of #73618. --- .../app/chat/sidebar/cron-jobs-section.tsx | 25 ++-- .../app/contrib/hooks/use-background-sync.ts | 118 +++++++++++++++--- apps/desktop/src/app/cron/index.tsx | 27 ++-- .../hooks/use-message-stream/gateway-event.ts | 29 +++++ .../src/components/pet/floating-pet.tsx | 35 +++++- apps/desktop/src/store/gateway-switch.ts | 2 + apps/desktop/src/store/live-sync.ts | 55 ++++++++ 7 files changed, 248 insertions(+), 43 deletions(-) create mode 100644 apps/desktop/src/store/live-sync.ts diff --git a/apps/desktop/src/app/chat/sidebar/cron-jobs-section.tsx b/apps/desktop/src/app/chat/sidebar/cron-jobs-section.tsx index 86e0f39d0381..21c43cd6c7cb 100644 --- a/apps/desktop/src/app/chat/sidebar/cron-jobs-section.tsx +++ b/apps/desktop/src/app/chat/sidebar/cron-jobs-section.tsx @@ -12,6 +12,7 @@ import { useI18n } from '@/i18n' import { fmtDayTime, relativeTime } from '@/lib/time' import { cn } from '@/lib/utils' import { updateCronJobs } from '@/store/cron' +import { $changeEventsAvailable, $cronChangeTick } from '@/store/live-sync' import { notify, notifyError } from '@/store/notifications' import { $selectedStoredSessionId } from '@/store/session' import type { CronJob } from '@/types/hermes' @@ -27,9 +28,11 @@ const INACTIVE_STATES = new Set(['completed', 'disabled', 'error', 'paused']) // without turning the sidebar into the full Cron page. const PEEK_RUN_LIMIT = 5 -// Runs are written by the background scheduler tick (no UI signal), so poll the -// open peek so a freshly-fired run shows up within a few seconds. +// Runs are written by the background scheduler tick. cron.changed reloads the +// open peek immediately on event-capable backends (poll drops to a backstop); +// older backends keep the legacy cadence. const PEEK_POLL_INTERVAL_MS = 8000 +const PEEK_BACKSTOP_INTERVAL_MS = 60_000 // Keep the section compact: show a few jobs up front, reveal more in larger // steps on demand (mirrors the messaging sections in the sidebar). @@ -322,6 +325,8 @@ function CronJobSidebarRuns({ jobId, onOpenRun }: { jobId: string; onOpenRun: (s const { t } = useI18n() const c = t.cron const selectedSessionId = useStore($selectedStoredSessionId) + const changeEventsAvailable = useStore($changeEventsAvailable) + const cronChangeTick = useStore($cronChangeTick) const [runs, setRuns] = useState(null) useEffect(() => { @@ -342,17 +347,21 @@ function CronJobSidebarRuns({ jobId, onOpenRun }: { jobId: string; onOpenRun: (s void load() - const intervalId = window.setInterval(() => { - if (document.visibilityState === 'visible') { - void load() - } - }, PEEK_POLL_INTERVAL_MS) + const intervalId = window.setInterval( + () => { + if (document.visibilityState === 'visible') { + void load() + } + }, + changeEventsAvailable ? PEEK_BACKSTOP_INTERVAL_MS : PEEK_POLL_INTERVAL_MS + ) return () => { cancelled = true window.clearInterval(intervalId) } - }, [jobId]) + // cronChangeTick: a fired run reloads the peek immediately. + }, [changeEventsAvailable, cronChangeTick, jobId]) return (
diff --git a/apps/desktop/src/app/contrib/hooks/use-background-sync.ts b/apps/desktop/src/app/contrib/hooks/use-background-sync.ts index 68a802dbc91d..f176d7f7867b 100644 --- a/apps/desktop/src/app/contrib/hooks/use-background-sync.ts +++ b/apps/desktop/src/app/contrib/hooks/use-background-sync.ts @@ -1,6 +1,8 @@ +import { useStore } from '@nanostores/react' import { useEffect } from 'react' import { createClientSessionState } from '@/lib/chat-runtime' +import { $changeEventsAvailable, $cronChangeTick, $sessionsChangeTick } from '@/store/live-sync' import { refreshActiveProfile } from '@/store/profile' import { $activeSessionId, $currentCwd, setCurrentCwd } from '@/store/session' import { @@ -14,10 +16,15 @@ import type { GatewayRequester } from '../types' // Cron sessions are written by a background scheduler tick, messaging turns by // the background gateway (Telegram, WeChat, Discord, …) — neither signals the -// desktop websocket, so poll the bounded lists while the app is visible. +// desktop websocket directly. Backends with the change watcher broadcast +// `cron.changed` / `sessions.changed` when those on-disk writes land, so the +// timers below become slow safety-net backstops; against an older backend +// (no `change_events` on gateway.ready) they stay at the legacy cadence. const CRON_POLL_INTERVAL_MS = 30_000 +const CRON_BACKSTOP_INTERVAL_MS = 5 * 60_000 const MESSAGING_POLL_INTERVAL_MS = 10_000 const ACTIVE_MESSAGING_SESSION_POLL_INTERVAL_MS = 5_000 +const ACTIVE_MESSAGING_SESSION_BACKSTOP_INTERVAL_MS = 30_000 // Match the TUI's live-session refresh cadence. Auto-compression can rotate a // stored session id while its turn keeps running; until the next snapshot the // sidebar row points at the new id while the renderer still knows the old one. @@ -25,6 +32,15 @@ const ACTIVE_MESSAGING_SESSION_POLL_INTERVAL_MS = 5_000 // alarming (and clicking the row appeared to "fix" it by touching the live // session). This snapshot is small and already polled at 1.5s by the TUI. const LIVE_SESSION_STATUS_POLL_INTERVAL_MS = 1_500 +// With change events the snapshot re-pulls on every sessions.changed tick, so +// the interval only covers the degraded-socket edge the stream can't replay +// (see rehydrateLiveSessionStatuses) — 30s is plenty for that. +const LIVE_SESSION_STATUS_BACKSTOP_INTERVAL_MS = 30_000 +// Coalesce tick-driven sidebar list refreshes: sessions.changed fires (floored +// to 2s server-side) on every state.db write during a streaming turn, and the +// full list refresh is heavier than the active_list snapshot. Trailing-edge +// scheduled, so the burst's last write always lands. +const SESSIONS_LIST_TICK_GAP_MS = 10_000 interface LiveSessionStatusItem { id?: string @@ -201,6 +217,10 @@ export function useBackgroundSync({ refreshSessions, requestGateway }: BackgroundSyncParams): void { + const changeEventsAvailable = useStore($changeEventsAvailable) + const cronChangeTick = useStore($cronChangeTick) + const sessionsChangeTick = useStore($sessionsChangeTick) + useEffect(() => { if (gatewayState !== 'open') { return @@ -229,8 +249,9 @@ export function useBackgroundSync({ // A reconnect loses renderer-only working/attention atoms while the backend // keeps the actual turns alive. Re-seed from the gateway's in-memory session - // registry immediately, then cheaply poll while visible so a profile switch - // or missed reconnect edge cannot leave running rows dark until clicked. + // registry immediately, then re-pull on every sessions.changed broadcast; a + // slow visible poll remains as the backstop for the degraded-socket edge the + // stream cannot replay (legacy cadence against older backends). useEffect(() => { if (gatewayState !== 'open') { return @@ -260,7 +281,10 @@ export function useBackgroundSync({ } } - const dispose = visiblePoll(LIVE_SESSION_STATUS_POLL_INTERVAL_MS, () => void refreshLiveStatuses()) + const dispose = visiblePoll( + changeEventsAvailable ? LIVE_SESSION_STATUS_BACKSTOP_INTERVAL_MS : LIVE_SESSION_STATUS_POLL_INTERVAL_MS, + () => void refreshLiveStatuses() + ) void refreshLiveStatuses() @@ -268,44 +292,98 @@ export function useBackgroundSync({ cancelled = true dispose() } - }, [activeGatewayProfile, gatewayState, requestGateway]) + // sessionsChangeTick: each sessions.changed broadcast re-seeds immediately + // via the effect re-run (already coalesced to 2s server-side). + }, [activeGatewayProfile, changeEventsAvailable, gatewayState, requestGateway, sessionsChangeTick]) + + // sessions.changed also means the *stored* list may have new rows (a cron + // run's session, an inbound messaging turn creating a thread). The full list + // refresh is heavier than the active_list snapshot, so trail it on a gap + // instead of firing per tick. Direct atom subscription: the throttle state + // lives in the effect closure, not in refs synced from renders. + useEffect(() => { + if (gatewayState !== 'open' || !changeEventsAvailable) { + return + } + + let lastRunAt = 0 + let timer: null | number = null + + const run = () => { + lastRunAt = Date.now() + void refreshSessions() + void refreshMessagingSessions() + } + + const unsubscribe = $sessionsChangeTick.listen(() => { + const since = Date.now() - lastRunAt + + if (since >= SESSIONS_LIST_TICK_GAP_MS) { + run() + } else if (timer === null) { + timer = window.setTimeout(() => { + timer = null + run() + }, SESSIONS_LIST_TICK_GAP_MS - since) + } + }) + + return () => { + unsubscribe() + + if (timer !== null) { + window.clearTimeout(timer) + } + } + }, [changeEventsAvailable, gatewayState, refreshMessagingSessions, refreshSessions]) // Keep the cron-jobs section live without a user action (scheduler ticks in - // the background); re-check on tab re-focus too. + // the background). cron.changed (jobs.json moved: CRUD or a scheduler tick's + // bookkeeping) drives the refresh; the visible poll is the backstop. useEffect(() => { if (gatewayState !== 'open') { return } - return visiblePoll(CRON_POLL_INTERVAL_MS, () => void refreshCronJobs()) - }, [gatewayState, refreshCronJobs]) - - // Keep the messaging-platform session lists live (inbound turns are written - // by the gateway, not the desktop websocket). - useEffect(() => { - if (gatewayState !== 'open') { - return + if (cronChangeTick > 0) { + void refreshCronJobs() } - return visiblePoll(MESSAGING_POLL_INTERVAL_MS, () => void refreshMessagingSessions()) - }, [gatewayState, refreshMessagingSessions]) + return visiblePoll( + changeEventsAvailable ? CRON_BACKSTOP_INTERVAL_MS : CRON_POLL_INTERVAL_MS, + () => void refreshCronJobs() + ) + }, [changeEventsAvailable, cronChangeTick, gatewayState, refreshCronJobs]) - // Only the open messaging transcript needs its own poll — local chats are - // live over the websocket already. + // Only the open messaging transcript needs its own cadence — local chats are + // live over the websocket already. sessions.changed re-pulls it via the tick + // dep; the visible poll is the backstop. useEffect(() => { if (gatewayState !== 'open' || !activeIsMessaging) { return } const dispose = visiblePoll( - ACTIVE_MESSAGING_SESSION_POLL_INTERVAL_MS, + changeEventsAvailable ? ACTIVE_MESSAGING_SESSION_BACKSTOP_INTERVAL_MS : ACTIVE_MESSAGING_SESSION_POLL_INTERVAL_MS, () => void refreshActiveMessagingTranscript() ) void refreshActiveMessagingTranscript() return dispose - }, [activeIsMessaging, gatewayState, refreshActiveMessagingTranscript]) + // sessionsChangeTick: an inbound turn re-pulls the open transcript. + }, [activeIsMessaging, changeEventsAvailable, gatewayState, refreshActiveMessagingTranscript, sessionsChangeTick]) + + // Messaging session lists against an older backend: no sessions.changed, so + // keep the legacy visible poll. (Event-capable backends fold this into the + // trailing sessions.changed refresh above.) + useEffect(() => { + if (gatewayState !== 'open' || changeEventsAvailable) { + return + } + + return visiblePoll(MESSAGING_POLL_INTERVAL_MS, () => void refreshMessagingSessions()) + }, [changeEventsAvailable, gatewayState, refreshMessagingSessions]) // A fresh new-session draft (gateway open, no active session) re-pulls the // model + config so the composer pill reflects the profile default. diff --git a/apps/desktop/src/app/cron/index.tsx b/apps/desktop/src/app/cron/index.tsx index a8c74e2a9c66..0a0eb418d2b0 100644 --- a/apps/desktop/src/app/cron/index.tsx +++ b/apps/desktop/src/app/cron/index.tsx @@ -48,6 +48,7 @@ import { AlertTriangle } from '@/lib/icons' import { requestModelOptions } from '@/lib/model-options' import { asText } from '@/lib/text' import { $cronFocusJobId, $cronJobs, setCronFocusJobId, setCronJobs, updateCronJobs } from '@/store/cron' +import { $changeEventsAvailable, $cronChangeTick } from '@/store/live-sync' import { notify, notifyError } from '@/store/notifications' import { $profileScope, ALL_PROFILES } from '@/store/profile' @@ -672,10 +673,12 @@ function formatRunTime(seconds?: null | number): string { return Number.isNaN(date.valueOf()) ? '—' : date.toLocaleString() } -// Runs are produced by the background scheduler tick (no UI signal), so poll -// while the panel is open + on tab re-focus so a fired run shows up within a few -// seconds instead of waiting for a reload. +// Runs are produced by the background scheduler tick. cron.changed / +// sessions.changed broadcasts re-load immediately on event-capable backends +// (the tick dep below), so the poll drops to a slow backstop there; older +// backends keep the legacy cadence. const RUNS_POLL_INTERVAL_MS = 8000 +const RUNS_BACKSTOP_INTERVAL_MS = 60_000 function CronJobRuns({ c, @@ -687,6 +690,8 @@ function CronJobRuns({ onOpenSession?: (sessionId: string) => void }) { const [runs, setRuns] = useState(null) + const changeEventsAvailable = useStore($changeEventsAvailable) + const cronChangeTick = useStore($cronChangeTick) useEffect(() => { let cancelled = false @@ -706,11 +711,14 @@ function CronJobRuns({ void load() - const intervalId = window.setInterval(() => { - if (document.visibilityState === 'visible') { - void load() - } - }, RUNS_POLL_INTERVAL_MS) + const intervalId = window.setInterval( + () => { + if (document.visibilityState === 'visible') { + void load() + } + }, + changeEventsAvailable ? RUNS_BACKSTOP_INTERVAL_MS : RUNS_POLL_INTERVAL_MS + ) const onVisible = () => { if (document.visibilityState === 'visible') { @@ -725,7 +733,8 @@ function CronJobRuns({ window.clearInterval(intervalId) document.removeEventListener('visibilitychange', onVisible) } - }, [jobId]) + // cronChangeTick: a fired run moves jobs.json bookkeeping → reload now. + }, [changeEventsAvailable, cronChangeTick, jobId]) return (
diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts index b90f57e282a1..dfdbdff87b64 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts @@ -24,6 +24,13 @@ import { setSessionCompacting } from '@/store/compaction' import { refreshBackgroundProcesses } from '@/store/composer-status' import { $gateway } from '@/store/gateway' import { applyGoalStatusText } from '@/store/goals' +import { + notifyCronChanged, + notifyPetChanged, + notifySessionsChanged, + type PetChangeMeta, + setChangeEventsAvailable +} from '@/store/live-sync' import { dispatchNativeNotification } from '@/store/native-notifications' import { notify } from '@/store/notifications' import { requestDesktopOnboarding, requestDesktopOnboardingForCredentialWarning } from '@/store/onboarding' @@ -275,6 +282,9 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { // Seed the active skin into the desktop theme registry without applying, // so a fresh connect never overrides the user's persisted desktop theme. ingestBackendSkin((payload as { skin?: HermesSkin } | undefined)?.skin, { apply: false }) + // Backends with the change watcher broadcast pet/cron/sessions change + // events; consumers demote their legacy polls to slow backstops. + setChangeEventsAvailable(Boolean((payload as { change_events?: boolean } | undefined)?.change_events)) return } else if (event.type === 'skin.changed') { @@ -287,6 +297,25 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { ingestBackendSkin(payload as HermesSkin | undefined, { apply: true }) } + return + } else if (event.type === 'pet.changed' || event.type === 'cron.changed' || event.type === 'sessions.changed') { + // Change-watcher broadcasts (server._broadcast_watched_changes): the + // backend's on-disk signature moved. Route to the live-sync ticks the + // former pollers now subscribe to. Only the active profile's changes + // apply — background profile sockets watch their own homes. + const fromActiveChangeProfile = + !event.profile || normalizeProfileKey(event.profile) === normalizeProfileKey($activeGatewayProfile.get()) + + if (fromActiveChangeProfile) { + if (event.type === 'pet.changed') { + notifyPetChanged(payload as PetChangeMeta | undefined) + } else if (event.type === 'cron.changed') { + notifyCronChanged() + } else { + notifySessionsChanged() + } + } + return } else if (event.type === 'session.info') { // Apply session-scoped fields when the event targets the active diff --git a/apps/desktop/src/components/pet/floating-pet.tsx b/apps/desktop/src/components/pet/floating-pet.tsx index 4a14991bf39c..4175427150dc 100644 --- a/apps/desktop/src/components/pet/floating-pet.tsx +++ b/apps/desktop/src/components/pet/floating-pet.tsx @@ -6,6 +6,7 @@ import { useOnProfileSwitch } from '@/app/hooks/use-on-profile-switch' import { useRouteOverlayActive } from '@/app/hooks/use-route-overlay-active' import { PetHeartField } from '@/components/chat/vibe-hearts' import { persistString, storedString } from '@/lib/storage' +import { $changeEventsAvailable, $petChange } from '@/store/live-sync' import { $petAtRest, $petInfo, @@ -115,6 +116,8 @@ export function FloatingPet() { const { resolvedMode } = useTheme() const gatewayState = useStore($gatewayState) const info = useStore($petInfo) + const changeEventsAvailable = useStore($changeEventsAvailable) + const petChange = useStore($petChange) const overlayActive = useStore($petOverlayActive) const roamEnabled = useStore($petRoam) const atRest = useStore($petAtRest) @@ -142,9 +145,11 @@ export function FloatingPet() { // edge can't leave the window cropping it. Shared by drag + the reclamp effect. const clamp = useCallback(({ x, y }: Point): Point => clampPoint(x, y, petW, petH), [petW, petH]) - // Fetch pet.info on connect. Poll quickly while inactive so an in-app - // `/pet ` appears, then slowly while active so regenerated spritesheets - // and row-count metadata replace the cached base64 payload. + // Fetch pet.info on connect, then let pet.changed drive refreshes: the + // change watcher broadcasts when /pet (de)activates a pet or the hatch flow + // rewrites a sheet, so event-capable backends need no interval at all — + // users with no pet especially (this used to poll hardest for them). Older + // backends keep the legacy fast-while-inactive poll. const active = info.enabled && Boolean(info.spritesheetBase64) useEffect(() => { if (gatewayState !== 'open') { @@ -153,6 +158,16 @@ export function FloatingPet() { let cancelled = false + // pet.changed already carries the meta payload — an enabled=false + // broadcast clears the mascot with zero round-trips, and an unchanged + // revision (scale-only move still changes the sig) short-circuits below + // via samePetRevision. + if (changeEventsAvailable && petChange.tick > 0 && petChange.meta?.enabled === false) { + setPetInfo({ enabled: false }) + + return + } + const pull = async () => { try { if (active) { @@ -202,15 +217,23 @@ export function FloatingPet() { } void pull() - const timer = window.setInterval(() => void pull(), active ? PET_ACTIVE_REFRESH_MS : PET_POLL_MS) window.addEventListener('focus', pull) + // Event-capable backend: pet.changed re-runs this effect (petChange dep), + // so no timer. Legacy backend: the historical poll. + const timer = changeEventsAvailable + ? null + : window.setInterval(() => void pull(), active ? PET_ACTIVE_REFRESH_MS : PET_POLL_MS) + return () => { cancelled = true window.removeEventListener('focus', pull) - window.clearInterval(timer) + + if (timer !== null) { + window.clearInterval(timer) + } } - }, [gatewayState, active, requestGateway]) + }, [gatewayState, active, changeEventsAvailable, petChange, requestGateway]) // Pets are per-profile. When the active profile changes, drop the previous // profile's mascot + gallery cache so the poll above refetches the new diff --git a/apps/desktop/src/store/gateway-switch.ts b/apps/desktop/src/store/gateway-switch.ts index 4811ad3a639e..cbcd2d03e87d 100644 --- a/apps/desktop/src/store/gateway-switch.ts +++ b/apps/desktop/src/store/gateway-switch.ts @@ -5,6 +5,7 @@ import { resetSidebarBatchCapability } from '@/hermes' import { invalidateProfileScopedQueries } from '@/lib/query-client' import { clearArtifactRegistry } from '@/store/artifacts' import { resetSessionsLimit } from '@/store/layout' +import { resetLiveSync } from '@/store/live-sync' import { $unreadFinishedSessionIds, setActiveSessionId, @@ -53,6 +54,7 @@ export function wipeSessionListsForGatewaySwitch(): void { // $unreadFinishedSessionIds is separate, so wipe it explicitly. clearAllSessionStates() resetLiveRuntimeTracking() + resetLiveSync() $unreadFinishedSessionIds.set([]) setSessionsLoading(true) resetSessionsLimit() diff --git a/apps/desktop/src/store/live-sync.ts b/apps/desktop/src/store/live-sync.ts new file mode 100644 index 000000000000..2464eb96dbd3 --- /dev/null +++ b/apps/desktop/src/store/live-sync.ts @@ -0,0 +1,55 @@ +import { atom } from 'nanostores' + +// Event-driven "backend data changed" signals — the workspace-events twin for +// gateway-side state. The change watcher in tui_gateway broadcasts +// `pet.changed` / `cron.changed` / `sessions.changed` when its on-disk +// signatures move (see server._broadcast_watched_changes), gateway-event.ts +// routes them here, and the surfaces that used to poll (cron sidebar/page, +// messaging lists, pet sprite, live session statuses) subscribe to these ticks +// and refresh — so they move exactly when the backend acts and stay idle +// otherwise. +// +// `$changeEventsAvailable` is the compat gate: gateway.ready advertises +// `change_events: true` on backends that broadcast. Consumers keep their old +// polls as SLOW backstops when it's true and at the legacy cadence when it's +// false (an older backend), so nothing goes dark mid-upgrade. + +export const $changeEventsAvailable = atom(false) + +export const $cronChangeTick = atom(0) +export const $sessionsChangeTick = atom(0) + +/** `pet.info.meta`-shaped payload carried on `pet.changed` — lets the pet skip + * the heavy sprite refetch when the broadcast already says enabled=false. */ +export interface PetChangeMeta { + enabled: boolean + slug?: string + displayName?: string + scale?: number + spritesheetRevision?: string +} + +export const $petChange = atom<{ meta?: PetChangeMeta; tick: number }>({ tick: 0 }) + +export function setChangeEventsAvailable(available: boolean): void { + $changeEventsAvailable.set(available) +} + +export function notifyPetChanged(meta?: PetChangeMeta): void { + $petChange.set({ meta, tick: $petChange.get().tick + 1 }) +} + +export function notifyCronChanged(): void { + $cronChangeTick.set($cronChangeTick.get() + 1) +} + +export function notifySessionsChanged(): void { + $sessionsChangeTick.set($sessionsChangeTick.get() + 1) +} + +/** Reset on gateway wipe/reconnect — a new backend re-advertises capability on + * its own gateway.ready, and stale ticks must not fire refreshes into stores + * the wipe just cleared. */ +export function resetLiveSync(): void { + $changeEventsAvailable.set(false) +} From c14ae3796e31eed04afbe1bfa22376f7bfe693c3 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Tue, 28 Jul 2026 15:45:45 -0500 Subject: [PATCH 3/4] feat(desktop): fs-watch the plugin dir + demote the status snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - watchDirectory IPC (same registry/channel as the preview file watchers) replaces the disk-plugin door's 5s readdir poll; older shells without the capability keep the poll, which self-upgrades to the watch once the plugins dir exists. - Status snapshot: 15s → 60s, skips round-trips while hidden, and refreshes immediately on visibilitychange so re-focus never shows stale health. Part of #73618. --- apps/desktop/electron/main.ts | 41 +++++++++++++ apps/desktop/electron/preload.ts | 1 + .../app/shell/hooks/use-status-snapshot.ts | 26 +++++++- apps/desktop/src/contrib/runtime-loader.ts | 61 ++++++++++++++++--- apps/desktop/src/global.d.ts | 4 ++ 5 files changed, 124 insertions(+), 9 deletions(-) diff --git a/apps/desktop/electron/main.ts b/apps/desktop/electron/main.ts index 4111ed4721f6..59bc792c7c6d 100644 --- a/apps/desktop/electron/main.ts +++ b/apps/desktop/electron/main.ts @@ -4927,6 +4927,45 @@ function closePreviewWatchers() { } } +/** Watch a DIRECTORY for entry churn (folders appearing/vanishing) — the + * disk-plugin door's "new plugin folder" signal, replacing the renderer's 5s + * readdir poll. Same registry + change channel as the preview file watchers + * (the renderer reconciles on any tick; per-file edits stay on their own + * watches), so stopPreviewFileWatch/closePreviewWatchers manage these too. */ +function watchDirectory(rawDir) { + const watchDir = path.resolve(String(rawDir || '')) + + if (!fs.existsSync(watchDir) || !fs.statSync(watchDir).isDirectory()) { + throw new Error(`Not a directory: ${watchDir}`) + } + + const id = crypto.randomBytes(12).toString('base64url') + let timer = null + + const watcher = fs.watch(watchDir, () => { + if (timer) { + clearTimeout(timer) + } + + timer = setTimeout(() => { + timer = null + sendPreviewFileChanged({ id, path: watchDir, url: pathToFileURL(watchDir).toString() }) + }, PREVIEW_WATCH_DEBOUNCE_MS) + }) + + previewWatchers.set(id, { + close: () => { + if (timer) { + clearTimeout(timer) + } + + watcher.close() + } + }) + + return { id, path: watchDir } +} + // Best-effort read of a gateway's advertised auth providers, cached per base // URL for the life of the process. Used by the oauth pre-flight guard to tell // a password-provider gateway (which cannot satisfy the bearer/cookie checks @@ -10273,6 +10312,8 @@ ipcMain.handle('hermes:normalizePreviewTarget', (_event, target, baseDir) => ipcMain.handle('hermes:watchPreviewFile', (_event, url) => watchPreviewFile(String(url || ''))) +ipcMain.handle('hermes:watchDirectory', (_event, dir) => watchDirectory(String(dir || ''))) + ipcMain.handle('hermes:stopPreviewFileWatch', (_event, id) => stopPreviewFileWatch(String(id || ''))) // Each renderer reports the turns it has in flight; the quit guard reads the diff --git a/apps/desktop/electron/preload.ts b/apps/desktop/electron/preload.ts index 815f048359b3..b110eb1670b7 100644 --- a/apps/desktop/electron/preload.ts +++ b/apps/desktop/electron/preload.ts @@ -117,6 +117,7 @@ contextBridge.exposeInMainWorld('hermesDesktop', { }, normalizePreviewTarget: (target, baseDir) => ipcRenderer.invoke('hermes:normalizePreviewTarget', target, baseDir), watchPreviewFile: url => ipcRenderer.invoke('hermes:watchPreviewFile', url), + watchDirectory: dir => ipcRenderer.invoke('hermes:watchDirectory', dir), stopPreviewFileWatch: id => ipcRenderer.invoke('hermes:stopPreviewFileWatch', id), setActiveWork: payload => ipcRenderer.send('hermes:active-work', payload), setTitleBarTheme: payload => ipcRenderer.send('hermes:titlebar-theme', payload), diff --git a/apps/desktop/src/app/shell/hooks/use-status-snapshot.ts b/apps/desktop/src/app/shell/hooks/use-status-snapshot.ts index 7c531d9cf811..d2b714904b8b 100644 --- a/apps/desktop/src/app/shell/hooks/use-status-snapshot.ts +++ b/apps/desktop/src/app/shell/hooks/use-status-snapshot.ts @@ -4,7 +4,11 @@ import { getStatus } from '@/hermes' import { evaluateRuntimeReadiness, type RuntimeReadinessResult } from '@/lib/runtime-readiness' import type { StatusResponse } from '@/types/hermes' -const REFRESH_MS = 15_000 +// Statusbar health is ambient chrome, not live data — nothing the user acts on +// within seconds. 60s + a hidden-tab skip keeps it honest at a quarter of the +// old traffic; the visibility listener refreshes immediately on return so a +// backgrounded window never shows stale health after re-focus. +const REFRESH_MS = 60_000 type GatewayRequester = (method: string, params?: Record) => Promise @@ -30,6 +34,14 @@ export function useStatusSnapshot(gatewayState: string | undefined, requestGatew } const refresh = async () => { + // Hidden window: skip the round-trips, keep the timer alive; the + // visibilitychange listener repaints immediately on return. + if (document.visibilityState !== 'visible') { + scheduleRefresh() + + return + } + try { // Wait for both legs before scheduling the next refresh. setInterval // allowed a slow runtime check to overlap with later polls, which @@ -67,10 +79,22 @@ export function useStatusSnapshot(gatewayState: string | undefined, requestGatew } } + const onVisible = () => { + if (document.visibilityState === 'visible' && !cancelled) { + if (timer !== undefined) { + window.clearTimeout(timer) + } + + void refresh() + } + } + + document.addEventListener('visibilitychange', onVisible) void refresh() return () => { cancelled = true + document.removeEventListener('visibilitychange', onVisible) if (timer !== undefined) { window.clearTimeout(timer) diff --git a/apps/desktop/src/contrib/runtime-loader.ts b/apps/desktop/src/contrib/runtime-loader.ts index 11545d5a996d..8ac7c60d63fd 100644 --- a/apps/desktop/src/contrib/runtime-loader.ts +++ b/apps/desktop/src/contrib/runtime-loader.ts @@ -183,8 +183,9 @@ export async function loadRuntimePlugin( // (agent- or user-written). SELF-MAINTAINING — no reload ceremony: // - each plugin.js is fs-watched (the preview watcher IPC, debounced in // main): saving the file hot-reloads the plugin in place; -// - a slow visible-tab poll of the directory picks up new folders (load + -// watch) and removed ones (unload + unwatch). +// - the directory itself is fs-watched too (watchDirectory IPC), so new +// folders load + removed ones unload on the change tick; older Electron +// shells without watchDirectory fall back to the slow visible-tab poll. // Panes land via placement adoption and STAY where the user drags them — // the tree treats not-yet-loaded pane ids as hidden, so boot and reload are // collapse -> appear, never a placeholder flash. @@ -307,7 +308,7 @@ async function scanDiskPlugins(): Promise { export const discoverRuntimePlugins = scanDiskPlugins /** Start the self-maintaining disk door: initial scan, per-file hot reload, - * slow folder reconciliation while the window is visible. Idempotent. */ + * fs-watched folder reconciliation (poll fallback on older shells). Idempotent. */ export function watchRuntimePlugins(): void { const desktop = window.hermesDesktop @@ -317,7 +318,16 @@ export function watchRuntimePlugins(): void { watching = true + let dirWatchId: null | string = null + desktop.onPreviewFileChanged(({ id }) => { + // Directory tick: a plugin folder appeared or vanished — reconcile. + if (dirWatchId && id === dirWatchId) { + void scanDiskPlugins() + + return + } + for (const [name, record] of disk) { if (record.watchId === id) { void loadDiskPlugin(name, record.file) @@ -327,10 +337,45 @@ export function watchRuntimePlugins(): void { } }) - void scanDiskPlugins() - window.setInterval(() => { - if (document.visibilityState === 'visible') { - void scanDiskPlugins() + const startDirWatch = async (): Promise => { + if (!desktop.watchDirectory) { + return false } - }, DISK_POLL_MS) + + try { + const { hermes_home } = await getStatus() + dirWatchId = (await desktop.watchDirectory(`${hermes_home}/desktop-plugins`)).id + + return true + } catch { + // Dir missing (no plugins yet) or unwatchable — fall back to the poll, + // which also handles the dir being created later. + return false + } + } + + void scanDiskPlugins() + void startDirWatch().then(watched => { + if (watched) { + return + } + + const timer = window.setInterval(() => { + if (document.visibilityState !== 'visible') { + return + } + + void scanDiskPlugins() + + // The dir may have been created since — upgrade to the watch and retire + // this poll once it lands. + if (dirWatchId === null) { + void startDirWatch().then(upgraded => { + if (upgraded) { + window.clearInterval(timer) + } + }) + } + }, DISK_POLL_MS) + }) } diff --git a/apps/desktop/src/global.d.ts b/apps/desktop/src/global.d.ts index 80c8104376e5..4a7eea4dbd8e 100644 --- a/apps/desktop/src/global.d.ts +++ b/apps/desktop/src/global.d.ts @@ -128,6 +128,10 @@ declare global { getPathForFile: (file: File) => string normalizePreviewTarget: (target: string, baseDir?: string) => Promise watchPreviewFile: (url: string) => Promise + /** Watch a directory for entry churn (disk-plugin door); same watcher + * registry + onPreviewFileChanged channel as watchPreviewFile. Optional: + * older Electron shells predate it and fall back to the readdir poll. */ + watchDirectory?: (dir: string) => Promise stopPreviewFileWatch: (id: string) => Promise setActiveWork?: (payload: HermesActiveWork) => void setTitleBarTheme?: (payload: HermesTitleBarTheme) => void From fa8fb82d0e47f9b0665018a2061a32fba0a721d5 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Tue, 28 Jul 2026 17:03:14 -0500 Subject: [PATCH 4/4] test(gateway): change-watcher behavior contracts; align status-snapshot test with 60s cadence Real temp-HERMES_HOME tests for _broadcast_watched_changes: silent seed, cron/sessions signature moves, the 2s sessions floor's trailing edge, the pet signature staying 'off' without a renderable pet, meta payload on pet.changed, and a broken probe never killing the pass. Part of #73618. --- .../shell/hooks/use-status-snapshot.test.ts | 2 +- tests/tui_gateway/test_change_watcher.py | 135 ++++++++++++++++++ 2 files changed, 136 insertions(+), 1 deletion(-) create mode 100644 tests/tui_gateway/test_change_watcher.py diff --git a/apps/desktop/src/app/shell/hooks/use-status-snapshot.test.ts b/apps/desktop/src/app/shell/hooks/use-status-snapshot.test.ts index 9ffbfe4b0a51..1c05ac05822a 100644 --- a/apps/desktop/src/app/shell/hooks/use-status-snapshot.test.ts +++ b/apps/desktop/src/app/shell/hooks/use-status-snapshot.test.ts @@ -158,7 +158,7 @@ describe('useStatusSnapshot', () => { }) await act(async () => { - await vi.advanceTimersByTimeAsync(14_999) + await vi.advanceTimersByTimeAsync(59_999) }) expect(requestGatewayMock).toHaveBeenCalledTimes(2) diff --git a/tests/tui_gateway/test_change_watcher.py b/tests/tui_gateway/test_change_watcher.py new file mode 100644 index 000000000000..275e54127fe8 --- /dev/null +++ b/tests/tui_gateway/test_change_watcher.py @@ -0,0 +1,135 @@ +"""The generalized change watcher (#73618): cheap on-disk signatures → +``pet.changed`` / ``cron.changed`` / ``sessions.changed`` global broadcasts. + +Behavior contracts, exercised against a real temp HERMES_HOME (no mocks on the +filesystem path): first sighting seeds silently, a moved signature broadcasts +once, the sessions floor coalesces a write burst but keeps its trailing edge, +and the pet signature only moves for a *renderable* pet. +""" + +import time + +import pytest + +from tui_gateway import server + + +@pytest.fixture() +def watcher_home(tmp_path, monkeypatch): + (tmp_path / "config.yaml").write_text("display: {}\n") + (tmp_path / "cron").mkdir() + + monkeypatch.setattr(server, "_hermes_home", str(tmp_path)) + monkeypatch.setattr(server, "_cfg_cache", None) + monkeypatch.setattr(server, "_change_sigs", {}) + monkeypatch.setattr(server, "_change_checked_at", {}) + monkeypatch.setattr(server, "_change_broadcast_at", {}) + + events = [] + monkeypatch.setattr( + server, "_broadcast_global_event", lambda ev, payload=None: events.append((ev, payload)) + ) + return tmp_path, events + + +def test_first_sighting_seeds_without_broadcasting(watcher_home): + home, events = watcher_home + (home / "cron" / "jobs.json").write_text("[]") + (home / "state.db").write_text("x") + + server._broadcast_watched_changes(now=0.0) + + assert events == [] + + +def test_cron_jobs_file_move_broadcasts_cron_changed(watcher_home): + home, events = watcher_home + server._broadcast_watched_changes(now=0.0) + + (home / "cron" / "jobs.json").write_text("[]") + server._broadcast_watched_changes(now=10.0) + + assert ("cron.changed", {}) in events + + +def test_state_db_move_broadcasts_sessions_changed(watcher_home): + home, events = watcher_home + server._broadcast_watched_changes(now=0.0) + + (home / "state.db").write_text("x") + server._broadcast_watched_changes(now=10.0) + + assert ("sessions.changed", {}) in events + + +def test_sessions_floor_coalesces_burst_but_keeps_trailing_edge(watcher_home): + home, events = watcher_home + server._broadcast_watched_changes(now=0.0) + + (home / "state.db").write_text("x") + server._broadcast_watched_changes(now=10.0) + events.clear() + + # A second write lands inside the 2s floor: no broadcast yet… + time.sleep(0.02) + (home / "state.db").write_text("xy") + server._broadcast_watched_changes(now=11.0) + assert events == [] + + # …but the change is not lost — it fires once the window opens. + server._broadcast_watched_changes(now=13.0) + assert ("sessions.changed", {}) in events + + +def test_pet_sig_stays_off_without_a_renderable_pet(watcher_home): + home, events = watcher_home + server._broadcast_watched_changes(now=0.0) + + # Config flips enabled but no pet exists on disk → signature stays ("off",). + (home / "config.yaml").write_text("display:\n pet:\n enabled: true\n slug: boba\n") + server._cfg_cache = None + server._broadcast_watched_changes(now=10.0) + + assert not [e for e in events if e[0] == "pet.changed"] + + +def test_renderable_pet_broadcasts_meta_payload(watcher_home, monkeypatch): + home, events = watcher_home + (home / "config.yaml").write_text("display:\n pet:\n enabled: true\n slug: boba\n") + server._cfg_cache = None + server._broadcast_watched_changes(now=0.0) + + sheet = home / "sheet.png" + sheet.write_text("png") + + class FakePet: + slug = "boba" + display_name = "Boba" + exists = True + spritesheet = sheet + + monkeypatch.setattr(server, "_pet_active_selection", lambda: (True, FakePet(), 0.33)) + server._broadcast_watched_changes(now=10.0) + + pet_events = [e for e in events if e[0] == "pet.changed"] + assert pet_events + payload = pet_events[0][1] + assert payload["enabled"] is True + assert payload["slug"] == "boba" + assert payload["spritesheetRevision"] + + +def test_broken_probe_never_kills_the_pass(watcher_home, monkeypatch): + home, events = watcher_home + server._broadcast_watched_changes(now=0.0) + + monkeypatch.setitem( + server._CHANGE_WATCHES, + "cron.changed", + (1.0, lambda: (_ for _ in ()).throw(RuntimeError("boom")), lambda: {}), + ) + (home / "state.db").write_text("x") + server._broadcast_watched_changes(now=10.0) + + # The broken cron probe is skipped; sessions still broadcasts. + assert ("sessions.changed", {}) in events