diff --git a/webui/backend/app/api/events.py b/webui/backend/app/api/events.py new file mode 100644 index 000000000..73cb3158c --- /dev/null +++ b/webui/backend/app/api/events.py @@ -0,0 +1,32 @@ +"""Server-sent stream of change notices for the app shell. + +The frontend opens one long-lived GET /api/events (EventSource) per tab. Every +notice published on the in-process `event_bus` — one per successful management +write — is forwarded here as an SSE `change` event, so a mutation made by ANY +caller (including an external script) refreshes every open page immediately. + +Not enveloped: this is a stream, so the router deliberately omits EnvelopeRoute +(which buffers and re-wraps JSON bodies). +""" +import json + +from fastapi import APIRouter +from sse_starlette.sse import EventSourceResponse + +from app.core.events import event_bus + +router = APIRouter(prefix="/api", tags=["events"]) + + +@router.get("/events") +async def events(): + # `@ant-design/x-sdk` is not involved here (native EventSource consumes this), + # but keep LF frame separation consistent with the chat stream. sse-starlette + # sends periodic pings on its own, which both keep the connection alive and + # detect a dropped client so the subscriber generator is cancelled. + async def stream(): + yield {"event": "ready"} + async for event in event_bus.subscribe(): + yield {"event": "change", "data": json.dumps(event)} + + return EventSourceResponse(stream(), sep="\n") diff --git a/webui/backend/app/core/envelope.py b/webui/backend/app/core/envelope.py index 8514bcd6a..e783ebfb7 100644 --- a/webui/backend/app/core/envelope.py +++ b/webui/backend/app/core/envelope.py @@ -41,6 +41,28 @@ def error_response(status_code: int, message: str, *, ) +_CHANGE_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"}) +# Successful writes under these prefixes must NOT broadcast: the presence poll +# would feed back into itself (every browser would refresh on every heartbeat), +# the chat stream/control own their own SSE, and recovery runs before the app is +# even usable. Everything else that mutates state notifies the open pages. +_NO_BROADCAST_PREFIXES = ("/api/presence", "/api/chat", "/api/recovery", "/api/events") + + +def _broadcast_change(request: Request, status: int) -> None: + """Announce a successful management write so every open page can refresh. + Carries the path and method; the frontend maps them to the lists to reload.""" + if request.method not in _CHANGE_METHODS or not (200 <= status < 300): + return + path = request.url.path + if any(path.startswith(p) for p in _NO_BROADCAST_PREFIXES): + return + # Imported lazily so this core module stays free of app-package import order + # concerns; publishing never blocks or raises. + from app.core.events import event_bus + event_bus.publish({"path": path, "method": request.method}) + + class EnvelopeRoute(APIRoute): """Wraps a route's serialized success payload into the standard envelope. @@ -68,6 +90,7 @@ async def custom(request: Request) -> Response: # Preserve the RESTful status code (e.g. 201 Created). A 204 becomes # 200 since the envelope now carries a body. status = 200 if response.status_code == 204 else response.status_code + _broadcast_change(request, status) return success_response(data, status_code=status) return custom diff --git a/webui/backend/app/core/events.py b/webui/backend/app/core/events.py new file mode 100644 index 000000000..d3cb05355 --- /dev/null +++ b/webui/backend/app/core/events.py @@ -0,0 +1,61 @@ +"""Process-local pub/sub for pushing "something changed" notices to the UI. + +A single-process broadcast hub: management endpoints publish a change notice +after a successful write, and every connected browser (subscribed via +GET /api/events) receives it and refreshes the affected lists. This is what +lets a change made by an EXTERNAL caller (a script hitting the REST API) reach +an already-open page, which the in-tab event bus on the frontend cannot do. + +Single-process only: subscribers live in one event loop's memory. A multi-worker +deployment would need a cross-process channel (e.g. Redis pub/sub) behind the +same `publish`/`subscribe` surface. +""" +from __future__ import annotations + +import asyncio +from typing import Any, AsyncIterator + +# Per-subscriber buffer. Change notices are tiny and rare; this only bounds a +# subscriber that has stopped reading (a wedged connection) so it cannot grow +# without limit. On overflow the oldest notice is dropped for the LATEST, since +# every notice triggers the same "refresh" and the freshest one wins. +_QUEUE_MAXSIZE = 128 + + +class EventBus: + """Fan out change notices to every live subscriber, in-process.""" + + def __init__(self, max_queue: int = _QUEUE_MAXSIZE) -> None: + self._subscribers: set[asyncio.Queue] = set() + self._max = max_queue + + def publish(self, event: dict[str, Any]) -> None: + """Deliver `event` to every subscriber. Never raises or blocks: a full + buffer drops its oldest notice so the newest still lands.""" + for q in list(self._subscribers): + try: + q.put_nowait(event) + except asyncio.QueueFull: + try: + q.get_nowait() + q.put_nowait(event) + except Exception: + pass + + async def subscribe(self) -> AsyncIterator[dict[str, Any]]: + """Yield notices until the consumer stops iterating (client disconnect + cancels the generator, and the `finally` unregisters the queue).""" + q: asyncio.Queue = asyncio.Queue(maxsize=self._max) + self._subscribers.add(q) + try: + while True: + yield await q.get() + finally: + self._subscribers.discard(q) + + @property + def subscriber_count(self) -> int: + return len(self._subscribers) + + +event_bus = EventBus() diff --git a/webui/backend/app/main.py b/webui/backend/app/main.py index a7f8a8a9e..a9a157894 100644 --- a/webui/backend/app/main.py +++ b/webui/backend/app/main.py @@ -6,6 +6,7 @@ from app.api import ( agent_settings, chat, + events, instructions, mcps, memory, @@ -88,6 +89,7 @@ async def _shutdown() -> None: skill_index.stop() app.include_router(chat.router) + app.include_router(events.router) app.include_router(presence.router) app.include_router(projects.router) app.include_router(sessions.router) diff --git a/webui/frontend/app/components/common/Composer.tsx b/webui/frontend/app/components/common/Composer.tsx index 72ce629a4..311b818f3 100644 --- a/webui/frontend/app/components/common/Composer.tsx +++ b/webui/frontend/app/components/common/Composer.tsx @@ -11,7 +11,13 @@ import { PillButton } from './PillButton' import { api } from '~/lib/api' import { useSessionModel, type SessionModelSelection } from '~/lib/sessionModel' import { useModelChanged } from '~/lib/modelChanged' -import { useOnMcpSkillChanged, dispatchWorkspaceChanged } from '~/lib/events' +import { + useOnMcpSkillChanged, + useOnModelsChanged, + useOnProjectSettingsChanged, + useOnProjectsChanged, + dispatchWorkspaceChanged +} from '~/lib/events' import type { ChatFileRef } from '~/lib/agentProvider' import { useT } from '~/lib/i18n' import type { @@ -299,6 +305,50 @@ export function Composer({ }, [effectiveProject?.id]) useOnMcpSkillChanged(refreshMcpSkill) + // Re-fetch when the model catalog changes elsewhere (Settings → Models, or an + // external API call relayed by the server-event bridge). The picker seeds + // these from the loader once at mount, so without this a new model never + // appears until the component remounts. + const refreshModels = useCallback(() => { + api + .listProviders() + .then(setProviders) + .catch(() => {}) + api + .listModels() + .then(setModels) + .catch(() => {}) + api + .getAgentSettings() + .then(setSettings) + .catch(() => {}) + }, []) + useOnModelsChanged(refreshModels) + + // Web-search config is edited on Settings → Search and can arrive via an + // external API call; both relay as a project-settings change. The pill seeds + // from the loader once at mount, so without this a toggle elsewhere never + // reflects here until the Composer remounts. + const refreshSearchSettings = useCallback(() => { + api + .getSearchSettings() + .then(setSearchSettings) + .catch(() => {}) + }, []) + useOnProjectSettingsChanged(refreshSearchSettings) + + // The picker seeds its project list from the loader once at mount; a project + // added, renamed or removed elsewhere (or by an external API call) reaches it + // through this event. Only meaningful when the picker is shown (homepage). + const refreshProjects = useCallback(() => { + if (!hasProjectPicker) return + api + .listProjects() + .then(setProjects) + .catch(() => {}) + }, [hasProjectPicker]) + useOnProjectsChanged(refreshProjects) + const mergedMcps = useMemo( () => [...globalMcps, ...projectMcps], [globalMcps, projectMcps] diff --git a/webui/frontend/app/components/layout/Sidebar.tsx b/webui/frontend/app/components/layout/Sidebar.tsx index ea9594848..2a9670289 100644 --- a/webui/frontend/app/components/layout/Sidebar.tsx +++ b/webui/frontend/app/components/layout/Sidebar.tsx @@ -457,12 +457,16 @@ function ProjectRowActions({ okButtonProps: { danger: true }, onOk: async () => { await api.deleteProject(project.id) - // Awaited: see the create handler — a navigation in the same tick would - // interrupt this refresh, and the route change itself no longer triggers - // one, so the deleted project would linger in the sidebar. + // Leave the deleted project's page and let that navigation fully settle + // before revalidating. Revalidating while the switch is still in flight + // aborts it (the router restarts the pending load), and revalidating in + // place re-runs the now-missing project's loader and flashes its 404 — + // awaiting navigate lands us on a live URL, so the refresh repaints the + // sidebar there instead. + if (location.pathname.startsWith(`/projects/${project.id}`)) { + await navigate('/', { replace: true }) + } await revalidator.revalidate() - if (location.pathname.startsWith(`/projects/${project.id}`)) - navigate('/') } }) } @@ -946,11 +950,14 @@ function SessionItem({ okButtonProps: { danger: true }, onOk: async () => { await api.deleteSession(session.id) - // Awaited for the same reason as project delete: the route change that - // follows no longer revalidates on its own. - await revalidator.revalidate() + // Same as project delete: leave the deleted session's page and let that + // navigation settle before revalidating, so the refresh runs on a live + // URL instead of aborting the switch or flashing the session's 404. const isActive = location.pathname.includes(`/sessions/${session.id}`) - if (isActive) navigate(`/projects/${projectId}`) + if (isActive) { + await navigate(`/projects/${projectId}`, { replace: true }) + } + await revalidator.revalidate() } }) } diff --git a/webui/frontend/app/components/project/McpTabPanel.tsx b/webui/frontend/app/components/project/McpTabPanel.tsx index 0ac2b5320..888f14547 100644 --- a/webui/frontend/app/components/project/McpTabPanel.tsx +++ b/webui/frontend/app/components/project/McpTabPanel.tsx @@ -6,7 +6,7 @@ import { CardSkeletonGrid } from '~/components/common/CardSkeletonGrid' import { EmptyState, EmptyStateAction } from '~/components/common/EmptyState' import { MsaButton } from '~/components/common/MsaButton' import { api } from '~/lib/api' -import { dispatchMcpSkillChanged } from '~/lib/events' +import { dispatchMcpSkillChanged, useOnMcpSkillChanged } from '~/lib/events' import { useT } from '~/lib/i18n' import { useMcpHealth } from '~/lib/mcpHealth' import type { Mcp, Project, Scope } from '~/lib/types' @@ -105,6 +105,10 @@ export function McpTabPanel({ project }: Props) { setPage(1) }, [activeScope]) + // Refresh when the set changes elsewhere (the other tab, or an external API + // call relayed by the server-event bridge). `fresh` re-runs the health sweep. + useOnMcpSkillChanged(() => refresh(true)) + // Reset state when project changes (the scope itself is URL-driven; a // cross-project navigation carries no ?scope, which already means global). // Closing the JSON dialog matters: its document belongs to the scope it was diff --git a/webui/frontend/app/components/project/SkillTabPanel.tsx b/webui/frontend/app/components/project/SkillTabPanel.tsx index b7478f99d..04698fcdf 100644 --- a/webui/frontend/app/components/project/SkillTabPanel.tsx +++ b/webui/frontend/app/components/project/SkillTabPanel.tsx @@ -5,7 +5,7 @@ import { CardSkeletonGrid } from '~/components/common/CardSkeletonGrid' import { EmptyState, EmptyStateAction } from '~/components/common/EmptyState' import { MsaButton } from '~/components/common/MsaButton' import { api } from '~/lib/api' -import { dispatchMcpSkillChanged } from '~/lib/events' +import { dispatchMcpSkillChanged, useOnMcpSkillChanged } from '~/lib/events' import { useT } from '~/lib/i18n' import type { Project, Scope, Skill } from '~/lib/types' import { SkillCard } from '~/components/resources/SkillCard' @@ -58,6 +58,10 @@ export function SkillTabPanel({ project }: Props) { setPage(1) }, [activeScope]) + // Refresh when the set changes elsewhere (the other tab, or an external API + // call relayed by the server-event bridge). + useOnMcpSkillChanged(refresh) + // Reset state when project changes (the scope itself is URL-driven). useEffect(() => { setPage(1) diff --git a/webui/frontend/app/components/resources/McpsPanel.tsx b/webui/frontend/app/components/resources/McpsPanel.tsx index 2243734c0..f5fb69034 100644 --- a/webui/frontend/app/components/resources/McpsPanel.tsx +++ b/webui/frontend/app/components/resources/McpsPanel.tsx @@ -4,6 +4,7 @@ import { useSearchParams } from 'react-router' import { CardSkeletonGrid } from '~/components/common/CardSkeletonGrid' import { EmptyState, EmptyStateAction } from '~/components/common/EmptyState' import { api } from '~/lib/api' +import { useOnMcpSkillChanged } from '~/lib/events' import { useT } from '~/lib/i18n' import { useMcpHealth } from '~/lib/mcpHealth' import type { Mcp, Scope } from '~/lib/types' @@ -108,6 +109,10 @@ export function McpsPanel({ refresh() }, [activeScope]) + // Refresh when the set changes elsewhere (e.g. an external API call relayed by + // the server-event bridge). `fresh` re-runs the health sweep. + useOnMcpSkillChanged(() => refresh(true)) + const scopeBadge = activeScope === 'global' ? t.mcpImport.hubGlobalBadge diff --git a/webui/frontend/app/components/resources/SkillsPanel.tsx b/webui/frontend/app/components/resources/SkillsPanel.tsx index 9132848b8..2d8904d9e 100644 --- a/webui/frontend/app/components/resources/SkillsPanel.tsx +++ b/webui/frontend/app/components/resources/SkillsPanel.tsx @@ -5,6 +5,7 @@ import { CardSkeletonGrid } from '~/components/common/CardSkeletonGrid' import { EmptyState, EmptyStateAction } from '~/components/common/EmptyState' import { MsaSwitch } from '~/components/common/MsaSwitch' import { api } from '~/lib/api' +import { useOnMcpSkillChanged } from '~/lib/events' import { useT } from '~/lib/i18n' import type { Scope, Skill } from '~/lib/types' import { SkillCard } from './SkillCard' @@ -53,6 +54,10 @@ export function SkillsPanel({ refresh() }, [activeScope]) + // Refresh when the set changes elsewhere (e.g. an external API call relayed by + // the server-event bridge). + useOnMcpSkillChanged(refresh) + return (