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 86e0f39d038..21c43cd6c7c 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 68a802dbc91..f176d7f7867 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 a8c74e2a9c6..0a0eb418d2b 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 b90f57e282a..dfdbdff87b6 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 4a14991bf39..4175427150d 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 4811ad3a639..cbcd2d03e87 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 00000000000..2464eb96dbd --- /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) +}