refac

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:

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<typeof setTimeout>;
 	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
+	);
+}

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<TerminalFeatures> {
+	try {
+		const result = await terminalRequest<{ features?: TerminalFeatures }>(
+			connection,
+			chatId,
+			'/api/config',
+			{ signal }
+		);
+		return result.features ?? {};
+	} catch {
+		return {};
+	}
+}
+
 export type TerminalFileRoot = {
 	path: string;
 	label: string;

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<T>(
+	connection: TerminalConnection,
+	chatId: string | null,
+	path: () => string,
+	onData: (data: T) => void | Promise<void>,
+	onError: (message: string) => void,
+	onUnavailable?: () => void
+) {
+	let stopped = false;
+	let received = false;
+	let socket: WebSocket | undefined;
+	let retry: ReturnType<typeof setTimeout>;
+	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();
+		}
+	};
+}

src/lib/components/chat/ChatTabs/GitBar.svelte

@@ -1,6 +1,6 @@
 <script lang="ts">
 	import { getContext, onDestroy, onMount } from 'svelte';
-	import type { TerminalConnection } from '$lib/apis/terminal';
+	import { getTerminalFeatures, type TerminalConnection } from '$lib/apis/terminal';
 	import { watchTerminalFiles } from '$lib/apis/terminal/events';
 	import {
 		gitRequest,
@@ -80,6 +80,8 @@
 	let diffGeneration = 0;
 	let stopWatching: (() => void) | undefined;
 	let watchError = '';
+	let gitEnabled = false;
+	let configuration = 0;
 	let pending: Promise<void> | null = null;
 	let syncAction: 'publish' | 'pull' | 'push' | 'fetch';
 	$: staged = status?.files.filter((file) => file.staged) ?? [];
@@ -127,7 +129,8 @@
 		body?: Record<string, unknown>
 	) => gitRequest<T>(connection!, chatId, status?.root ?? directory, operation, params, body);
 	const refresh = (force = false): Promise<void> => {
-		if (!connection || !active || (!force && document.hidden)) return Promise.resolve();
+		if (!connection || !active || !gitEnabled || (!force && document.hidden))
+			return Promise.resolve();
 		if (pending) return pending;
 		const current = generation;
 		const task = gitRequest<GitStatus>(connection, chatId, directory, 'status')
@@ -174,10 +177,12 @@
 		diff = '';
 		loadingDiff = false;
 	};
-	const configure = (key: string, enabled: boolean) => {
+	const configure = async (key: string, enabled: boolean) => {
+		const version = ++configuration;
 		stopWatching?.();
 		stopWatching = undefined;
 		watchError = '';
+		gitEnabled = false;
 		if (key !== context) {
 			context = key;
 			++generation;
@@ -201,7 +206,15 @@
 			resetSelection();
 		}
 		if (enabled && connection) {
+			const features = await getTerminalFeatures(connection, chatId);
+			if (version !== configuration || !mounted) return;
+			gitEnabled = features.git === true;
+			if (!gitEnabled) {
+				status = null;
+				return;
+			}
 			void refresh();
+			if (features.events !== true) return;
 			const current = generation;
 			stopWatching = watchTerminalFiles(
 				connection,
@@ -335,6 +348,8 @@
 		return () => document.removeEventListener('visibilitychange', visibility);
 	});
 	onDestroy(() => {
+		mounted = false;
+		++configuration;
 		++generation;
 		++diffGeneration;
 		stopWatching?.();

src/lib/components/chat/FileNav/PortList.svelte

@@ -1,7 +1,8 @@
 <script lang="ts">
-	import { onDestroy, getContext, createEventDispatcher } from 'svelte';
+	import { onMount, onDestroy, getContext, createEventDispatcher } from 'svelte';
 	import type { ListeningPort } from '$lib/apis/terminal';
-	import { getListeningPorts, getPortProxyUrl } from '$lib/apis/terminal';
+	import { getListeningPorts, getPortProxyUrl, getTerminalFeatures } from '$lib/apis/terminal';
+	import { watchTerminalPorts } from '$lib/apis/terminal/events';
 	import Tooltip from '$lib/components/common/Tooltip.svelte';
 	import Icon from './Icon.svelte';
 
@@ -16,24 +17,46 @@
 
 	let ports: ListeningPort[] = [];
 	let loading = false;
-	let pollTimer: ReturnType<typeof setInterval> | null = null;
+	let mounted = false;
+	let context = '';
+	let generation = 0;
+	let stopWatching: (() => void) | undefined;
 
 	const loadPorts = async () => {
+		const current = generation;
 		loading = true;
-		ports = await getListeningPorts(baseUrl, apiKey, chatId);
+		const next = await getListeningPorts(baseUrl, apiKey, chatId);
+		if (current !== generation) return;
+		ports = next;
 		loading = false;
 	};
 
-	const startPolling = () => {
-		stopPolling();
-		loadPorts();
-		pollTimer = setInterval(loadPorts, 5000);
-	};
-
-	const stopPolling = () => {
-		if (pollTimer) {
-			clearInterval(pollTimer);
-			pollTimer = null;
+	const configure = async (key: string, enabled: boolean) => {
+		const current = ++generation;
+		stopWatching?.();
+		stopWatching = undefined;
+		if (context !== key) ports = [];
+		context = key;
+		loading = false;
+		if (!enabled || !baseUrl) return;
+		const connection = { baseUrl, key: apiKey, selector: '', system: false };
+		const features = await getTerminalFeatures(connection, chatId);
+		if (current !== generation) return;
+		if (features.events === true) {
+			loading = true;
+			stopWatching = watchTerminalPorts(
+				connection,
+				chatId,
+				(next) => {
+					ports = next;
+					loading = false;
+				},
+				() => {
+					loading = false;
+				}
+			);
+		} else {
+			await loadPorts();
 		}
 	};
 
@@ -47,12 +70,13 @@
 	};
 
 	// Only discover ports while the browser start view is visible.
-	$: if (baseUrl && active) {
-		startPolling();
-	} else stopPolling();
-
+	$: configure(JSON.stringify([baseUrl, apiKey, chatId]), mounted && active);
+	onMount(() => {
+		mounted = true;
+	});
 	onDestroy(() => {
-		stopPolling();
+		++generation;
+		stopWatching?.();
 	});
 </script>
 

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<typeof setInterval> | 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<TerminalProcessOutput>(
+				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<TerminalProcess[]>(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<TerminalProcess[]>(
+				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);
 		};
 	});

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));