fix: integrate sidebar workspace alerts

This commit is contained in:
2026-10-03 23:43:27 +09:00
17 changed files with 900 additions and 137 deletions
@@ -40,6 +40,14 @@ export function pushWorkspaceAlert(
): string {
const id = options.id ??
`${Date.now().toString(36)}-${(sequence++).toString(36)}`;
const existing = alerts.find((alert) => alert.id === id);
if (
existing?.level === level &&
existing.title === options.title &&
existing.message === message
) {
return id;
}
alerts = [
...alerts.filter((alert) => alert.id !== id),
{
@@ -1438,7 +1438,7 @@ Deno.test("Workspace Worker list and Console share the multiplexed connection",
consolePage.includes("workspaceMultiplexer(target.workspaceId)") &&
sidebarStore.includes("workspaceMultiplexer(workspaceId)") &&
multiplexer.includes("const multiplexers = new Map") &&
multiplexer.includes("frame: 'worker_protocol'"),
multiplexer.includes('frame: "worker_protocol"'),
"Sidebar and Console should share one Workspace multiplexer and route Worker methods through a subscription lane",
);
assert(
@@ -99,7 +99,10 @@ test('invalid inbound frame closes before dispatch and reconnects for a fresh sn
expect(onFrame).not.toHaveBeenCalled();
expect(first.closes).toEqual([[4002, 'Invalid workspace protocol frame']]);
expect(onStatus).toHaveBeenLastCalledWith('closed', 'Workspace protocol frame rejected');
expect(onStatus).toHaveBeenLastCalledWith('closed', {
kind: 'protocol',
message: 'Workspace protocol frame rejected',
});
expect(JSON.stringify(onStatus.mock.calls)).not.toContain('untrusted detail');
vi.advanceTimersByTime(500);
@@ -108,9 +111,84 @@ test('invalid inbound frame closes before dispatch and reconnects for a fresh sn
second.open();
expect(second.sent).toHaveLength(1);
expect(JSON.parse(second.sent[0]).message.method).toBe('subscribe_events');
first.open();
first.message('{"protocol_version":2,"frame":"invalid"}');
expect(second.sent).toHaveLength(1);
expect(second.closes).toEqual([]);
multiplexer.dispose();
});
test('normal disposal closes silently and never schedules a reconnect', () => {
vi.useFakeTimers();
const multiplexer = new WorkspaceMultiplexer('workspace-dispose-test');
const onStatus = vi.fn();
multiplexer.subscribe(
{ topic: 'workspace_workers' },
{ onFrame: vi.fn(), onStatus },
);
const socket = FakeWebSocket.instances[0];
socket.open();
onStatus.mockClear();
multiplexer.dispose();
expect(socket.closes).toEqual([[undefined, undefined]]);
expect(onStatus).not.toHaveBeenCalled();
vi.advanceTimersByTime(1_000);
expect(FakeWebSocket.instances).toHaveLength(1);
});
test('terminal subscription closure does not retry or report normal socket cleanup as transport failure', () => {
vi.useFakeTimers();
const multiplexer = new WorkspaceMultiplexer('workspace-terminal-test');
const onFrame = vi.fn();
const onStatus = vi.fn();
multiplexer.subscribe(
{ topic: 'workspace_workers' },
{ onFrame, onStatus },
);
const socket = FakeWebSocket.instances[0];
socket.open();
const request = JSON.parse(socket.sent[0]);
socket.message(JSON.stringify({
protocol_version: 2,
frame: 'response',
message: {
result: 'subscribed',
payload: {
request_id: request.message.params.request_id,
subscription_id: 'subscription-terminal',
selector: { topic: 'workspace_workers' },
snapshot: { topic: 'workers', data: { workers: [] } },
},
},
}));
onStatus.mockClear();
const sentBeforeClosure = socket.sent.length;
socket.message(JSON.stringify({
protocol_version: 2,
frame: 'event',
message: {
event: 'subscription_closed',
data: {
subscription_id: 'subscription-terminal',
code: 'unauthorized',
message: 'Workspace Worker subscription is no longer authorized',
},
},
}));
expect(onFrame).toHaveBeenCalledTimes(2);
expect(onStatus).not.toHaveBeenCalled();
expect(socket.sent).toHaveLength(sentBeforeClosure);
expect(socket.closes).toEqual([[undefined, undefined]]);
vi.advanceTimersByTime(1_000);
expect(FakeWebSocket.instances).toHaveLength(1);
});
test('valid subscription closure remains isolated and resubscribes only its selector', () => {
const multiplexer = new WorkspaceMultiplexer('workspace-isolation-test');
const workersFrame = vi.fn();
@@ -173,15 +251,9 @@ test('valid subscription closure remains isolated and resubscribes only its sele
);
expect(workersFrame).toHaveBeenCalledTimes(2);
expect(workersStatus).toHaveBeenLastCalledWith(
'connecting',
'resubscribe for a fresh snapshot',
);
expect(workersStatus).toHaveBeenLastCalledWith('connecting');
expect(workdirsFrame).toHaveBeenCalledTimes(1);
expect(workdirsStatus).not.toHaveBeenCalledWith(
'connecting',
'resubscribe for a fresh snapshot',
);
expect(workdirsStatus).not.toHaveBeenCalledWith('connecting');
expect(socket.sent).toHaveLength(sentBeforeClosure + 1);
expect(JSON.parse(socket.sent.at(-1)!).message.params.selector).toEqual({
topic: 'workspace_workers',
+96 -49
View File
@@ -1,19 +1,27 @@
import { browser } from '$app/env';
import { browser } from "$app/env";
import type {
EventSubscriptionSelector,
Method,
SubscriptionFrame,
SubscriptionId,
} from '#lib/generated/protocol.ts';
} from "#lib/generated/protocol.ts";
import {
decodeSubscriptionFrame,
subscriptionFrameMatchesSelector,
} from '#lib/generated/protocol-validator.ts';
import { workspaceApiPath } from '#lib/workspace/api/http.ts';
} from "#lib/generated/protocol-validator.ts";
import { workspaceApiPath } from "#lib/workspace/api/http.ts";
export type WorkspaceMultiplexerFailure = {
kind: "protocol" | "transport";
message: string;
};
type Listener = {
onFrame(frame: SubscriptionFrame): void;
onStatus?(status: 'connecting' | 'open' | 'closed', message?: string): void;
onStatus?(
status: "connecting" | "open" | "closed",
failure?: WorkspaceMultiplexerFailure,
): void;
};
type ActiveSubscription = {
@@ -33,12 +41,14 @@ const multiplexers = new Map<string, WorkspaceMultiplexer>();
const INVALID_PROTOCOL_CLOSE_CODE = 4002;
let nextMultiplexerSequence = 0;
function nextMultiplexerId(kind: 'client' | 'request'): string {
function nextMultiplexerId(kind: "client" | "request"): string {
nextMultiplexerSequence += 1;
return `${kind}-${nextMultiplexerSequence}`;
}
export function workspaceMultiplexer(workspaceId: string): WorkspaceMultiplexer {
export function workspaceMultiplexer(
workspaceId: string,
): WorkspaceMultiplexer {
let multiplexer = multiplexers.get(workspaceId);
if (!multiplexer) {
multiplexer = new WorkspaceMultiplexer(workspaceId);
@@ -61,7 +71,7 @@ export class WorkspaceMultiplexer {
readonly #runtimeSubscriptions = new Map<string, string>();
#socket: WebSocket | null = null;
#reconnectTimer: ReturnType<typeof setTimeout> | null = null;
#disconnectMessage: string | null = null;
#disconnectFailure: WorkspaceMultiplexerFailure | null = null;
#closed = false;
constructor(workspaceId: string) {
@@ -72,7 +82,7 @@ export class WorkspaceMultiplexer {
selector: EventSubscriptionSelector,
listener: Listener,
): WorkspaceMultiplexerSubscription {
const clientId = nextMultiplexerId('client');
const clientId = nextMultiplexerId("client");
const subscription: ActiveSubscription = {
clientId,
selector,
@@ -83,7 +93,7 @@ export class WorkspaceMultiplexer {
this.#subscriptions.set(clientId, subscription);
this.#closed = false;
if (this.#socket?.readyState === WebSocket.OPEN) {
subscription.listener.onStatus?.('open');
subscription.listener.onStatus?.("open");
this.#sendSubscribe(subscription);
} else {
this.#ensureConnected();
@@ -95,37 +105,48 @@ export class WorkspaceMultiplexer {
}
#ensureConnected(): void {
if (!browser || this.#socket || this.#closed || this.#subscriptions.size === 0) return;
if (
!browser || this.#socket || this.#closed || this.#subscriptions.size === 0
) return;
for (const subscription of this.#subscriptions.values()) {
subscription.listener.onStatus?.('connecting');
subscription.listener.onStatus?.("connecting");
}
const url = new URL(
workspaceApiPath(this.#workspaceId, '/protocol/ws'),
workspaceApiPath(this.#workspaceId, "/protocol/ws"),
window.location.origin,
);
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
url.protocol = url.protocol === "https:" ? "wss:" : "ws:";
const socket = new WebSocket(url);
this.#socket = socket;
socket.addEventListener('open', () => {
socket.addEventListener("open", () => {
if (this.#socket !== socket) return;
for (const subscription of this.#subscriptions.values()) {
subscription.listener.onStatus?.('open');
subscription.listener.onStatus?.("open");
this.#sendSubscribe(subscription);
}
});
socket.addEventListener('message', (event) => this.#receive(event.data));
socket.addEventListener('error', () => socket.close());
socket.addEventListener('close', () => {
socket.addEventListener("message", (event) => {
if (this.#socket !== socket) return;
this.#receive(event.data);
});
socket.addEventListener("error", () => {
if (this.#socket !== socket) return;
socket.close();
});
socket.addEventListener("close", () => {
if (this.#socket !== socket) return;
this.#socket = null;
const disconnectMessage =
this.#disconnectMessage ?? 'Workspace subscription disconnected';
this.#disconnectMessage = null;
const failure = this.#disconnectFailure ?? {
kind: "transport" as const,
message: "Workspace subscription disconnected",
};
this.#disconnectFailure = null;
this.#requests.clear();
this.#runtimeSubscriptions.clear();
for (const subscription of this.#subscriptions.values()) {
subscription.requestId = null;
subscription.subscriptionId = null;
subscription.listener.onStatus?.('closed', disconnectMessage);
subscription.listener.onStatus?.("closed", failure);
}
if (!this.#closed && this.#subscriptions.size > 0) {
this.#reconnectTimer = setTimeout(() => this.#ensureConnected(), 500);
@@ -134,14 +155,14 @@ export class WorkspaceMultiplexer {
}
#sendSubscribe(subscription: ActiveSubscription): void {
const requestId = nextMultiplexerId('request');
const requestId = nextMultiplexerId("request");
subscription.requestId = requestId;
this.#requests.set(requestId, subscription.clientId);
this.#send({
protocol_version: 2,
frame: 'request',
frame: "request",
message: {
method: 'subscribe_events',
method: "subscribe_events",
params: { request_id: requestId, selector: subscription.selector },
},
});
@@ -154,9 +175,11 @@ export class WorkspaceMultiplexer {
return;
}
const frame = decoded.value;
if (frame.frame === 'response' && frame.message.result === 'subscribed') {
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;
const subscription = clientId
? this.#subscriptions.get(clientId)
: undefined;
if (!clientId || !subscription) return;
if (!subscriptionFrameMatchesSelector(frame, subscription.selector)) {
this.#rejectInboundFrame();
@@ -169,9 +192,14 @@ export class WorkspaceMultiplexer {
subscription.listener.onFrame(frame);
return;
}
if (frame.frame === 'response' && frame.message.result === 'subscription_rejected') {
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;
const subscription = clientId
? this.#subscriptions.get(clientId)
: undefined;
subscription?.listener.onFrame(frame);
if (clientId) {
this.#requests.delete(frame.message.payload.request_id);
@@ -179,9 +207,13 @@ export class WorkspaceMultiplexer {
}
return;
}
if (frame.frame === 'event') {
const clientId = this.#runtimeSubscriptions.get(frame.message.data.subscription_id);
const subscription = clientId ? this.#subscriptions.get(clientId) : undefined;
if (frame.frame === "event") {
const clientId = this.#runtimeSubscriptions.get(
frame.message.data.subscription_id,
);
const subscription = clientId
? this.#subscriptions.get(clientId)
: undefined;
if (
subscription &&
!subscriptionFrameMatchesSelector(frame, subscription.selector)
@@ -191,30 +223,45 @@ export class WorkspaceMultiplexer {
}
subscription?.listener.onFrame(frame);
if (
frame.message.event === 'subscription_closed' &&
frame.message.event === "subscription_closed" &&
clientId &&
subscription &&
this.#socket?.readyState === WebSocket.OPEN
subscription
) {
this.#runtimeSubscriptions.delete(frame.message.data.subscription_id);
subscription.subscriptionId = null;
subscription.listener.onStatus?.('connecting', frame.message.data.message);
this.#sendSubscribe(subscription);
if (
(frame.message.data.code === "lagged" ||
frame.message.data.code === "server_shutdown") &&
this.#socket?.readyState === WebSocket.OPEN
) {
subscription.listener.onStatus?.("connecting");
this.#sendSubscribe(subscription);
} else {
this.#remove(clientId);
}
}
}
}
#rejectInboundFrame(): void {
this.#disconnectMessage = 'Workspace protocol frame rejected';
this.#socket?.close(INVALID_PROTOCOL_CLOSE_CODE, 'Invalid workspace protocol frame');
this.#disconnectFailure = {
kind: "protocol",
message: "Workspace protocol frame rejected",
};
this.#socket?.close(
INVALID_PROTOCOL_CLOSE_CODE,
"Invalid workspace protocol frame",
);
}
#sendWorkerMethod(clientId: string, method: Method): void {
const subscription = this.#subscriptions.get(clientId);
if (!subscription?.subscriptionId) throw new Error('Worker protocol subscription is not open');
if (!subscription?.subscriptionId) {
throw new Error("Worker protocol subscription is not open");
}
this.#send({
protocol_version: 2,
frame: 'worker_protocol',
frame: "worker_protocol",
message: { subscription_id: subscription.subscriptionId, method },
});
}
@@ -223,14 +270,16 @@ export class WorkspaceMultiplexer {
const subscription = this.#subscriptions.get(clientId);
if (!subscription) return;
this.#subscriptions.delete(clientId);
if (subscription.subscriptionId && this.#socket?.readyState === WebSocket.OPEN) {
if (
subscription.subscriptionId && this.#socket?.readyState === WebSocket.OPEN
) {
this.#send({
protocol_version: 2,
frame: 'request',
frame: "request",
message: {
method: 'unsubscribe_events',
method: "unsubscribe_events",
params: {
request_id: nextMultiplexerId('request'),
request_id: nextMultiplexerId("request"),
subscription_id: subscription.subscriptionId,
},
},
@@ -251,12 +300,10 @@ export class WorkspaceMultiplexer {
clearTimeout(this.#reconnectTimer);
this.#reconnectTimer = null;
}
for (const subscription of this.#subscriptions.values()) {
subscription.listener.onStatus?.('closed', 'Workspace selection changed');
}
this.#subscriptions.clear();
this.#requests.clear();
this.#runtimeSubscriptions.clear();
this.#disconnectFailure = null;
this.#socket?.close();
this.#socket = null;
}
@@ -44,10 +44,6 @@
<section class="sidebar-nav-section sidebar-nav-section--category" aria-labelledby="global-workspaces-heading">
<h2 id="global-workspaces-heading" class="sidebar-nav-section__header">workspaces</h2>
{#if workspaceError}
<p class="workspace-status error">Workspace list unavailable.</p>
{/if}
{#if workspaces.length > 0}
<div class="sidebar-list">
{#each workspaces as workspace (workspace.workspace_id)}
@@ -30,7 +30,6 @@
let { currentPath = '/', workspaceId }: Props = $props();
let loading = $state(true);
let error = $state<string | null>(null);
let workers = $state<SidebarWorker[]>([]);
let expanded = $state(false);
let openWorkerKey = $state<string | null>(null);
@@ -141,7 +140,6 @@
const subscription = workspaceWorkersStore(workspaceId);
return subscription.subscribe((state) => {
loading = state.loading;
error = state.error;
workers = state.workers.filter(canShowWorkerInSidebar);
});
});
@@ -167,7 +165,7 @@
>
New
</a>
{#if !loading && !error && workers.length > 0}
{#if !loading && workers.length > 0}
<span class="section-count">{workers.length}</span>
{/if}
</div>
@@ -175,9 +173,8 @@
{#if loading}
<p class="section-state">Checking workers…</p>
{:else if workers.length === 0}
<p class="section-state" class:error={Boolean(error)}>{error ?? 'No Workers are active.'}</p>
<p class="section-state">No Workers are active.</p>
{:else}
{#if error}<p class="section-state error">{error}</p>{/if}
<ul class="nav-list" aria-label="Workers">
{#each visibleWorkers as worker (`${worker.runtime_id}:${worker.worker_id}`)}
{@const href = workerConsoleHref(workspaceId, worker)}
@@ -66,9 +66,7 @@
{:else}
<div class="workspace-label">
<div class="workspace-name">Yoi workspace</div>
{#if workspaceError}
<p class="workspace-status error">Workspace summary unavailable.</p>
{:else}
{#if !workspaceError}
<p class="workspace-status">Loading workspace…</p>
{/if}
</div>
@@ -1,5 +1,6 @@
<script lang="ts">
import { onMount, tick } from "svelte";
import { dismissWorkspaceAlert, pushWorkspaceAlert } from "#lib/workspace/alerts/store.ts";
import {
listWorkspaces,
type WorkspaceCatalogRecord,
@@ -17,11 +18,11 @@
let workspaces = $state<WorkspaceCatalogRecord[]>([]);
let loading = $state(true);
let error = $state("");
let open = $state(false);
let root = $state.raw<HTMLDivElement>();
let trigger = $state.raw<HTMLButtonElement>();
let menu = $state.raw<HTMLDivElement>();
let loadGeneration = 0;
const menuId = $derived(`workspace-menu-popover-${variant}`);
const menuWorkspaces = $derived.by(() => {
@@ -39,14 +40,22 @@
});
async function loadWorkspaces() {
const generation = ++loadGeneration;
loading = true;
error = "";
try {
workspaces = await listWorkspaces(fetch);
const nextWorkspaces = await listWorkspaces(fetch);
if (generation !== loadGeneration) return;
workspaces = nextWorkspaces;
dismissWorkspaceAlert("workspace-catalog-load");
} catch (cause) {
error = cause instanceof Error ? cause.message : "Failed to load Workspaces.";
if (generation !== loadGeneration) return;
pushWorkspaceAlert(
"error",
cause instanceof Error ? cause.message : "Failed to load Workspaces.",
{ id: "workspace-catalog-load", title: "Workspace list unavailable" },
);
} finally {
loading = false;
if (generation === loadGeneration) loading = false;
}
}
@@ -111,7 +120,10 @@
onMount(() => {
document.addEventListener("pointerdown", handleDocumentPointerDown);
void loadWorkspaces();
return () => document.removeEventListener("pointerdown", handleDocumentPointerDown);
return () => {
loadGeneration += 1;
document.removeEventListener("pointerdown", handleDocumentPointerDown);
};
});
</script>
@@ -191,8 +203,6 @@
{#if loading}
<p class="workspace-menu-status">Loading Workspaces…</p>
{:else if error}
<p class="workspace-menu-status error">{error}</p>
{/if}
</div>
{/if}
@@ -0,0 +1,303 @@
// @vitest-environment happy-dom
import { cleanup, render, screen, waitFor } from "@testing-library/svelte";
import { afterEach, beforeEach, expect, test, vi } from "vitest";
import type {
SubscriptionFrame,
SubscriptionWorker,
} from "#lib/generated/protocol.ts";
import WorkspaceAlerts from "#lib/workspace/alerts/WorkspaceAlerts.svelte";
import {
clearWorkspaceAlerts,
workspaceAlerts,
} from "#lib/workspace/alerts/store.ts";
import WorkersNavSection from "./WorkersNavSection.svelte";
import {
disposeWorkspaceWorkersStore,
workspaceWorkersStore,
} from "./worker-subscription";
const transport = vi.hoisted(() => {
type Listener = {
onFrame(frame: SubscriptionFrame): void;
onStatus?(
status: "connecting" | "open" | "closed",
failure?: { kind: "protocol" | "transport"; message: string },
): void;
};
const listeners = new Map<string, Listener[]>();
const closes = new Map<string, ReturnType<typeof vi.fn>>();
return {
listeners,
closes,
subscribe: vi.fn((workspaceId: string, listener: Listener) => {
const workspaceListeners = listeners.get(workspaceId) ?? [];
workspaceListeners.push(listener);
listeners.set(workspaceId, workspaceListeners);
const close = closes.get(workspaceId) ?? vi.fn();
closes.set(workspaceId, close);
listener.onStatus?.("connecting");
return { close, sendWorkerMethod: vi.fn() };
}),
};
});
vi.mock("#lib/workspace/multiplexer.ts", () => ({
workspaceMultiplexer: (workspaceId: string) => ({
subscribe: (
_selector: unknown,
listener: Parameters<typeof transport.subscribe>[1],
) => transport.subscribe(workspaceId, listener),
}),
}));
beforeEach(() => {
transport.listeners.clear();
transport.closes.clear();
transport.subscribe.mockClear();
clearWorkspaceAlerts();
vi.stubGlobal(
"fetch",
vi.fn(async (input: RequestInfo | URL) => {
const workspaceId = decodeURIComponent(
new URL(String(input), "https://example.test").pathname.split("/")[3] ??
"",
);
return Response.json({
workspace_id: workspaceId,
items: [],
diagnostics: [],
});
}),
);
});
afterEach(() => {
cleanup();
for (const workspaceId of transport.listeners.keys()) {
disposeWorkspaceWorkersStore(workspaceId);
}
clearWorkspaceAlerts();
vi.unstubAllGlobals();
});
function worker(workspaceId: string, workerId: string): SubscriptionWorker {
return {
worker_id: workerId,
runtime_id: `runtime-${workspaceId}`,
resource_key: `W-${workerId}`,
availability: "observed",
state: "idle",
worker_state: null,
has_running_internal_workers: false,
display_name: `Worker ${workerId}`,
workspace_id: workspaceId,
workdir_attachments: [],
};
}
function snapshot(
workspaceId: string,
subscriptionId: string,
workers: SubscriptionWorker[],
): SubscriptionFrame {
return {
protocol_version: 2,
frame: "response",
message: {
result: "subscribed",
payload: {
request_id: `request-${subscriptionId}`,
subscription_id: subscriptionId,
selector: { topic: "workspace_workers" },
snapshot: { topic: "workers", data: { workers } },
},
},
};
}
function latestListener(workspaceId: string) {
const listeners = transport.listeners.get(workspaceId) ?? [];
const listener = listeners.at(-1);
if (!listener) throw new Error(`missing listener for ${workspaceId}`);
return listener;
}
function mount(workspaceId = "workspace-a") {
render(WorkspaceAlerts);
return render(WorkersNavSection, {
workspaceId,
currentPath: `/w/${workspaceId}`,
});
}
function alerts(): Element[] {
return [...document.querySelectorAll(".workspace-alert")];
}
function deferred<T>() {
let resolve!: (value: T) => void;
const promise = new Promise<T>((done) => {
resolve = done;
});
return { promise, resolve };
}
test("explicit store disposal closes and fences pending fetch and subscription callbacks", async () => {
const pendingFetch = deferred<Response>();
vi.stubGlobal("fetch", vi.fn(() => pendingFetch.promise));
render(WorkspaceAlerts);
const states = vi.fn();
const unsubscribe = workspaceWorkersStore("workspace-a").subscribe(states);
await waitFor(() => expect(transport.subscribe).toHaveBeenCalledOnce());
const listener = latestListener("workspace-a");
disposeWorkspaceWorkersStore("workspace-a");
expect(transport.closes.get("workspace-a")).toHaveBeenCalledOnce();
listener.onStatus?.("closed", {
kind: "transport",
message: "late transport failure",
});
listener.onFrame(
snapshot("workspace-a", "subscription-late", [
worker("workspace-a", "late"),
]),
);
pendingFetch.resolve(new Response("late fetch failure", { status: 503 }));
await pendingFetch.promise;
await Promise.resolve();
expect(alerts()).toHaveLength(0);
expect(JSON.stringify(states.mock.calls)).not.toContain("late");
unsubscribe();
});
test("disconnect and reconnect keep Worker rows, avoid sidebar diagnostics, and dedupe the alert episode", async () => {
mount();
await waitFor(() => expect(transport.subscribe).toHaveBeenCalledOnce());
const listener = latestListener("workspace-a");
listener.onFrame(
snapshot("workspace-a", "subscription-a", [worker("workspace-a", "old")]),
);
expect(await screen.findByText("Worker old")).toBeTruthy();
const failure = { kind: "transport" as const, message: "connection lost" };
const alertSnapshots = vi.fn();
const unsubscribeAlerts = workspaceAlerts.subscribe(alertSnapshots);
alertSnapshots.mockClear();
listener.onStatus?.("closed", failure);
listener.onStatus?.("closed", failure);
expect(alertSnapshots).toHaveBeenCalledOnce();
unsubscribeAlerts();
expect(screen.getByText("Worker old")).toBeTruthy();
expect(document.querySelector(".sidebar-nav-section")?.textContent).not
.toContain(
"connection lost",
);
await waitFor(() => expect(alerts()).toHaveLength(1));
expect(alerts()[0]?.textContent).toContain("connection lost");
listener.onStatus?.("connecting");
listener.onFrame(
snapshot("workspace-a", "subscription-b", [worker("workspace-a", "new")]),
);
await waitFor(() => expect(screen.getByText("Worker new")).toBeTruthy());
expect(screen.queryByText("Worker old")).toBeNull();
expect(alerts()).toHaveLength(0);
});
test("route switch silently fences delayed old Workspace callbacks and starts the new subscription", async () => {
const view = mount("workspace-a");
await waitFor(() => expect(transport.subscribe).toHaveBeenCalledOnce());
const oldListener = latestListener("workspace-a");
oldListener.onFrame(
snapshot("workspace-a", "subscription-a", [worker("workspace-a", "a")]),
);
expect(await screen.findByText("Worker a")).toBeTruthy();
await view.rerender({
workspaceId: "workspace-a",
currentPath: "/w/workspace-a/workers",
});
await waitFor(() => expect(transport.subscribe).toHaveBeenCalledTimes(2));
expect(transport.closes.get("workspace-a")).toHaveBeenCalledTimes(1);
latestListener("workspace-a").onFrame(
snapshot("workspace-a", "subscription-a2", [worker("workspace-a", "same")]),
);
expect(await screen.findByText("Worker same")).toBeTruthy();
expect(alerts()).toHaveLength(0);
await view.rerender({
workspaceId: "workspace-b",
currentPath: "/w/workspace-b",
});
await waitFor(() => expect(transport.subscribe).toHaveBeenCalledTimes(3));
expect(transport.closes.get("workspace-a")).toHaveBeenCalledTimes(2);
oldListener.onStatus?.("closed", {
kind: "protocol",
message: "OLD WORKSPACE FAILURE",
});
oldListener.onFrame(
snapshot("workspace-a", "subscription-late", [
worker("workspace-a", "late"),
]),
);
latestListener("workspace-b").onFrame(
snapshot("workspace-b", "subscription-b", [worker("workspace-b", "b")]),
);
expect(await screen.findByText("Worker b")).toBeTruthy();
expect(screen.queryByText("Worker late")).toBeNull();
expect(document.body.textContent).not.toContain("OLD WORKSPACE FAILURE");
expect(alerts()).toHaveLength(0);
});
test("terminal subscription rejection is shown only through WorkspaceAlerts", async () => {
mount();
await waitFor(() => expect(transport.subscribe).toHaveBeenCalledOnce());
latestListener("workspace-a").onFrame({
protocol_version: 2,
frame: "response",
message: {
result: "subscription_rejected",
payload: {
request_id: "request-rejected",
code: "unauthorized",
message: "Worker subscription denied",
},
},
});
expect(await screen.findByText("No Workers are active.")).toBeTruthy();
expect(document.querySelector(".sidebar-nav-section")?.textContent).not
.toContain(
"Worker subscription denied",
);
expect(alerts()).toHaveLength(1);
expect(alerts()[0]?.textContent).toContain("Worker subscription denied");
});
test("Workdir fetch failure preserves Worker updates and uses a separate stable alert", async () => {
vi.stubGlobal(
"fetch",
vi.fn(async () => new Response("Workdir lookup failed", { status: 503 })),
);
mount();
await waitFor(() => expect(transport.subscribe).toHaveBeenCalledOnce());
latestListener("workspace-a").onFrame(
snapshot("workspace-a", "subscription-a", [
worker("workspace-a", "visible"),
]),
);
expect(await screen.findByText("Worker visible")).toBeTruthy();
await waitFor(() => expect(alerts()).toHaveLength(1));
expect(alerts()[0]?.textContent).toContain("Workdir lookup failed");
expect(document.querySelector(".sidebar-nav-section")?.textContent).not
.toContain(
"Workdir lookup failed",
);
});
@@ -1,9 +1,17 @@
import { type Readable, readable } from "svelte/store";
import type { WorkingDirectorySummary } from "#lib/generated/workdir-api.ts";
import type { SubscriptionWorker } from "#lib/generated/protocol.ts";
import {
dismissWorkspaceAlert,
pushWorkspaceAlert,
} from "#lib/workspace/alerts/store.ts";
import { loadJson, workspaceApiPath } from "#lib/workspace/api/http.ts";
import { parseWorkingDirectoryListResponse } from "#lib/workspace/api/workdirs.ts";
import { workspaceMultiplexer } from "#lib/workspace/multiplexer.ts";
import {
workspaceMultiplexer,
type WorkspaceMultiplexerFailure,
type WorkspaceMultiplexerSubscription,
} from "#lib/workspace/multiplexer.ts";
import {
applyWorkspaceWorkersFrame,
createWorkspaceWorkersProjection,
@@ -22,105 +30,213 @@ export type SidebarWorker = Omit<Worker, "workdir_attachments"> & {
export type WorkspaceWorkersState = {
loading: boolean;
error: string | null;
workers: SidebarWorker[];
};
const stores = new Map<string, Readable<WorkspaceWorkersState>>();
type WorkerAlertKind = "subscription" | "workdirs";
type WorkspaceWorkersStoreEntry = {
store: Readable<WorkspaceWorkersState>;
dispose(): void;
};
const stores = new Map<string, WorkspaceWorkersStoreEntry>();
function workerAlertId(workspaceId: string, kind: WorkerAlertKind): string {
return `workspace:${workspaceId}:workers:${kind}`;
}
function reportWorkerFailure(
workspaceId: string,
kind: WorkerAlertKind,
message: string,
level: "warning" | "error" = "error",
): void {
pushWorkspaceAlert(level, message, {
id: workerAlertId(workspaceId, kind),
title: kind === "workdirs"
? "Worker Workdirs unavailable"
: "Worker updates unavailable",
});
}
function clearWorkerFailure(workspaceId: string, kind: WorkerAlertKind): void {
dismissWorkspaceAlert(workerAlertId(workspaceId, kind));
}
function clearWorkerFailures(workspaceId: string): void {
clearWorkerFailure(workspaceId, "subscription");
clearWorkerFailure(workspaceId, "workdirs");
}
export function disposeWorkspaceWorkersStore(workspaceId: string): void {
const entry = stores.get(workspaceId);
if (!entry) {
clearWorkerFailures(workspaceId);
return;
}
stores.delete(workspaceId);
entry.dispose();
}
export function workspaceWorkersStore(
workspaceId: string,
): Readable<WorkspaceWorkersState> {
const cached = stores.get(workspaceId);
if (cached) return cached;
if (cached) return cached.store;
let stopActive: (() => void) | null = null;
const store = readable<WorkspaceWorkersState>(
{ loading: true, error: null, workers: [] },
{ loading: true, workers: [] },
(set) => {
if (!workspaceId) {
set({ loading: false, error: null, workers: [] });
set({ loading: false, workers: [] });
return;
}
const projection = createWorkspaceWorkersProjection();
const workdirRequest = new AbortController();
let subscription: WorkspaceMultiplexerSubscription | null = null;
let workdirs = new Map<string, WorkingDirectorySummary>();
let workers: SidebarWorker[] = [];
let loading = true;
let error: string | null = null;
let disposed = false;
const publish = () => {
const workers = [...projection.workers.values()]
.map((worker) => projectWorker(worker, workdirs))
.sort(compareWorkersForSidebar);
set({ loading, error, workers });
const publish = (): boolean => {
if (disposed) return false;
try {
workers = [...projection.workers.values()]
.map((worker) => projectWorker(worker, workdirs))
.sort(compareWorkersForSidebar);
set({ loading, workers });
return true;
} catch (cause) {
reportWorkerFailure(
workspaceId,
"subscription",
cause instanceof Error
? cause.message
: "Invalid Worker subscription frame",
);
set({ loading: false, workers });
return false;
}
};
const loadWorkdirs = async () => {
const result = await loadJson(
fetch,
workspaceApiPath(workspaceId, "/working-directories"),
undefined,
{ signal: workdirRequest.signal },
parseWorkingDirectoryListResponse,
);
if (disposed || !result.data) return;
if (disposed) return;
if (!result.data) {
reportWorkerFailure(
workspaceId,
"workdirs",
result.error ?? "Worker Workdirs could not be loaded",
"warning",
);
return;
}
workdirs = new Map(
result.data.items.map((
workdir,
) => [workdir.working_directory_id, workdir]),
);
clearWorkerFailure(workspaceId, "workdirs");
publish();
};
const subscription = workspaceMultiplexer(workspaceId).subscribe(
subscription = workspaceMultiplexer(workspaceId).subscribe(
{ topic: "workspace_workers" },
{
onFrame: (frame) => {
if (disposed) return;
if (
frame.frame === "event" &&
frame.message.event === "subscription_closed"
) {
loading = projection.workers.size === 0;
if (
frame.message.data.code === "resource_gone" ||
frame.message.data.code === "unauthorized"
) {
reportWorkerFailure(
workspaceId,
"subscription",
frame.message.data.message,
);
}
publish();
return;
}
if (
frame.frame === "response" &&
frame.message.result === "subscription_rejected"
) {
loading = false;
reportWorkerFailure(
workspaceId,
"subscription",
frame.message.payload.message,
);
publish();
return;
}
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);
loading = false;
error = null;
publish();
if (publish()) clearWorkerFailure(workspaceId, "subscription");
} catch (cause) {
loading = false;
error = cause instanceof Error
? cause.message
: "invalid Worker subscription frame";
reportWorkerFailure(
workspaceId,
"subscription",
cause instanceof Error
? cause.message
: "Invalid Worker subscription frame",
);
publish();
}
},
onStatus: (status, message) => {
onStatus: (status, failure?: WorkspaceMultiplexerFailure) => {
if (disposed) return;
if (status === "connecting") {
loading = projection.workers.size === 0;
error = null;
publish();
}
if (status === "closed") {
if (status === "closed" && failure) {
loading = projection.workers.size === 0;
error = message ?? null;
reportWorkerFailure(
workspaceId,
"subscription",
failure.message,
failure.kind === "transport" ? "warning" : "error",
);
publish();
}
},
},
);
void loadWorkdirs();
return () => {
const stop = () => {
if (disposed) return;
disposed = true;
subscription.close();
workdirRequest.abort();
subscription?.close();
clearWorkerFailures(workspaceId);
if (stopActive === stop) stopActive = null;
};
stopActive = stop;
return stop;
},
);
stores.set(workspaceId, store);
const entry: WorkspaceWorkersStoreEntry = {
store,
dispose: () => {
stopActive?.();
clearWorkerFailures(workspaceId);
},
};
stores.set(workspaceId, entry);
return store;
}
@@ -0,0 +1,70 @@
// @vitest-environment happy-dom
import {
cleanup,
fireEvent,
render,
screen,
waitFor,
} from "@testing-library/svelte";
import { afterEach, expect, test, vi } from "vitest";
import WorkspaceAlerts from "#lib/workspace/alerts/WorkspaceAlerts.svelte";
import { clearWorkspaceAlerts } from "#lib/workspace/alerts/store.ts";
import WorkspaceSwitcher from "./WorkspaceSwitcher.svelte";
afterEach(() => {
cleanup();
clearWorkspaceAlerts();
vi.unstubAllGlobals();
});
test("Workspace list fetch failure uses WorkspaceAlerts and keeps the menu free of diagnostics", async () => {
vi.stubGlobal(
"fetch",
vi.fn(async () =>
new Response("Catalog temporarily unavailable", { status: 503 })
),
);
render(WorkspaceAlerts);
render(WorkspaceSwitcher, {
currentWorkspaceId: "workspace-a",
currentWorkspaceName: "Workspace A",
variant: "header",
});
const alert = await screen.findByRole("alert");
expect(alert.textContent).toContain("Workspace list unavailable");
expect(alert.textContent).toContain("503");
await fireEvent.click(screen.getByRole("button", { name: "Workspace A" }));
const menu = screen.getByRole("menu", { name: "Workspace menu" });
expect(menu.textContent).not.toContain("503");
expect(menu.textContent).not.toContain("Workspace list unavailable");
expect(screen.getByRole("menuitem", { name: "Workspace A" })).toBeTruthy();
});
test("unmounted Workspace menu ignores a delayed catalog failure", async () => {
let reject!: (cause: Error) => void;
vi.stubGlobal(
"fetch",
vi.fn(() =>
new Promise<Response>((_resolve, rejectPromise) => {
reject = rejectPromise;
})
),
);
render(WorkspaceAlerts);
const view = render(WorkspaceSwitcher, {
currentWorkspaceId: "workspace-a",
currentWorkspaceName: "Workspace A",
});
await waitFor(() => expect(fetch).toHaveBeenCalledOnce());
view.unmount();
reject(new Error("OLD WORKSPACE CATALOG FAILURE"));
await Promise.resolve();
await Promise.resolve();
expect(document.body.textContent).not.toContain(
"OLD WORKSPACE CATALOG FAILURE",
);
expect(document.querySelectorAll(".workspace-alert")).toHaveLength(0);
});
+13
View File
@@ -3,6 +3,7 @@
import { setContext } from 'svelte';
import { MediaQuery } from 'svelte/reactivity';
import WorkspaceAlerts from '#lib/workspace/alerts/WorkspaceAlerts.svelte';
import { dismissWorkspaceAlert, pushWorkspaceAlert } from '#lib/workspace/alerts/store.ts';
import Bevel from '#lib/workspace/ui/Bevel.svelte';
import {
provideHeaderController,
@@ -45,6 +46,18 @@
const mobileLayout = new MediaQuery('(max-width: 760px)');
const sidebarOpen = $derived((!mobileLayout.current && sidebarMode === 'pinned') || sidebarTransientOpen);
$effect(() => {
const error = data.workspaceCatalogError;
if (error) {
pushWorkspaceAlert('error', error, {
id: 'workspace-catalog-load',
title: 'Workspace list unavailable',
});
} else {
dismissWorkspaceAlert('workspace-catalog-load');
}
});
$effect(() => {
const mode = sidebarMode;
try {
@@ -30,6 +30,23 @@ afterEach(() => {
localStorage.clear();
});
test("Workspace catalog failure uses the shared alert without rendering diagnostics in the sidebar", async () => {
render(Layout, {
props: {
...props,
data: {
accessibleWorkspaces: [],
workspaceCatalogError: "Workspace catalog request failed",
},
},
});
expect((await screen.findByRole("alert")).textContent).toContain(
"Workspace catalog request failed",
);
expect(frame().textContent).not.toContain("Workspace catalog request failed");
expect(frame().textContent).not.toContain("Workspace list unavailable");
});
test("unpinning changes mode, and pointer departure closes the preview", async () => {
render(Layout, { props });
expect(screen.getByRole("button", { name: "Unpin sidebar" }).getAttribute("aria-pressed")).toBe("true");
@@ -1,5 +1,5 @@
<script lang="ts">
import { setContext } from 'svelte';
import { onDestroy, setContext } from 'svelte';
import { page } from '$app/state';
import HeaderOverride from '#lib/workspace/header/HeaderOverride.svelte';
import WorkspaceBreadcrumbs from '#lib/workspace/header/WorkspaceBreadcrumbs.svelte';
@@ -37,13 +37,23 @@
workspaceId !== '' && ownsRoutePath(workspaceRoute(workspaceId), page.url.pathname),
);
let activeWorkspaceId: string | null = null;
function disposeWorkspaceResources(workspaceId: string): void {
disposeWorkspaceWorkersStore(workspaceId);
disposeWorkspaceMultiplexer(workspaceId);
}
$effect(() => {
const workspaceId = data.workspace?.workspace_id;
if (!workspaceId) return;
return () => {
disposeWorkspaceMultiplexer(workspaceId);
disposeWorkspaceWorkersStore(workspaceId);
};
const nextWorkspaceId = data.workspace?.workspace_id ?? null;
if (nextWorkspaceId === activeWorkspaceId) return;
const previousWorkspaceId = activeWorkspaceId;
activeWorkspaceId = nextWorkspaceId;
if (previousWorkspaceId) disposeWorkspaceResources(previousWorkspaceId);
});
onDestroy(() => {
if (activeWorkspaceId) disposeWorkspaceResources(activeWorkspaceId);
});
</script>
@@ -1387,14 +1387,14 @@
];
}
},
onStatus: (status, message) => {
onStatus: (status, failure) => {
if (token !== reloadToken) return;
protocolState = status === "open" ? "connecting" : status;
if (status === "connecting" || status === "open") {
showConsoleReconnecting(message);
showConsoleReconnecting(failure?.message);
}
if (status === "closed") {
showConsoleFailure(message);
showConsoleFailure(failure?.message);
rejectPendingCompletion(new Error("Worker protocol WebSocket closed."));
}
},
@@ -25,7 +25,10 @@ vi.mock("svelte/motion", () => ({ prefersReducedMotion: { current: true } }));
const multiplexer = vi.hoisted(() => {
type Listener = {
onFrame(frame: unknown): void;
onStatus?(status: "connecting" | "open" | "closed", message?: string): void;
onStatus?(
status: "connecting" | "open" | "closed",
failure?: { kind: "protocol" | "transport"; message: string },
): void;
};
const listeners: Listener[] = [];
return {
@@ -393,7 +396,7 @@ test("queue-only state hides notifications and disables cancellation when discon
await screen.findByText("送信内容");
expect(screen.queryByRole("group", { name: "Notifications" })).toBeNull();
expect(screen.queryByRole("heading", { name: "Notifications" })).toBeNull();
latestListener().onStatus?.("closed", "connection lost");
latestListener().onStatus?.("closed", { kind: "transport", message: "connection lost" });
await waitFor(() => expect((screen.getByRole("button", { name: "Cancel queued input 1" }) as HTMLButtonElement).disabled).toBe(true));
});
@@ -450,7 +453,7 @@ test("file completions consume stale responses before the latest prefix and clos
expect(screen.queryByRole("option")).toBeNull();
receive("new-result");
expect((await screen.findByRole("option")).textContent).toContain("@new-result");
latestListener().onStatus?.("closed", "connection lost");
latestListener().onStatus?.("closed", { kind: "transport", message: "connection lost" });
await waitFor(() => expect(screen.queryByRole("listbox")).toBeNull());
receive("late-result");
await settleMicrotasks();
@@ -689,7 +692,10 @@ test("distinguishes typed unavailability and an initial live connection closure"
);
await fireEvent.click(screen.getByRole("button", { name: "Retry" }));
await waitFor(() => expect(multiplexer.subscribe).toHaveBeenCalledOnce());
latestListener().onStatus?.("closed", "subscription ended before snapshot");
latestListener().onStatus?.("closed", {
kind: "transport",
message: "subscription ended before snapshot",
});
expect((await screen.findByRole("alert")).textContent).toContain(
"subscription ended before snapshot",
@@ -714,7 +720,10 @@ test("preserves rendered content while reconnecting", async () => {
.findByText("kept message"),
).not.toBeNull();
latestListener().onStatus?.("closed", "Workspace subscription disconnected");
latestListener().onStatus?.("closed", {
kind: "transport",
message: "Workspace subscription disconnected",
});
expect(await screen.findByText("Conversation updates are unavailable")).not
.toBeNull();
expect(
@@ -724,7 +733,7 @@ test("preserves rendered content while reconnecting", async () => {
).not.toBeNull();
expect(screen.queryByText("No conversation to display")).toBeNull();
latestListener().onStatus?.("connecting", "resubscribing");
latestListener().onStatus?.("connecting");
expect(await screen.findByText("Refreshing conversation")).not.toBeNull();
expect(
within(screen.getByRole("article", { name: "main transcript" })).getByText(
@@ -761,8 +770,11 @@ test("reconnect snapshot refreshes lineage and removes stale sibling history", a
expectHistoryTurn("question 1")
);
latestListener().onStatus?.("closed", "disconnected during rewind");
latestListener().onStatus?.("connecting", "reconnecting");
latestListener().onStatus?.("closed", {
kind: "transport",
message: "disconnected during rewind",
});
latestListener().onStatus?.("connecting");
latestListener().onFrame(
subscribedFrame(sessionWithUserMessage("lineage B current")),
);
@@ -898,7 +910,10 @@ test("route changes clear prior content and fence stale Session responses", asyn
.findByText("previous worker content"),
).not.toBeNull();
latestListener().onStatus?.("closed", "reload old worker");
latestListener().onStatus?.("closed", {
kind: "transport",
message: "reload old worker",
});
await fireEvent.click(
await screen.findByRole("button", { name: "Retry" }),
);
@@ -0,0 +1,91 @@
// @vitest-environment happy-dom
import { cleanup, render } from "@testing-library/svelte";
import { afterEach, expect, test, vi } from "vitest";
import { createRawSnippet } from "svelte";
import Layout from "./+layout.svelte";
const resources = vi.hoisted(() => {
const events: string[] = [];
return {
events,
disposeMultiplexer: vi.fn((workspaceId: string) => {
events.push(`multiplexer:${workspaceId}`);
}),
disposeWorkers: vi.fn((workspaceId: string) => {
events.push(`workers:${workspaceId}`);
}),
};
});
vi.mock("$app/state", () => ({
page: {
params: { workspaceId: "workspace-a" },
url: new URL("https://example.test/w/workspace-a"),
},
}));
vi.mock("#lib/workspace/multiplexer.ts", () => ({
disposeWorkspaceMultiplexer: resources.disposeMultiplexer,
}));
vi.mock("#lib/workspace/sidebar/worker-subscription.ts", () => ({
disposeWorkspaceWorkersStore: resources.disposeWorkers,
workspaceWorkersStore: () => ({
subscribe: (
subscriber: (state: { loading: boolean; workers: never[] }) => void,
) => {
subscriber({ loading: false, workers: [] });
return () => {};
},
}),
}));
vi.mock("#lib/workspace/sidebar/context.ts", () => ({
SIDEBAR_CONTEXT: Symbol("test-sidebar"),
getSidebarController: () => ({ registerSidebar: () => () => {} }),
}));
vi.mock("#lib/workspace/header/context.ts", () => ({
getHeaderController: () => ({ registerContent: () => () => {} }),
}));
const children = createRawSnippet(() => ({ render: () => "<p>content</p>" }));
function data(workspaceId: string, displayName = workspaceId) {
return {
workspace: { workspace_id: workspaceId, display_name: displayName },
workspaceError: null,
repositories: [],
repositoriesError: null,
};
}
afterEach(() => {
cleanup();
resources.events.length = 0;
resources.disposeMultiplexer.mockClear();
resources.disposeWorkers.mockClear();
});
test("same-Workspace layout data refresh keeps resources, while route switch and exit dispose in order", async () => {
const view = render(Layout, {
props: { data: data("workspace-a"), children } as never,
});
expect(resources.events).toEqual([]);
await view.rerender({
data: data("workspace-a", "Updated display name"),
children,
} as never);
expect(resources.events).toEqual([]);
await view.rerender({ data: data("workspace-b"), children } as never);
expect(resources.events).toEqual([
"workers:workspace-a",
"multiplexer:workspace-a",
]);
view.unmount();
expect(resources.events).toEqual([
"workers:workspace-a",
"multiplexer:workspace-a",
"workers:workspace-b",
"multiplexer:workspace-b",
]);
});