From 70c5e38fc7ce3864426b484c07ee201cba2b9d07 Mon Sep 17 00:00:00 2001 From: Timothy Jaeryang Baek Date: Sun, 1 Nov 2026 23:26:19 +0900 Subject: [PATCH] refac --- backend/open_webui/routers/terminals.py | 35 ++++- src/lib/apis/terminal/events.ts | 66 ++++++--- src/lib/apis/terminal/index.ts | 21 +++ src/lib/apis/terminal/socket.ts | 77 +++++++++++ .../components/chat/ChatTabs/GitBar.svelte | 21 ++- .../components/chat/FileNav/PortList.svelte | 62 ++++++--- src/lib/components/chat/TerminalDock.svelte | 126 +++++++++++++----- src/lib/components/chat/XTerminal.svelte | 46 +++---- 8 files changed, 350 insertions(+), 104 deletions(-) create mode 100644 src/lib/apis/terminal/socket.ts diff --git a/backend/open_webui/routers/terminals.py b/backend/open_webui/routers/terminals.py index b0e2ae7fbd..0018fcf5ae 100644 --- a/backend/open_webui/routers/terminals.py +++ b/backend/open_webui/routers/terminals.py @@ -431,11 +431,23 @@ async def _resolve_terminal_access(ws: WebSocket, server_id: str, token: str): return user, connection -@router.websocket('/{server_id}/files/events') -async def ws_file_events(ws: WebSocket, server_id: str): +@router.websocket('/{server_id}/events') +async def ws_events(ws: WebSocket, server_id: str): await _proxy_terminal_websocket(ws, server_id) +@router.websocket('/{server_id}/execute') +async def ws_processes(ws: WebSocket, server_id: str): + await _proxy_terminal_websocket(ws, server_id, stream_path='execute') + + +@router.websocket('/{server_id}/execute/{process_id}/status') +async def ws_process_output(ws: WebSocket, server_id: str, process_id: str): + from urllib.parse import quote + await _proxy_terminal_websocket(ws, server_id, + stream_path=f'execute/{quote(process_id, safe="")}/status') + + @router.websocket('/{server_id}/api/terminals/{session_id}') async def ws_terminal(ws: WebSocket, server_id: str, session_id: str): await _proxy_terminal_websocket(ws, server_id, session_id) @@ -445,6 +457,7 @@ async def _proxy_terminal_websocket( ws: WebSocket, server_id: str, session_id: str | None = None, + stream_path: str = 'events', ): """Proxy terminal sessions and filesystem notifications to a terminal server. @@ -470,6 +483,10 @@ async def _proxy_terminal_websocket( upstream_params = {} # For orchestrator-backed servers, pass user_id upstream_params['user_id'] = user.id + if stream_path.startswith('execute/'): + for key in ('offset', 'tail'): + if key in ws.query_params: + upstream_params[key] = ws.query_params[key] context_id = terminal_context_id(connection, {'chat_id': chat_id}, 'chat') upstream_headers = {'X-User-Id': user.id, 'X-Session-Id': chat_id} if terminal_context_config(connection, 'chat').get('context_id') == 'chat_id' and not context_id: @@ -484,7 +501,7 @@ async def _proxy_terminal_websocket( # decode depth) and inject an attacker-chosen user_id ahead of the one appended below. upstream_path = ( f'api/terminals/{urllib.parse.quote(session_id, safe="")}' - if session_id is not None else 'files/events' + if session_id is not None else stream_path ) upstream_url = f'{ws_base}/{upstream_path}' if upstream_params: @@ -549,8 +566,9 @@ async def _proxy_terminal_websocket( aiohttp.WSMsgType.ERROR, ): break - if session_id is None and upstream.close_code and 4000 <= upstream.close_code < 5000: - await ws.close(code=upstream.close_code) + if session_id is None: + code = upstream.close_code or 1011 + await ws.close(code=1011 if code in (1005, 1006, 1015) else code) except Exception: pass @@ -580,6 +598,13 @@ async def _proxy_terminal_websocket( await asyncio.gather(*tasks, return_exceptions=True) except Exception as e: log.exception('Terminal WebSocket proxy error: %s', e) + if session_id is None: + status = getattr(e, 'status', None) + code = {401: 4001, 403: 4003, 404: 4004}.get(status, 1011) + try: + await ws.close(code=code) + except Exception: + pass finally: await session.close() if opened: diff --git a/src/lib/apis/terminal/events.ts b/src/lib/apis/terminal/events.ts index 85eab09149..c7969c6e68 100644 --- a/src/lib/apis/terminal/events.ts +++ b/src/lib/apis/terminal/events.ts @@ -1,10 +1,11 @@ -import type { TerminalConnection } from './index'; +import type { ListeningPort, TerminalConnection } from './index'; +import { openTerminalSocket } from './socket'; -export function watchTerminalFiles( +function watchTerminalEvents( connection: TerminalConnection, chatId: string | null, - directory: string, - onChange: () => void, + subscription: { path?: string; ports?: boolean }, + onMessage: (message: any) => void, onError: (message: string) => void ) { let stopped = false; @@ -12,26 +13,18 @@ export function watchTerminalFiles( let retry: ReturnType; let retryDelay = 1000; const connect = () => { - const url = new URL(`${connection.baseUrl.replace(/\/$/, '')}/files/events`, location.href); - url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; - socket = new WebSocket(url); - socket.onopen = () => { - socket.send( - JSON.stringify({ type: 'auth', token: connection.key.trim(), chat_id: chatId ?? '' }) - ); - socket.send(JSON.stringify({ type: 'subscribe', path: directory })); - }; + socket = openTerminalSocket(connection, chatId, '/events', (socket) => { + socket.send(JSON.stringify({ type: 'subscribe', ...subscription })); + }); socket.onmessage = (event) => { if (stopped) return; try { const message = JSON.parse(event.data); - if (message.type === 'ready') { - retryDelay = 1000; - onError(''); - onChange(); - } else if (message.type === 'change') onChange(); + retryDelay = 1000; + onError(''); + onMessage(message); } catch { - /* Ignore frames that are not filesystem notifications. */ + /* Ignore invalid event frames. */ } }; socket.onclose = (event) => { @@ -50,3 +43,38 @@ export function watchTerminalFiles( socket.close(); }; } + +export function watchTerminalFiles( + connection: TerminalConnection, + chatId: string | null, + directory: string, + onChange: () => void, + onError: (message: string) => void +) { + return watchTerminalEvents( + connection, + chatId, + { path: directory }, + (message) => { + if (message.type === 'fs:ready' || message.type === 'fs:change') onChange(); + }, + onError + ); +} + +export function watchTerminalPorts( + connection: TerminalConnection, + chatId: string | null, + onPorts: (ports: ListeningPort[]) => void, + onError: (message: string) => void +) { + return watchTerminalEvents( + connection, + chatId, + { ports: true }, + (message) => { + if (message.type === 'ports:change') onPorts(message.ports); + }, + onError + ); +} diff --git a/src/lib/apis/terminal/index.ts b/src/lib/apis/terminal/index.ts index 8a89f8ff7f..b76cb35dfa 100644 --- a/src/lib/apis/terminal/index.ts +++ b/src/lib/apis/terminal/index.ts @@ -81,8 +81,29 @@ export type ListeningPort = { export type TerminalFeatures = { terminal?: boolean; + git?: boolean; + events?: boolean; }; +// Missing flags mean unsupported; never probe a new endpoint on an older server. +export async function getTerminalFeatures( + connection: TerminalConnection, + chatId: string | null, + signal?: AbortSignal +): Promise { + try { + const result = await terminalRequest<{ features?: TerminalFeatures }>( + connection, + chatId, + '/api/config', + { signal } + ); + return result.features ?? {}; + } catch { + return {}; + } +} + export type TerminalFileRoot = { path: string; label: string; diff --git a/src/lib/apis/terminal/socket.ts b/src/lib/apis/terminal/socket.ts new file mode 100644 index 0000000000..ae68f4f74e --- /dev/null +++ b/src/lib/apis/terminal/socket.ts @@ -0,0 +1,77 @@ +import type { TerminalConnection } from './index'; + +export function openTerminalSocket( + connection: TerminalConnection, + chatId: string | null, + path: string, + onOpen?: (socket: WebSocket) => void +) { + const url = new URL(`${connection.baseUrl.replace(/\/$/, '')}${path}`, location.href); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + const socket = new WebSocket(url); + socket.binaryType = 'arraybuffer'; + socket.onopen = () => { + socket.send( + JSON.stringify({ type: 'auth', token: connection.key.trim(), chat_id: chatId ?? '' }) + ); + onOpen?.(socket); + }; + return socket; +} + +// Streams use the same response bodies as their HTTP endpoints. +export function readTerminalStream( + connection: TerminalConnection, + chatId: string | null, + path: () => string, + onData: (data: T) => void | Promise, + onError: (message: string) => void, + onUnavailable?: () => void +) { + let stopped = false; + let received = false; + let socket: WebSocket | undefined; + let retry: ReturnType; + let delay = 1000; + let pending = Promise.resolve(); + const connect = async () => { + // Apply all received output before constructing the reconnect cursor. + await pending; + if (stopped) return; + socket = openTerminalSocket(connection, chatId, path()); + socket.onmessage = (event) => { + pending = pending + .then(async () => { + if (stopped) return; + const data = JSON.parse(event.data); + received = true; + await onData(data); + delay = 1000; + onError(''); + }) + .catch(() => onError('Live updates unavailable')); + }; + socket.onclose = async (event) => { + await pending; + if (stopped) return; + if (!received && onUnavailable && event.code !== 4001 && event.code !== 4003) { + stopped = true; + onUnavailable(); + return; + } + if (event.code !== 1000) onError('Live updates unavailable'); + if (event.code === 1000 || (event.code >= 4000 && event.code < 5000)) return; + retry = setTimeout(connect, delay); + delay = Math.min(delay * 2, 30000); + }; + }; + void connect(); + return () => { + stopped = true; + clearTimeout(retry); + if (socket) { + socket.onopen = socket.onmessage = socket.onclose = null; + socket.close(); + } + }; +} diff --git a/src/lib/components/chat/ChatTabs/GitBar.svelte b/src/lib/components/chat/ChatTabs/GitBar.svelte index 4798b6e63b..cfdf607763 100644 --- a/src/lib/components/chat/ChatTabs/GitBar.svelte +++ b/src/lib/components/chat/ChatTabs/GitBar.svelte @@ -1,6 +1,6 @@ diff --git a/src/lib/components/chat/TerminalDock.svelte b/src/lib/components/chat/TerminalDock.svelte index a0112d5dcf..3e2bf8aecb 100644 --- a/src/lib/components/chat/TerminalDock.svelte +++ b/src/lib/components/chat/TerminalDock.svelte @@ -9,6 +9,7 @@ type TerminalProcess, type TerminalProcessOutput } from '$lib/apis/terminal'; + import { readTerminalStream } from '$lib/apis/terminal/socket'; import XTerminal from './XTerminal.svelte'; import Icon from './FileNav/Icon.svelte'; @@ -37,14 +38,83 @@ const controller = new AbortController(); let polling = false; let disposed = false; + let streaming = true; + let stopList: (() => void) | undefined; + let stopOutput: (() => void) | undefined; + let outputContext = ''; + let interval: ReturnType | undefined; + + $: configureOutput( + streaming && active && expanded + ? (tabs.find((tab) => tab.id === activeId && tab.available && !tab.finished)?.id ?? '') + : '' + ); + + function configureOutput(id: string) { + if (id === outputContext) return; + outputContext = id; + stopOutput?.(); + stopOutput = undefined; + if (id && connection) { + stopOutput = readTerminalStream( + connection, + chatId, + () => + `/execute/${encodeURIComponent(id)}/status?offset=${tabs.find((tab) => tab.id === id)?.offset ?? 0}&tail=1000`, + writeOutput, + (message) => { + error = message; + } + ); + } + } $: if (active) void tick().then(poll); $: if (expanded) void tick().then(poll); $: hasSessions = tabs.length > 0; $: running = tabs.filter((tab) => tab.status === 'running' && tab.available).length; + function updateProcesses(processes: TerminalProcess[]) { + const live = new Map(processes.map((process) => [process.id, process])); + tabs = tabs.map((tab) => ({ ...tab, ...live.get(tab.id), available: live.has(tab.id) })); + const known = new Set(tabs.map((tab) => tab.id)); + for (const process of processes) { + if (!known.has(process.id) && !dismissed.has(process.id)) { + tabs = [ + ...tabs, + { ...process, offset: 0, loaded: false, available: true, finished: false } + ]; + } + } + if (!tabs.some((tab) => tab.id === activeId)) activeId = tabs[0]?.id ?? ''; + } + + async function writeOutput(result: TerminalProcessOutput) { + await tick(); + const tab = tabs.find((item) => item.id === result.id); + if (!tab) return; + if (disposed || !panes[tab.id]) return; + if (!tab.loaded) { + panes[tab.id].write(`$ ${tab.command.replace(/\r?\n/g, '\r\n')}\r\n`); + } + if (result.truncated) panes[tab.id].write('\r\n[Earlier output omitted]\r\n'); + panes[tab.id].write(result.output.map((entry) => entry.data).join('')); + tabs = tabs.map((item) => + item.id === tab.id + ? { + ...item, + status: result.status, + exit_code: result.exit_code, + offset: result.next_offset, + loaded: true, + finished: result.status !== 'running' + } + : item + ); + } + async function poll() { - if (!connection || !active || polling || disposed || document.hidden) return; + if (!connection || streaming || !active || polling || disposed || document.hidden) return; polling = true; try { const processes = await terminalRequest(connection, chatId, '/execute', { @@ -52,18 +122,7 @@ }); if (disposed) return; error = ''; - const live = new Map(processes.map((process) => [process.id, process])); - tabs = tabs.map((tab) => ({ ...tab, ...live.get(tab.id), available: live.has(tab.id) })); - const known = new Set(tabs.map((tab) => tab.id)); - for (const process of processes) { - if (!known.has(process.id) && !dismissed.has(process.id)) { - tabs = [ - ...tabs, - { ...process, offset: 0, loaded: false, available: true, finished: false } - ]; - } - } - if (!tabs.some((tab) => tab.id === activeId)) activeId = tabs[0]?.id ?? ''; + updateProcesses(processes); const tab = tabs.find((tab) => tab.id === activeId); if (expanded && tab?.available && !tab.finished) { await tick(); @@ -74,24 +133,7 @@ `/execute/${encodeURIComponent(tab.id)}/status?wait=0&offset=${tab.offset}&tail=1000`, { signal: controller.signal } ); - if (disposed || !panes[tab.id]) return; - if (!tab.loaded) { - panes[tab.id].write(`$ ${tab.command.replace(/\r?\n/g, '\r\n')}\r\n`); - } - if (result.truncated) panes[tab.id].write('\r\n[Earlier output omitted]\r\n'); - panes[tab.id].write(result.output.map((entry) => entry.data).join('')); - tabs = tabs.map((item) => - item.id === tab.id - ? { - ...item, - status: result.status, - exit_code: result.exit_code, - offset: result.next_offset, - loaded: true, - finished: result.status !== 'running' - } - : item - ); + await writeOutput(result); } } catch (cause) { if (!disposed) error = String(cause); @@ -138,13 +180,31 @@ ($settings as { terminalServers?: { url: string; key?: string }[] })?.terminalServers ?? [], localStorage.getItem('token') ?? '' ); - poll(); - const interval = setInterval(poll, 1000); + if (connection) { + stopList = readTerminalStream( + connection, + chatId, + () => '/execute', + updateProcesses, + (message) => { + error = message; + }, + () => { + if (disposed) return; + stopList?.(); + streaming = false; + void poll(); + interval = setInterval(poll, 1000); + } + ); + } document.addEventListener('visibilitychange', poll); return () => { disposed = true; controller.abort(); clearInterval(interval); + stopList?.(); + stopOutput?.(); document.removeEventListener('visibilitychange', poll); }; }); diff --git a/src/lib/components/chat/XTerminal.svelte b/src/lib/components/chat/XTerminal.svelte index fd93016705..938f304e7a 100644 --- a/src/lib/components/chat/XTerminal.svelte +++ b/src/lib/components/chat/XTerminal.svelte @@ -5,6 +5,7 @@ import { WebLinksAddon } from '@xterm/addon-web-links'; import '@xterm/xterm/css/xterm.css'; import { terminalRequest, type TerminalConnection } from '$lib/apis/terminal'; + import { openTerminalSocket } from '$lib/apis/terminal/socket'; import { connectedUserTerminals } from '$lib/stores'; export let connection: TerminalConnection; @@ -94,32 +95,27 @@ ); return; } - const url = new URL( - `${connection.baseUrl}/api/terminals/${encodeURIComponent(sessionId)}`, - location.href - ); - url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; - ws = new WebSocket(url); - ws.binaryType = 'arraybuffer'; - ws.onopen = () => { - if (destroyed) return; - ws?.send( - JSON.stringify({ type: 'auth', token: connection.key.trim(), chat_id: chatId ?? '' }) - ); - if (connection.selector && chatId) { - const shell = { terminalId: connection.selector, chatId }; - connectedUserTerminals.update((entries) => entries.set(terminalOwner, shell)); + ws = openTerminalSocket( + connection, + chatId, + `/api/terminals/${encodeURIComponent(sessionId)}`, + () => { + if (destroyed) return; + if (connection.selector && chatId) { + const shell = { terminalId: connection.selector, chatId }; + connectedUserTerminals.update((entries) => entries.set(terminalOwner, shell)); + } + connected = true; + connecting = false; + reportActivity(active); + fit(); + if (active) term?.focus(); + if (term) ws?.send(JSON.stringify({ type: 'resize', cols: term.cols, rows: term.rows })); + pingInterval = setInterval(() => { + if (ws?.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'ping' })); + }, 25000); } - connected = true; - connecting = false; - reportActivity(active); - fit(); - if (active) term?.focus(); - if (term) ws?.send(JSON.stringify({ type: 'resize', cols: term.cols, rows: term.rows })); - pingInterval = setInterval(() => { - if (ws?.readyState === WebSocket.OPEN) ws.send(JSON.stringify({ type: 'ping' })); - }, 25000); - }; + ); ws.onmessage = (event) => { if (destroyed) return; if (event.data instanceof ArrayBuffer) term?.write(new Uint8Array(event.data));