diff --git a/web/workspace/deno.json b/web/workspace/deno.json index 2ec7064a..f382edc4 100644 --- a/web/workspace/deno.json +++ b/web/workspace/deno.json @@ -6,7 +6,7 @@ "dev": "deno run -A npm:vite@7.2.7 dev", "dev:backend": "cd ../.. && cargo run -p yoi-workspace-server --bin yoi-server -- serve --listen 127.0.0.1:8787", "check": "deno run -A npm:@sveltejs/kit@2.49.4 sync && deno run -A npm:svelte-check@4.3.4 --tsconfig ./tsconfig.json", - "test": "deno test --allow-read=src --allow-env=VSCODE_TEXTMATE_DEBUG src/lib/workspace/auth/model.test.ts src/lib/workspace/api/http.test.ts src/lib/workspace/header/breadcrumb-model.test.ts src/lib/workspace/console/chat-submit.test.ts src/lib/workspace/console/composer-command.test.ts src/lib/workspace/console/composer-completion.test.ts src/lib/workspace/console/markdown.test.ts src/lib/workspace/console/model.test.ts src/lib/workspace/console/worker-console.ui.test.ts src/lib/workspace/settings/model.test.ts src/lib/workspace/sidebar/workers.test.ts src/lib/workspace/sidebar/worker-launch.test.ts src/lib/workspace/sidebar/repository-nav.test.ts", + "test": "deno test --allow-read=src --allow-env=VSCODE_TEXTMATE_DEBUG src/lib/workspace/auth/model.test.ts src/lib/workspace/api/http.test.ts src/lib/workspace/header/breadcrumb-model.test.ts src/lib/workspace/console/chat-submit.test.ts src/lib/workspace/console/composer-command.test.ts src/lib/workspace/console/composer-completion.test.ts src/lib/workspace/console/markdown.test.ts src/lib/workspace/console/model.test.ts src/lib/workspace/console/worker-console.ui.test.ts src/lib/workspace/settings/model.test.ts src/lib/workspace/sidebar/workers.test.ts src/lib/workspace/sidebar/worker-subscription.test.ts src/lib/workspace/sidebar/worker-launch.test.ts src/lib/workspace/sidebar/repository-nav.test.ts", "build": "deno run -A npm:vite@7.2.7 build", "preview": "deno run -A npm:vite@7.2.7 preview" }, diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index 34fe0f72..9ad47a8c 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts @@ -310,7 +310,8 @@ Deno.test("Worker Console uses protocol observation events without transcript fe assert( consolePage.includes("connectProtocolTransport") && consolePage.includes("handleIncomingProtocolEvent") && - consolePage.includes("/protocol/ws") && + consolePage.includes("workspaceMultiplexer") && + consolePage.includes('topic: "worker_protocol"') && !consolePage.includes("seenObservationEventIds") && consolePage.includes("createConsoleProjector") && consolePage.includes("consoleProjector.append(eventBatch)") && @@ -541,7 +542,8 @@ Deno.test("Worker Console page is routed by runtime_id and worker_id through bac ); assert( !consolePage.includes("/transcript") && - consolePage.includes("/protocol/ws") && + consolePage.includes("workspaceMultiplexer") && + consolePage.includes("sendWorkerMethod") && !consolePage.includes("/events" + "/ws") && !consolePage.includes("/input") && !consolePage.includes("/completions"), @@ -708,3 +710,25 @@ Deno.test("Account UI owns browser passkey session state without workspace autho "Root layout should not redirect account and device-login public routes to a workspace", ); }); + +Deno.test("Workspace Worker list and Console share the multiplexed connection", async () => { + const consolePage = await Deno.readTextFile( + new URL( + "./../../../routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte", + import.meta.url, + ), + ); + const sidebarStore = await Deno.readTextFile( + new URL("./../sidebar/worker-subscription.ts", import.meta.url), + ); + const multiplexer = await Deno.readTextFile( + new URL("./../multiplexer.ts", import.meta.url), + ); + assert( + consolePage.includes("workspaceMultiplexer(workspaceId)") && + sidebarStore.includes("workspaceMultiplexer(workspaceId)") && + multiplexer.includes("const multiplexers = new Map") && + multiplexer.includes("frame: 'worker_protocol'"), + "Sidebar and Console should share one Workspace multiplexer and route Worker methods through a subscription lane", + ); +}); diff --git a/web/workspace/src/lib/workspace/multiplexer.ts b/web/workspace/src/lib/workspace/multiplexer.ts new file mode 100644 index 00000000..a748476f --- /dev/null +++ b/web/workspace/src/lib/workspace/multiplexer.ts @@ -0,0 +1,214 @@ +import { browser } from '$app/environment'; +import type { + EventSubscriptionSelector, + Method, + SubscriptionFrame, + SubscriptionId, +} from '$lib/generated/protocol'; +import { workspaceApiPath } from '$lib/workspace/api/http'; + +type Listener = { + onFrame(frame: SubscriptionFrame): void; + onStatus?(status: 'connecting' | 'open' | 'closed', message?: string): void; +}; + +type ActiveSubscription = { + clientId: string; + selector: EventSubscriptionSelector; + listener: Listener; + requestId: string | null; + subscriptionId: SubscriptionId | null; +}; + +export type WorkspaceMultiplexerSubscription = { + close(): void; + sendWorkerMethod(method: Method): void; +}; + +const multiplexers = new Map(); + +export function workspaceMultiplexer(workspaceId: string): WorkspaceMultiplexer { + let multiplexer = multiplexers.get(workspaceId); + if (!multiplexer) { + multiplexer = new WorkspaceMultiplexer(workspaceId); + multiplexers.set(workspaceId, multiplexer); + } + return multiplexer; +} + +export class WorkspaceMultiplexer { + readonly #workspaceId: string; + readonly #subscriptions = new Map(); + readonly #requests = new Map(); + readonly #runtimeSubscriptions = new Map(); + #socket: WebSocket | null = null; + #reconnectTimer: ReturnType | null = null; + #closed = false; + + constructor(workspaceId: string) { + this.#workspaceId = workspaceId; + } + + subscribe( + selector: EventSubscriptionSelector, + listener: Listener, + ): WorkspaceMultiplexerSubscription { + const clientId = crypto.randomUUID(); + this.#subscriptions.set(clientId, { + clientId, + selector, + listener, + requestId: null, + subscriptionId: null, + }); + this.#closed = false; + this.#ensureConnected(); + return { + close: () => this.#remove(clientId), + sendWorkerMethod: (method) => this.#sendWorkerMethod(clientId, method), + }; + } + + #ensureConnected(): void { + if (!browser || this.#socket || this.#closed || this.#subscriptions.size === 0) return; + for (const subscription of this.#subscriptions.values()) { + subscription.listener.onStatus?.('connecting'); + } + const url = new URL( + workspaceApiPath(this.#workspaceId, '/protocol/ws'), + window.location.origin, + ); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + const socket = new WebSocket(url); + this.#socket = socket; + socket.addEventListener('open', () => { + for (const subscription of this.#subscriptions.values()) { + subscription.listener.onStatus?.('open'); + this.#sendSubscribe(subscription); + } + }); + socket.addEventListener('message', (event) => this.#receive(String(event.data))); + socket.addEventListener('error', () => socket.close()); + socket.addEventListener('close', () => { + if (this.#socket !== socket) return; + this.#socket = null; + this.#requests.clear(); + this.#runtimeSubscriptions.clear(); + for (const subscription of this.#subscriptions.values()) { + subscription.requestId = null; + subscription.subscriptionId = null; + subscription.listener.onStatus?.('closed', 'Workspace subscription disconnected'); + } + if (!this.#closed && this.#subscriptions.size > 0) { + this.#reconnectTimer = setTimeout(() => this.#ensureConnected(), 500); + } + }); + } + + #sendSubscribe(subscription: ActiveSubscription): void { + const requestId = crypto.randomUUID(); + subscription.requestId = requestId; + this.#requests.set(requestId, subscription.clientId); + this.#send({ + protocol_version: 1, + frame: 'request', + message: { + method: 'subscribe_events', + params: { request_id: requestId, selector: subscription.selector }, + }, + }); + } + + #receive(text: string): void { + let frame: SubscriptionFrame; + try { + frame = JSON.parse(text) as SubscriptionFrame; + } catch { + this.#socket?.close(); + return; + } + if (frame.protocol_version !== 1) { + this.#socket?.close(); + return; + } + if (frame.frame === 'response' && frame.message.result === 'subscribed') { + const clientId = this.#requests.get(frame.message.payload.request_id); + const subscription = clientId ? this.#subscriptions.get(clientId) : undefined; + if (!clientId || !subscription) return; + this.#requests.delete(frame.message.payload.request_id); + const subscriptionId = frame.message.payload.subscription_id; + if (!subscriptionId) return; + subscription.subscriptionId = subscriptionId; + this.#runtimeSubscriptions.set(subscriptionId, clientId); + subscription.listener.onFrame(frame); + return; + } + if (frame.frame === 'response' && frame.message.result === 'subscription_rejected') { + const clientId = this.#requests.get(frame.message.payload.request_id); + const subscription = clientId ? this.#subscriptions.get(clientId) : undefined; + subscription?.listener.onFrame(frame); + if (clientId) { + this.#requests.delete(frame.message.payload.request_id); + this.#remove(clientId); + } + return; + } + if (frame.frame === 'event') { + const clientId = this.#runtimeSubscriptions.get(frame.message.data.subscription_id); + const subscription = clientId ? this.#subscriptions.get(clientId) : undefined; + subscription?.listener.onFrame(frame); + if ( + frame.message.event === 'subscription_closed' && + clientId && + subscription && + this.#socket?.readyState === WebSocket.OPEN + ) { + this.#runtimeSubscriptions.delete(frame.message.data.subscription_id); + subscription.subscriptionId = null; + subscription.listener.onStatus?.('connecting', frame.message.data.message); + this.#sendSubscribe(subscription); + } + } + } + + #sendWorkerMethod(clientId: string, method: Method): void { + const subscription = this.#subscriptions.get(clientId); + if (!subscription?.subscriptionId) throw new Error('Worker protocol subscription is not open'); + this.#send({ + protocol_version: 1, + frame: 'worker_protocol', + message: { subscription_id: subscription.subscriptionId, method }, + }); + } + + #remove(clientId: string): void { + const subscription = this.#subscriptions.get(clientId); + if (!subscription) return; + this.#subscriptions.delete(clientId); + if (subscription.subscriptionId && this.#socket?.readyState === WebSocket.OPEN) { + this.#send({ + protocol_version: 1, + frame: 'request', + message: { + method: 'unsubscribe_events', + params: { + request_id: crypto.randomUUID(), + subscription_id: subscription.subscriptionId, + }, + }, + }); + this.#runtimeSubscriptions.delete(subscription.subscriptionId); + } + if (this.#subscriptions.size === 0) { + this.#closed = true; + if (this.#reconnectTimer) clearTimeout(this.#reconnectTimer); + this.#socket?.close(); + this.#socket = null; + } + } + + #send(frame: SubscriptionFrame): void { + if (this.#socket?.readyState !== WebSocket.OPEN) return; + this.#socket.send(JSON.stringify(frame)); + } +} diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts index 50c39a8d..ab55dc38 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts @@ -1,7 +1,6 @@ -import { browser } from '$app/environment'; import { readable, type Readable } from 'svelte/store'; -import type { SubscriptionFrame, SubscriptionWorker } from '$lib/generated/protocol'; -import { workspaceApiPath } from '$lib/workspace/api/http'; +import type { SubscriptionWorker } from '$lib/generated/protocol'; +import { workspaceMultiplexer } from '$lib/workspace/multiplexer'; import { applyWorkspaceWorkersFrame, createWorkspaceWorkersProjection, @@ -22,15 +21,11 @@ export function workspaceWorkersStore(workspaceId: string): Readable( { loading: true, error: null, workers: [] }, (set) => { - if (!browser || !workspaceId) { + if (!workspaceId) { set({ loading: false, error: null, workers: [] }); return; } - let closed = false; - let socket: WebSocket | null = null; - let reconnectTimer: ReturnType | null = null; const projection = createWorkspaceWorkersProjection(); - const publish = (loading = false, error: string | null = null) => { const workers = [...projection.workers.values()] .map(projectWorker) @@ -40,61 +35,33 @@ export function workspaceWorkersStore(workspaceId: string): Readable { - if (closed) return; - const url = new URL(workspaceApiPath(workspaceId, '/protocol/ws'), window.location.origin); - url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; - socket = new WebSocket(url); - socket.addEventListener('open', () => { - const frame: SubscriptionFrame = { - protocol_version: 1, - frame: 'request', - message: { - method: 'subscribe_events', - params: { - request_id: crypto.randomUUID(), - selector: { topic: 'workspace_workers' }, - }, - }, - }; - socket?.send(JSON.stringify(frame)); - }); - socket.addEventListener('message', (message) => { - try { - const frame = JSON.parse(String(message.data)) as SubscriptionFrame; - let closedMessage: string | null = null; - if (frame.frame === 'event' && frame.message.event === 'subscription_closed') { - closedMessage = frame.message.data.message; - } else if ( - frame.frame === 'response' && - frame.message.result === 'subscription_rejected' - ) { - closedMessage = frame.message.payload.message; + const subscription = workspaceMultiplexer(workspaceId).subscribe( + { topic: 'workspace_workers' }, + { + onFrame: (frame) => { + try { + if (frame.frame === 'event' && frame.message.event === 'subscription_closed') { + throw new Error(frame.message.data.message); + } + if ( + frame.frame === 'response' && + frame.message.result === 'subscription_rejected' + ) { + throw new Error(frame.message.payload.message); + } + applyWorkspaceWorkersFrame(projection, frame); + publish(false, null); + } catch (error) { + publish(false, error instanceof Error ? error.message : 'invalid Worker subscription frame'); } - if (closedMessage) { - socket?.close(); - throw new Error(closedMessage); - } - applyWorkspaceWorkersFrame(projection, frame); - publish(false, null); - } catch (error) { - publish(false, error instanceof Error ? error.message : 'invalid Worker subscription frame'); - } - }); - socket.addEventListener('close', () => { - socket = null; - if (closed) return; - publish(projection.workers.size === 0, 'Worker subscription disconnected; reconnecting…'); - reconnectTimer = setTimeout(connect, 500); - }); - socket.addEventListener('error', () => socket?.close()); - }; - connect(); - return () => { - closed = true; - if (reconnectTimer) clearTimeout(reconnectTimer); - socket?.close(); - }; + }, + onStatus: (status, message) => { + if (status === 'connecting') publish(projection.workers.size === 0, null); + if (status === 'closed') publish(projection.workers.size === 0, message ?? null); + }, + }, + ); + return () => subscription.close(); }, ); stores.set(workspaceId, store); diff --git a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte index 8980da8b..dcd21196 100644 --- a/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte +++ b/web/workspace/src/routes/w/[workspaceId]/runtimes/[runtimeId]/workers/[workerId]/console/+page.svelte @@ -24,6 +24,7 @@ } from "$lib/workspace/console/model"; import type { Event as ProtocolEvent, Method as ProtocolMethod, RewindTarget, Segment } from "$lib/generated/protocol"; import { workspaceApiPath } from "$lib/workspace/api/http"; + import { workspaceMultiplexer, type WorkspaceMultiplexerSubscription } from "$lib/workspace/multiplexer"; import type { Diagnostic, Worker, @@ -105,7 +106,7 @@ let protocolState = $state<"connecting" | "open" | "closed" | "error">( "connecting", ); - let protocolSocket: WebSocket | null = null; + let protocolSubscription: WorkspaceMultiplexerSubscription | null = null; let pendingCompletionRequest: { resolve: (entries: ComposerCompletionEntry[]) => void; reject: (error: Error) => void; @@ -514,80 +515,76 @@ return; } protocolState = "connecting"; - const protocol = window.location.protocol === "https:" ? "wss:" : "ws:"; - const wsPath = workerApiPath( - `/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent( - target.workerId, - )}/protocol/ws`, + const subscription = workspaceMultiplexer(workspaceId).subscribe( + { + topic: "worker_protocol", + worker_id: target.workerId, + runtime_id: target.runtimeId, + }, + { + onFrame: (frame) => { + if (token !== reloadToken) return; + try { + if ( + frame.frame === "response" && + frame.message.result === "subscribed" && + frame.message.payload.snapshot.topic === "worker_protocol" + ) { + for (const event of frame.message.payload.snapshot.data.events) { + handleIncomingProtocolEvent(event); + } + protocolState = "open"; + } else if ( + frame.frame === "event" && + frame.message.event === "event" && + frame.message.data.payload.event === "worker_protocol" + ) { + handleIncomingProtocolEvent(frame.message.data.payload.data.event); + } else if ( + frame.frame === "event" && + frame.message.event === "subscription_closed" + ) { + protocolState = "closed"; + rejectPendingCompletion(new Error(frame.message.data.message)); + } else if ( + frame.frame === "response" && + frame.message.result === "subscription_rejected" + ) { + protocolState = "error"; + throw new Error(frame.message.payload.message); + } + } catch (error) { + streamDiagnostics = [ + ...streamDiagnostics, + { + code: "worker_protocol_frame_invalid", + severity: "warning", + message: error instanceof Error ? error.message : String(error), + }, + ]; + } + }, + onStatus: (status) => { + if (token !== reloadToken) return; + protocolState = status === "open" ? "connecting" : status; + if (status === "closed") { + rejectPendingCompletion(new Error("Worker protocol WebSocket closed.")); + } + }, + }, ); - const ws = new WebSocket( - `${protocol}//${window.location.host}${wsPath}`, - ); - protocolSocket = ws; - - ws.onopen = () => { - if (token === reloadToken) { - protocolState = "open"; - } - }; - ws.onmessage = (message) => { - if (token !== reloadToken) { - return; - } - try { - handleIncomingProtocolEvent( - JSON.parse(String(message.data)) as ProtocolEvent, - ); - } catch (error) { - streamDiagnostics = [ - ...streamDiagnostics, - { - code: "worker_protocol_frame_invalid", - severity: "warning", - message: - error instanceof Error ? error.message : String(error), - }, - ]; - } - }; - ws.onerror = () => { - if (token === reloadToken) { - protocolState = "error"; - streamDiagnostics = [ - ...streamDiagnostics, - { - code: "worker_protocol_ws_error", - severity: "error", - message: "Worker protocol WebSocket failed.", - }, - ]; - } - }; - ws.onclose = () => { - if (protocolSocket === ws) { - protocolSocket = null; - } - if (token === reloadToken && protocolState !== "error") { - protocolState = "closed"; - } - rejectPendingCompletion( - new Error("Worker protocol WebSocket closed."), - ); - }; - + protocolSubscription = subscription; return () => { - if (protocolSocket === ws) { - protocolSocket = null; - } - ws.close(); + if (protocolSubscription === subscription) protocolSubscription = null; + subscription.close(); }; } function sendProtocolMethod(method: ProtocolMethod) { - if (!protocolSocket || protocolSocket.readyState !== WebSocket.OPEN) { + if (!protocolSubscription || protocolState !== "open") { throw new Error("Worker protocol WebSocket is not open."); } - protocolSocket.send(JSON.stringify(method)); + protocolSubscription.sendWorkerMethod(method); } function handleProtocolCommandEvent(event: ProtocolEvent) {