From dae86edba4a389140850a4d7d9bf81e35dbf04b6 Mon Sep 17 00:00:00 2001 From: chengyongru <2755839590@qq.com> Date: Fri, 5 Jun 2026 23:59:07 +0800 Subject: [PATCH] fix(webui): preserve desktop restart and replay state --- desktop/src/notifications.ts | 37 +++++- nanobot/channels/websocket.py | 56 ++++----- nanobot/webui/file_preview.py | 2 + tests/channels/test_websocket_channel.py | 40 +++++++ webui/src/App.tsx | 106 +++++++++--------- .../src/components/settings/SettingsView.tsx | 29 +++-- webui/src/tests/settings-view.test.tsx | 62 ++++++++++ 7 files changed, 239 insertions(+), 93 deletions(-) diff --git a/desktop/src/notifications.ts b/desktop/src/notifications.ts index 6ae2f72ef..2fda51e50 100644 --- a/desktop/src/notifications.ts +++ b/desktop/src/notifications.ts @@ -14,6 +14,7 @@ type WsMessageFrame = { event?: unknown; kind?: unknown; source?: NotificationSource; + stream_id?: unknown; text?: unknown; }; @@ -25,15 +26,17 @@ const MAX_NOTIFICATION_BODY_LENGTH = 180; const MAX_NOTIFICATION_TITLE_LENGTH = 80; let unreadNotificationCount = 0; +const streamTextBuffers = new Map(); export function handleDesktopNotificationFrame( data: string, options: DesktopNotifierOptions, ): void { const frame = parseWsMessageFrame(data); - if (!frame || !isAssistantNotificationFrame(frame)) return; + const notificationFrame = frame ? notificationFrameFromWsFrame(frame) : null; + if (!notificationFrame) return; if (!shouldNotify(options.getWindow())) return; - showDesktopNotification(frame, options); + showDesktopNotification(notificationFrame, options); } export function clearDesktopNotificationBadge(): void { @@ -67,6 +70,36 @@ function isAssistantNotificationFrame(frame: WsMessageFrame): frame is WsMessage ); } +function notificationFrameFromWsFrame(frame: WsMessageFrame): WsMessageFrame & { + chat_id: string; + text: string; +} | null { + if (isAssistantNotificationFrame(frame)) return frame; + if (frame.event === "delta") { + if (typeof frame.chat_id === "string" && typeof frame.text === "string") { + const key = streamNotificationKey(frame); + streamTextBuffers.set(key, `${streamTextBuffers.get(key) ?? ""}${frame.text}`); + } + return null; + } + if (frame.event === "stream_end" && typeof frame.chat_id === "string") { + const key = streamNotificationKey(frame); + const text = typeof frame.text === "string" + ? frame.text + : streamTextBuffers.get(key) ?? ""; + streamTextBuffers.delete(key); + return text.trim().length > 0 + ? { ...frame, chat_id: frame.chat_id, text } + : null; + } + return null; +} + +function streamNotificationKey(frame: WsMessageFrame): string { + const streamId = typeof frame.stream_id === "string" ? frame.stream_id : ""; + return `${frame.chat_id ?? ""}\u0000${streamId}`; +} + function shouldNotify(win: BrowserWindow | null): boolean { if (!Notification.isSupported()) return false; if (!win || win.isDestroyed()) return false; diff --git a/nanobot/channels/websocket.py b/nanobot/channels/websocket.py index 09b3f900e..8675b6252 100644 --- a/nanobot/channels/websocket.py +++ b/nanobot/channels/websocket.py @@ -860,20 +860,21 @@ class WebSocketChannel(BaseChannel): self.logger.debug("no active subscribers for chat_id={}", msg.chat_id) else: self.logger.warning("no active subscribers for chat_id={}", msg.chat_id) - return if msg.metadata.get("_goal_state_sync"): - blob = msg.metadata.get("goal_state") - await self.send_goal_state(msg.chat_id, blob if isinstance(blob, dict) else {"active": False}) + if conns: + blob = msg.metadata.get("goal_state") + await self.send_goal_state(msg.chat_id, blob if isinstance(blob, dict) else {"active": False}) return if msg.metadata.get("_goal_status"): - status = msg.metadata.get("goal_status") - if status in ("running", "idle"): - started_raw = msg.metadata.get("started_at", msg.metadata.get("goal_started_at")) - await self.send_goal_status( - msg.chat_id, - status, - started_at=float(started_raw) if isinstance(started_raw, int | float) else None, - ) + if conns: + status = msg.metadata.get("goal_status") + if status in ("running", "idle"): + started_raw = msg.metadata.get("started_at", msg.metadata.get("goal_started_at")) + await self.send_goal_status( + msg.chat_id, + status, + started_at=float(started_raw) if isinstance(started_raw, int | float) else None, + ) return # Signal that the agent has fully finished processing the current turn. if msg.metadata.get("_turn_end"): @@ -889,11 +890,12 @@ class WebSocketChannel(BaseChannel): ) return if msg.metadata.get("_session_updated"): - scope = msg.metadata.get("_session_update_scope") - await self.send_session_updated( - msg.chat_id, - scope=scope if isinstance(scope, str) else None, - ) + if conns: + scope = msg.metadata.get("_session_update_scope") + await self.send_session_updated( + msg.chat_id, + scope=scope if isinstance(scope, str) else None, + ) return if msg.metadata.get("_file_edit_events"): edits = msg.metadata.get("_file_edit_events") @@ -946,6 +948,8 @@ class WebSocketChannel(BaseChannel): transcript_overrides={"text": text}, ) raw = json.dumps(payload, ensure_ascii=False) + if not conns: + return for connection in conns: await self._safe_send_to(connection, raw, label=" ") @@ -961,7 +965,7 @@ class WebSocketChannel(BaseChannel): until the matching ``reasoning_end`` arrives. """ conns = list(self._subs.get(chat_id, ())) - if not conns or not delta: + if not delta: return meta = metadata or {} body: dict[str, Any] = { @@ -979,6 +983,8 @@ class WebSocketChannel(BaseChannel): phase="reasoning", ) raw = json.dumps(body, ensure_ascii=False) + if not conns: + return for connection in conns: await self._safe_send_to(connection, raw, label=" reasoning ") @@ -989,8 +995,6 @@ class WebSocketChannel(BaseChannel): ) -> None: """Close the current reasoning stream segment for in-place renderers.""" conns = list(self._subs.get(chat_id, ())) - if not conns: - return meta = metadata or {} body: dict[str, Any] = { "event": "reasoning_end", @@ -1006,6 +1010,8 @@ class WebSocketChannel(BaseChannel): phase="reasoning", ) raw = json.dumps(body, ensure_ascii=False) + if not conns: + return for connection in conns: await self._safe_send_to(connection, raw, label=" reasoning_end ") @@ -1016,8 +1022,6 @@ class WebSocketChannel(BaseChannel): metadata: dict[str, Any] | None = None, ) -> None: conns = list(self._subs.get(chat_id, ())) - if not conns: - return payload: dict[str, Any] = { "event": "file_edit", "chat_id": chat_id, @@ -1030,6 +1034,8 @@ class WebSocketChannel(BaseChannel): phase="activity", ) raw = json.dumps(payload, ensure_ascii=False) + if not conns: + return for connection in conns: await self._safe_send_to(connection, raw, label=" file_edit ") @@ -1040,8 +1046,6 @@ class WebSocketChannel(BaseChannel): metadata: dict[str, Any] | None = None, ) -> None: conns = list(self._subs.get(chat_id, ())) - if not conns: - return meta = metadata or {} stream_key = (chat_id, str(meta.get("_stream_id") or "")) if meta.get("_stream_end"): @@ -1069,6 +1073,8 @@ class WebSocketChannel(BaseChannel): phase="answer", ) raw = json.dumps(body, ensure_ascii=False) + if not conns: + return for connection in conns: await self._safe_send_to(connection, raw, label=" stream ") @@ -1082,8 +1088,6 @@ class WebSocketChannel(BaseChannel): ) -> None: """Signal that the agent has fully finished processing the current turn.""" conns = list(self._subs.get(chat_id, ())) - if not conns: - return body: dict[str, Any] = {"event": "turn_end", "chat_id": chat_id} if latency_ms is not None: body["latency_ms"] = int(latency_ms) @@ -1096,6 +1100,8 @@ class WebSocketChannel(BaseChannel): phase="complete", ) raw = json.dumps(body, ensure_ascii=False) + if not conns: + return for connection in conns: await self._safe_send_to(connection, raw, label=" turn_end ") diff --git a/nanobot/webui/file_preview.py b/nanobot/webui/file_preview.py index 958f3ce85..6e1048823 100644 --- a/nanobot/webui/file_preview.py +++ b/nanobot/webui/file_preview.py @@ -90,6 +90,8 @@ def _clean_preview_path(raw_path: str | None) -> str: if value.startswith("file://"): parsed = urlparse(value) value = unquote(parsed.path) + if re.match(r"^/[A-Za-z]:[\\/]", value): + value = value[1:] else: value = unquote(value) value = value.split("?", 1)[0].split("#", 1)[0].strip() diff --git a/tests/channels/test_websocket_channel.py b/tests/channels/test_websocket_channel.py index 85530da65..3e358b076 100644 --- a/tests/channels/test_websocket_channel.py +++ b/tests/channels/test_websocket_channel.py @@ -1166,6 +1166,37 @@ async def test_send_reasoning_without_subscribers_is_noop() -> None: assert channel._subs == {} +@pytest.mark.asyncio +async def test_stream_transcript_persists_without_subscribers() -> None: + from nanobot.webui.transcript import build_webui_thread_response, read_transcript_lines + + bus = MagicMock() + channel = WebSocketChannel( + {"enabled": True, "allowFrom": ["*"], "streaming": True}, + bus, + gateway=_basic_handler(bus), + ) + + await channel.send_delta("chat-1", "hello", {"_stream_delta": True, "_stream_id": "s1"}) + await channel.send_delta("chat-1", " world", {"_stream_delta": True, "_stream_id": "s1"}) + await channel.send_delta("chat-1", "", {"_stream_end": True, "_stream_id": "s1"}) + await channel.send(OutboundMessage( + channel="websocket", + chat_id="chat-1", + content="", + metadata={"_turn_end": True, "latency_ms": 42}, + )) + + assert channel._subs == {} + lines = read_transcript_lines("websocket:chat-1") + assert [line["event"] for line in lines] == ["delta", "delta", "stream_end", "turn_end"] + body = build_webui_thread_response("websocket:chat-1") + assert body is not None + assert body["messages"][-1]["role"] == "assistant" + assert body["messages"][-1]["content"] == "hello world" + assert body["messages"][-1]["latencyMs"] == 42 + + @pytest.mark.asyncio async def test_send_turn_end_emits_turn_end_event() -> None: bus = MagicMock() @@ -2600,6 +2631,15 @@ def test_handle_file_preview_returns_workspace_file(tmp_path) -> None: assert body["truncated"] is False +def test_file_preview_normalizes_windows_file_url() -> None: + from nanobot.webui.file_preview import _clean_preview_path + + assert _clean_preview_path("file:///C:/Users/me/project/app.py") == ( + "C:/Users/me/project/app.py" + ) + assert _clean_preview_path("file:///tmp/project/app.py") == "/tmp/project/app.py" + + def test_handle_file_preview_rejects_paths_outside_workspace(tmp_path) -> None: from urllib.parse import quote diff --git a/webui/src/App.tsx b/webui/src/App.tsx index 982322d93..6c75ac0da 100644 --- a/webui/src/App.tsx +++ b/webui/src/App.tsx @@ -37,6 +37,7 @@ import { Input } from "@/components/ui/input"; import { fetchSettings, fetchWorkspaces } from "@/lib/api"; import { createRuntimeHost, + getHostApi, toRuntimeSurface, } from "@/lib/runtime"; import { projectNameFromPath } from "@/lib/workspace"; @@ -341,6 +342,36 @@ export default function App() { const [state, setState] = useState({ status: "loading" }); const bootstrapSecretRef = useRef(""); + const refreshReadyClient = useCallback( + async (client: NanobotClient, fallbackSurface: RuntimeSurface) => { + const boot = await fetchBootstrap("", bootstrapSecretRef.current); + const url = deriveWsUrl(boot.ws_path, boot.token, boot.ws_url); + const runtimeSurface = boot.runtime_surface + ? toRuntimeSurface(boot.runtime_surface) + : fallbackSurface; + const runtimeHost = createRuntimeHost(runtimeSurface, boot.runtime_capabilities); + const tokenExpiresAt = bootstrapTokenExpiresAt(boot.expires_in); + if (runtimeHost.socketFactory) { + client.updateUrl(url, runtimeHost.socketFactory); + } else { + client.updateUrl(url); + } + setState((current) => + current.status === "ready" && current.client === client + ? { + ...current, + token: boot.token, + tokenExpiresAt, + modelName: boot.model_name ?? current.modelName, + runtimeSurface, + } + : current, + ); + return { token: boot.token, url }; + }, + [], + ); + const bootstrapWithSecret = useCallback( (secret: string) => { let cancelled = false; @@ -358,37 +389,8 @@ export default function App() { socketFactory: runtimeHost.socketFactory, onReauth: async () => { try { - const refreshed = await fetchBootstrap("", bootstrapSecretRef.current); - const refreshedUrl = deriveWsUrl( - refreshed.ws_path, - refreshed.token, - refreshed.ws_url, - ); - const refreshedSurface = refreshed.runtime_surface - ? toRuntimeSurface(refreshed.runtime_surface) - : runtimeSurface; - const refreshedHost = createRuntimeHost( - refreshedSurface, - refreshed.runtime_capabilities, - ); - const tokenExpiresAt = bootstrapTokenExpiresAt(refreshed.expires_in); - if (refreshedHost.socketFactory) { - client.updateUrl(refreshedUrl, refreshedHost.socketFactory); - } else { - client.updateUrl(refreshedUrl); - } - setState((current) => - current.status === "ready" && current.client === client - ? { - ...current, - token: refreshed.token, - tokenExpiresAt, - modelName: refreshed.model_name ?? current.modelName, - runtimeSurface: refreshedSurface, - } - : current, - ); - return refreshedUrl; + const refreshed = await refreshReadyClient(client, runtimeSurface); + return refreshed.url; } catch { return null; } @@ -418,7 +420,7 @@ export default function App() { cancelled = true; }; }, - [], + [refreshReadyClient], ); useEffect(() => { @@ -426,29 +428,7 @@ export default function App() { const client = state.client; const timer = window.setTimeout(async () => { try { - const boot = await fetchBootstrap("", bootstrapSecretRef.current); - const url = deriveWsUrl(boot.ws_path, boot.token, boot.ws_url); - const runtimeSurface = boot.runtime_surface - ? toRuntimeSurface(boot.runtime_surface) - : state.runtimeSurface; - const runtimeHost = createRuntimeHost(runtimeSurface, boot.runtime_capabilities); - const tokenExpiresAt = bootstrapTokenExpiresAt(boot.expires_in); - if (runtimeHost.socketFactory) { - client.updateUrl(url, runtimeHost.socketFactory); - } else { - client.updateUrl(url); - } - setState((current) => - current.status === "ready" && current.client === client - ? { - ...current, - token: boot.token, - tokenExpiresAt, - modelName: boot.model_name ?? current.modelName, - runtimeSurface, - } - : current, - ); + await refreshReadyClient(client, state.runtimeSurface); } catch (e) { const msg = (e as Error).message; if (msg.includes("HTTP 401") || msg.includes("HTTP 403")) { @@ -457,7 +437,7 @@ export default function App() { } }, tokenRefreshDelayMs(state.tokenExpiresAt)); return () => window.clearTimeout(timer); - }, [state]); + }, [refreshReadyClient, state]); useEffect(() => { const saved = loadSavedSecret(); @@ -515,6 +495,16 @@ export default function App() { setState({ status: "auth" }); }; + const handleNativeEngineRestart = async (): Promise => { + const hostApi = getHostApi(); + if (!hostApi?.restartEngine) { + throw new Error("native engine restart is unavailable"); + } + await hostApi.restartEngine(); + const refreshed = await refreshReadyClient(state.client, state.runtimeSurface); + return refreshed.token; + }; + return ( ); @@ -534,10 +525,12 @@ function Shell({ runtimeSurface, onModelNameChange, onLogout, + onNativeEngineRestart, }: { runtimeSurface: RuntimeSurface; onModelNameChange: (modelName: string | null) => void; onLogout: () => void; + onNativeEngineRestart: () => Promise; }) { const { t, i18n } = useTranslation(); const { client, token } = useClient(); @@ -1519,6 +1512,7 @@ function Shell({ onSectionChange={onSettingsSectionChange} onLogout={onLogout} onRestart={onRestart} + onNativeEngineRestart={onNativeEngineRestart} isRestarting={isRestarting} hostChromeInset={showHostChrome} /> diff --git a/webui/src/components/settings/SettingsView.tsx b/webui/src/components/settings/SettingsView.tsx index 2c4d8831c..e41986697 100644 --- a/webui/src/components/settings/SettingsView.tsx +++ b/webui/src/components/settings/SettingsView.tsx @@ -287,6 +287,7 @@ interface SettingsViewProps { onSectionChange?: (section: SettingsSectionKey) => void; onLogout?: () => void; onRestart?: () => void; + onNativeEngineRestart?: () => Promise; isRestarting?: boolean; hostChromeInset?: boolean; } @@ -458,6 +459,7 @@ export function SettingsView({ onSectionChange, onLogout, onRestart, + onNativeEngineRestart, isRestarting = false, hostChromeInset = false, }: SettingsViewProps) { @@ -744,12 +746,15 @@ export function SettingsView({ const restartViaSettingsSurface = useCallback(async () => { const isNativeHost = (settings?.surface ?? settings?.runtime_surface) === "native"; - const hostApi = getHostApi(); - if (isNativeHost && settings?.runtime_capabilities?.can_restart_engine && hostApi) { + if ( + isNativeHost && + settings?.runtime_capabilities?.can_restart_engine && + onNativeEngineRestart + ) { setHostEngineApplying(true); try { - await hostApi.restartEngine(); - const payload = await fetchSettings(token); + const nextToken = await onNativeEngineRestart(); + const payload = await fetchSettings(nextToken); applyPayload(payload); setPendingRestartSections(EMPTY_PENDING_RESTART_SECTIONS); setError(null); @@ -761,21 +766,25 @@ export function SettingsView({ return; } onRestart?.(); - }, [applyPayload, onRestart, settings, token]); + }, [applyPayload, onNativeEngineRestart, onRestart, settings]); const maybeRestartHostEngine = useCallback( async (payload: RestartAwarePayload) => { const surface = payload.surface ?? payload.runtime_surface ?? settings?.surface ?? settings?.runtime_surface; const capabilities = payload.runtime_capabilities ?? settings?.runtime_capabilities; const isNativeHost = surface === "native"; - const hostApi = getHostApi(); - if (!payload.requires_restart || !isNativeHost || !capabilities?.can_restart_engine || !hostApi) { + if ( + !payload.requires_restart || + !isNativeHost || + !capabilities?.can_restart_engine || + !onNativeEngineRestart + ) { return; } setHostEngineApplying(true); try { - await hostApi.restartEngine(); - const refreshed = await fetchSettings(token); + const nextToken = await onNativeEngineRestart(); + const refreshed = await fetchSettings(nextToken); applyPayload(refreshed); setPendingRestartSections(EMPTY_PENDING_RESTART_SECTIONS); setError(null); @@ -785,7 +794,7 @@ export function SettingsView({ setHostEngineApplying(false); } }, - [applyPayload, settings, token], + [applyPayload, onNativeEngineRestart, settings], ); const saveModelSettings = async () => { diff --git a/webui/src/tests/settings-view.test.tsx b/webui/src/tests/settings-view.test.tsx index 2204f210c..970426515 100644 --- a/webui/src/tests/settings-view.test.tsx +++ b/webui/src/tests/settings-view.test.tsx @@ -120,6 +120,7 @@ function renderSettingsView( options: { initialSection?: "overview" | "apps" | "advanced" | "models"; onSettingsChange?: (payload: SettingsPayload) => void; + onNativeEngineRestart?: () => Promise; } = {}, ) { render( @@ -131,6 +132,7 @@ function renderSettingsView( onBackToChat={() => {}} onModelNameChange={() => {}} onSettingsChange={options.onSettingsChange} + onNativeEngineRestart={options.onNativeEngineRestart} /> , ); @@ -766,4 +768,64 @@ describe("SettingsView Apps catalog", () => { expect(screen.queryByText("Web safety")).not.toBeInTheDocument(); expect(screen.getByText("Allow Full Access shell commands to reach services on this Mac.")).toBeInTheDocument(); }); + + it("refreshes settings with a fresh token after native engine restart", async () => { + const payload = { + ...settingsPayload(), + surface: "native" as const, + runtime_surface: "native" as const, + runtime_capabilities: { + can_restart_engine: true, + can_pick_folder: true, + can_open_logs: true, + can_export_diagnostics: true, + }, + }; + const restartedPayload = { + ...payload, + advanced: { ...payload.advanced, webui_allow_local_service_access: false }, + requires_restart: true, + restart_required_sections: ["runtime"], + }; + const refreshedPayload = { + ...restartedPayload, + requires_restart: false, + restart_required_sections: [], + }; + const restartEngine = vi.fn(async () => "fresh-token"); + const fetchMock = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + const auth = (init?.headers as Record | undefined)?.Authorization; + if (url === "/api/settings" && auth === "Bearer fresh-token") { + return jsonResponse(refreshedPayload); + } + if (url === "/api/settings") return jsonResponse(payload); + if (url === "/api/settings/cli-apps") return jsonResponse({ apps: [], installed_count: 0 }); + if (url === "/api/settings/mcp-presets") return jsonResponse({ presets: [], installed_count: 0 }); + if (url === "/api/settings/network-safety/update?webui_allow_local_service_access=false&webui_default_access_mode=default") { + return jsonResponse(restartedPayload); + } + return { ok: false, status: 404, json: async () => ({}) } as Response; + }); + vi.stubGlobal("fetch", fetchMock); + + renderSettingsView({ + initialSection: "advanced", + onNativeEngineRestart: restartEngine, + }); + + expect(await screen.findByText("App safety")).toBeInTheDocument(); + fireEvent.click(screen.getByRole("switch", { name: "Local services" })); + fireEvent.click(screen.getByRole("button", { name: "Save" })); + + await waitFor(() => expect(restartEngine).toHaveBeenCalledTimes(1)); + await waitFor(() => + expect(fetchMock).toHaveBeenCalledWith( + "/api/settings", + expect.objectContaining({ + headers: { Authorization: "Bearer fresh-token" }, + }), + ), + ); + }); });