web: share workspace multiplexer with console
This commit is contained in:
@@ -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",
|
||||
);
|
||||
});
|
||||
|
||||
@@ -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<string, WorkspaceMultiplexer>();
|
||||
|
||||
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<string, ActiveSubscription>();
|
||||
readonly #requests = new Map<string, string>();
|
||||
readonly #runtimeSubscriptions = new Map<string, string>();
|
||||
#socket: WebSocket | null = null;
|
||||
#reconnectTimer: ReturnType<typeof setTimeout> | 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));
|
||||
}
|
||||
}
|
||||
@@ -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<WorkspaceWo
|
||||
const store = readable<WorkspaceWorkersState>(
|
||||
{ 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<typeof setTimeout> | 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<WorkspaceWo
|
||||
);
|
||||
set({ loading, error, workers });
|
||||
};
|
||||
const connect = () => {
|
||||
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);
|
||||
|
||||
+64
-67
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user