From 606cd5fa31690f84ed7e62ef89fe9d310c12f533 Mon Sep 17 00:00:00 2001 From: Hare Date: Sun, 2 Aug 2026 03:43:53 +0900 Subject: [PATCH] web: subscribe console on route changes --- .../console/worker-console.ui.test.ts | 17 +++++++++--- .../src/lib/workspace/multiplexer.ts | 12 ++++++--- .../workers/[workerId]/console/+page.svelte | 27 +++++++++++-------- 3 files changed, 38 insertions(+), 18 deletions(-) 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 9ad47a8c..fc9170d3 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 @@ -560,10 +560,11 @@ Deno.test("Worker Console page is routed by runtime_id and worker_id through bac "reload token advancement should not synchronously read and write the rune state", ); assert( - consolePage.includes("advanceReloadToken();") && - consolePage.includes("void loadConsoleData(target);") && + consolePage.includes("const token = advanceReloadToken();") && + consolePage.includes("worker = targetWorker;") && + consolePage.includes("if (!targetWorker) void loadWorker(target, token);") && !consolePage.includes("void refreshConsole();\n });\n\n $effect"), - "target-change effect should load data without depending on manual refresh state reads", + "target-change effect should install route data and guard fallback loading with the new target token", ); assert( consolePage.includes( @@ -725,10 +726,18 @@ Deno.test("Workspace Worker list and Console share the multiplexed connection", new URL("./../multiplexer.ts", import.meta.url), ); assert( - consolePage.includes("workspaceMultiplexer(workspaceId)") && + consolePage.includes("workspaceMultiplexer(target.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", ); + assert( + multiplexer.includes("this.#socket?.readyState === WebSocket.OPEN") && + multiplexer.includes("this.#sendSubscribe(subscription)") && + consolePage.includes("const targetWorker = data.worker") && + consolePage.includes("worker = targetWorker") && + consolePage.includes("const consoleTarget = $derived({ workspaceId, runtimeId, workerId })"), + "A reused Console route should subscribe immediately on the live Workspace socket and install the new route Worker", + ); }); diff --git a/web/workspace/src/lib/workspace/multiplexer.ts b/web/workspace/src/lib/workspace/multiplexer.ts index a748476f..5f8ab10c 100644 --- a/web/workspace/src/lib/workspace/multiplexer.ts +++ b/web/workspace/src/lib/workspace/multiplexer.ts @@ -54,15 +54,21 @@ export class WorkspaceMultiplexer { listener: Listener, ): WorkspaceMultiplexerSubscription { const clientId = crypto.randomUUID(); - this.#subscriptions.set(clientId, { + const subscription: ActiveSubscription = { clientId, selector, listener, requestId: null, subscriptionId: null, - }); + }; + this.#subscriptions.set(clientId, subscription); this.#closed = false; - this.#ensureConnected(); + if (this.#socket?.readyState === WebSocket.OPEN) { + subscription.listener.onStatus?.('open'); + this.#sendSubscribe(subscription); + } else { + this.#ensureConnected(); + } return { close: () => this.#remove(clientId), sendWorkerMethod: (method) => this.#sendWorkerMethod(clientId, method), 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 dcd21196..4b911ca8 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 @@ -140,11 +140,12 @@ let reloadToken = $state(0); type ConsoleTarget = { + workspaceId: string; runtimeId: string; workerId: string; }; - const consoleTarget = $derived({ runtimeId, workerId }); + const consoleTarget = $derived({ workspaceId, runtimeId, workerId }); const lines = $derived(consoleProjection.lines); const timelineLayout = $derived( @@ -179,17 +180,20 @@ return response.json() as Promise; } - async function loadWorker(target: ConsoleTarget) { + async function loadWorker(target: ConsoleTarget, token: number) { workerError = null; try { const payload = await getJson( - workerApiPath( + workspaceApiPath( + target.workspaceId, `/runtimes/${encodeURIComponent(target.runtimeId)}/workers/${encodeURIComponent(target.workerId)}`, ), ); + if (token !== reloadToken) return; worker = payload; liveWorkerState = payload.state; } catch (error) { + if (token !== reloadToken) return; workerError = error instanceof Error ? error.message : String(error); worker = null; @@ -197,10 +201,6 @@ } } - async function loadConsoleData(target: ConsoleTarget) { - if (!worker) await loadWorker(target); - } - function advanceReloadToken(): number { nextReloadToken += 1; reloadToken = nextReloadToken; @@ -515,7 +515,7 @@ return; } protocolState = "connecting"; - const subscription = workspaceMultiplexer(workspaceId).subscribe( + const subscription = workspaceMultiplexer(target.workspaceId).subscribe( { topic: "worker_protocol", worker_id: target.workerId, @@ -1083,11 +1083,16 @@ $effect(() => { const target = consoleTarget; + const targetWorker = data.worker; + const targetWorkerError = data.workerError; resetObservedEvents(); - liveWorkerState = null; + worker = targetWorker; + workerError = targetWorkerError; + liveWorkerState = targetWorker?.state ?? null; streamDiagnostics = []; - advanceReloadToken(); - void loadConsoleData(target); + protocolState = "connecting"; + const token = advanceReloadToken(); + if (!targetWorker) void loadWorker(target, token); }); $effect(() => connectProtocolTransport(worker, reloadToken, consoleTarget));