@@ -94,11 +51,10 @@
{#if loading}
Checking workers…
- {:else if error}
- {error}
{:else if workers.length === 0}
- {placeholder ?? 'Workers will appear here when an API is connected.'}
+ {error ?? 'No Workers are active.'}
{:else}
+ {#if error}{error}
{/if}
{#each workers as worker (`${worker.runtime_id}:${worker.worker_id}`)}
{@const href = workerConsoleHref(worker, workspaceId)}
diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription-model.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription-model.ts
new file mode 100644
index 00000000..e2df7b0d
--- /dev/null
+++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription-model.ts
@@ -0,0 +1,60 @@
+import type {
+ SubscriptionEventPayload,
+ SubscriptionFrame,
+ SubscriptionWorker,
+} from '$lib/generated/protocol';
+
+export type WorkspaceWorkersProjection = {
+ workers: Map;
+ revisions: Map;
+};
+
+export function createWorkspaceWorkersProjection(): WorkspaceWorkersProjection {
+ return { workers: new Map(), revisions: new Map() };
+}
+
+export function applyWorkspaceWorkersFrame(
+ projection: WorkspaceWorkersProjection,
+ frame: SubscriptionFrame,
+): void {
+ if (frame.protocol_version !== 1) throw new Error('unsupported Worker subscription protocol');
+ if (frame.frame === 'response' && frame.message.result === 'subscribed') {
+ if (frame.message.payload.selector.topic !== 'workspace_workers') return;
+ const snapshot = frame.message.payload.snapshot;
+ if (snapshot.topic !== 'workers') throw new Error('workspace_workers returned a non-Worker snapshot');
+ projection.workers.clear();
+ projection.revisions.clear();
+ for (const worker of snapshot.data.workers) {
+ const key = workerKey(worker.runtime_id, worker.worker_id);
+ projection.workers.set(key, worker);
+ projection.revisions.set(key, worker.subject_revision);
+ }
+ return;
+ }
+ if (frame.frame !== 'event' || frame.message.event !== 'event') return;
+ applyPayload(projection, frame.message.data.subject_revision, frame.message.data.payload);
+}
+
+function applyPayload(
+ projection: WorkspaceWorkersProjection,
+ subjectRevision: number,
+ payload: SubscriptionEventPayload,
+): void {
+ if (payload.event === 'worker_upserted') {
+ const worker = payload.data.worker;
+ const key = workerKey(worker.runtime_id, worker.worker_id);
+ if (subjectRevision <= (projection.revisions.get(key) ?? 0)) return;
+ projection.revisions.set(key, subjectRevision);
+ projection.workers.set(key, worker);
+ } else if (payload.event === 'worker_removed') {
+ const key = workerKey(payload.data.runtime_id, payload.data.worker_id);
+ if (subjectRevision <= (projection.revisions.get(key) ?? 0)) return;
+ projection.revisions.set(key, subjectRevision);
+ projection.workers.delete(key);
+ }
+}
+
+function workerKey(runtimeId: string | null | undefined, workerId: string): string {
+ if (!runtimeId) throw new Error('Workspace Worker projection is missing runtime_id');
+ return `${runtimeId}:${workerId}`;
+}
diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts
new file mode 100644
index 00000000..8b6d91bf
--- /dev/null
+++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts
@@ -0,0 +1,89 @@
+import type { SubscriptionFrame, SubscriptionWorker } from '$lib/generated/protocol';
+
+function assertEquals(actual: unknown, expected: unknown): void {
+ if (JSON.stringify(actual) !== JSON.stringify(expected)) {
+ throw new Error(`expected ${JSON.stringify(expected)}, got ${JSON.stringify(actual)}`);
+ }
+}
+import {
+ applyWorkspaceWorkersFrame,
+ createWorkspaceWorkersProjection,
+} from './worker-subscription-model';
+
+declare const Deno: {
+ test(name: string, fn: () => void | Promise): void;
+};
+
+function worker(runtimeId: string, workerId: string, revision: number): SubscriptionWorker {
+ return {
+ worker_id: workerId,
+ runtime_id: runtimeId,
+ subject_revision: revision,
+ state: 'idle',
+ workspace_id: 'workspace-test',
+ display_name: null,
+ profile: null,
+ working_directory_id: null,
+ };
+}
+
+Deno.test('workspace Worker snapshot keeps equal local ids from different Runtimes', () => {
+ const projection = createWorkspaceWorkersProjection();
+ const frame: SubscriptionFrame = {
+ protocol_version: 1,
+ frame: 'response',
+ message: {
+ result: 'subscribed',
+ payload: {
+ request_id: 'request-1',
+ subscription_id: 'subscription-1',
+ selector: { topic: 'workspace_workers' },
+ snapshot_revision: 1,
+ snapshot: {
+ topic: 'workers',
+ data: { workers: [worker('runtime-a', '1', 1), worker('runtime-b', '1', 1)] },
+ },
+ },
+ },
+ };
+ applyWorkspaceWorkersFrame(projection, frame);
+ assertEquals([...projection.workers.keys()].sort(), ['runtime-a:1', 'runtime-b:1']);
+});
+
+Deno.test('workspace Worker reducer ignores stale events and removes composite subject', () => {
+ const projection = createWorkspaceWorkersProjection();
+ projection.workers.set('runtime-a:1', worker('runtime-a', '1', 3));
+ projection.revisions.set('runtime-a:1', 3);
+
+ applyWorkspaceWorkersFrame(projection, {
+ protocol_version: 1,
+ frame: 'event',
+ message: {
+ event: 'event',
+ data: {
+ subscription_id: 'subscription-1',
+ subject_revision: 2,
+ payload: { event: 'worker_upserted', data: { worker: worker('runtime-a', '1', 2) } },
+ },
+ },
+ });
+ assertEquals(projection.revisions.get('runtime-a:1'), 3);
+
+ applyWorkspaceWorkersFrame(projection, {
+ protocol_version: 1,
+ frame: 'event',
+ message: {
+ event: 'event',
+ data: {
+ subscription_id: 'subscription-1',
+ subject_revision: 4,
+ payload: {
+ event: 'worker_removed',
+ data: { worker_id: '1', runtime_id: 'runtime-a' },
+ },
+ },
+ },
+ });
+ assertEquals(projection.workers.size, 0);
+ assertEquals(projection.revisions.get('runtime-a:1'), 4);
+});
diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts
new file mode 100644
index 00000000..50c39a8d
--- /dev/null
+++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts
@@ -0,0 +1,130 @@
+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 {
+ applyWorkspaceWorkersFrame,
+ createWorkspaceWorkersProjection,
+} from './worker-subscription-model';
+import type { Worker } from './types';
+
+export type WorkspaceWorkersState = {
+ loading: boolean;
+ error: string | null;
+ workers: Worker[];
+};
+
+const stores = new Map>();
+
+export function workspaceWorkersStore(workspaceId: string): Readable {
+ const cached = stores.get(workspaceId);
+ if (cached) return cached;
+ const store = readable(
+ { loading: true, error: null, workers: [] },
+ (set) => {
+ if (!browser || !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)
+ .sort((left, right) =>
+ left.runtime_id.localeCompare(right.runtime_id) ||
+ left.worker_id.localeCompare(right.worker_id)
+ );
+ 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;
+ }
+ 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();
+ };
+ },
+ );
+ stores.set(workspaceId, store);
+ return store;
+}
+
+function projectWorker(worker: SubscriptionWorker): Worker {
+ if (!worker.runtime_id) throw new Error('Workspace Worker projection is missing runtime_id');
+ const displayName = worker.display_name ?? `Worker ${worker.worker_id}`;
+ return {
+ runtime_id: worker.runtime_id,
+ worker_id: worker.worker_id,
+ host_id: worker.runtime_id,
+ display_name: displayName,
+ label: displayName,
+ profile: worker.profile ?? null,
+ tags: [],
+ workspace: { visibility: 'workspace', identity: 'runtime_subscription_worker' },
+ state: worker.state,
+ pinned: false,
+ retention_state: 'transient',
+ implementation: {
+ kind: 'runtime_subscription_worker',
+ display_hint: 'Workspace-authorized Runtime Worker',
+ },
+ capabilities: {
+ can_stop: worker.state !== 'stopped' && worker.state !== 'cancelled',
+ can_spawn_followup: false,
+ },
+ working_directory: null,
+ diagnostics: [],
+ };
+}