web: subscribe sidebar to workspace workers

web/workspace/src/lib/workspace/sidebar/WorkersNavSection.svelte

@@ -1,8 +1,8 @@
 <script lang="ts">
-  import { workspaceApiPath } from '$lib/workspace/api/http';
   import { workerConsoleHref } from '$lib/workspace/console/model';
+  import { workspaceWorkersStore } from './worker-subscription';
   import { canShowWorkerInSidebar } from './workers';
-  import type { ListResponse, Worker } from './types';
+  import type { Worker } from './types';
 
   const MAX_VISIBLE_WORKERS = 6;
 
@@ -12,61 +12,18 @@
   };
 
   let { currentPath = '/', workspaceId }: Props = $props();
-
-  function workerApiPath(path: string): string {
-    return workspaceApiPath(workspaceId, path);
-  }
-
   let loading = $state(true);
   let error = $state<string | null>(null);
   let workers = $state<Worker[]>([]);
-  let placeholder = $state<string | null>(null);
 
   $effect(() => {
-    if (!workspaceId) {
+    const subscription = workspaceWorkersStore(workspaceId);
-      loading = false;
+    return subscription.subscribe((state) => {
-      workers = [];
+      loading = state.loading;
-      return;
+      error = state.error;
-    }
+      workers = state.workers.filter(canShowWorkerInSidebar).slice(0, MAX_VISIBLE_WORKERS);
-
+    });
-    const controller = new AbortController();
-    void loadWorkers(controller.signal);
-    return () => controller.abort();
   });
-
-  async function loadWorkers(signal?: AbortSignal) {
-    loading = true;
-    error = null;
-    placeholder = null;
-    try {
-      const response = await fetch(workerApiPath('/workers'), { signal });
-      if (response.status === 404) {
-        workers = [];
-        placeholder = 'Worker API is not integrated in this build yet.';
-        return;
-      }
-      if (!response.ok) {
-        throw new Error(`workers request failed (${response.status})`);
-      }
-      const payload = (await response.json()) as ListResponse<Worker>;
-      workers = Array.isArray(payload.items)
-        ? payload.items.filter(canShowWorkerInSidebar).slice(0, MAX_VISIBLE_WORKERS)
-        : [];
-      if (workers.length === 0) {
-        placeholder = 'No workers reported by the current API.';
-      }
-    } catch (err) {
-      if (err instanceof DOMException && err.name === 'AbortError') {
-        return;
-      }
-      error = err instanceof Error ? err.message : 'workers request failed';
-      workers = [];
-    } finally {
-      if (!signal?.aborted) {
-        loading = false;
-      }
-    }
-  }
 </script>
 
 <section class="nav-section" aria-labelledby="workers-heading">
@@ -94,11 +51,10 @@
 
   {#if loading}
     <p class="section-state">Checking workers…</p>
-  {:else if error}
-    <p class="section-state error">{error}</p>
   {:else if workers.length === 0}
-    <p class="section-state">{placeholder ?? 'Workers will appear here when an API is connected.'}</p>
+    <p class="section-state" class:error={Boolean(error)}>{error ?? 'No Workers are active.'}</p>
   {:else}
+    {#if error}<p class="section-state error">{error}</p>{/if}
     <ul class="nav-list" aria-label="Workers">
       {#each workers as worker (`${worker.runtime_id}:${worker.worker_id}`)}
         {@const href = workerConsoleHref(worker, workspaceId)}

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<string, SubscriptionWorker>;
+  revisions: Map<string, number>;
+};
+
+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}`;
+}

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>): 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);
+});

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<string, Readable<WorkspaceWorkersState>>();
+
+export function workspaceWorkersStore(workspaceId: string): Readable<WorkspaceWorkersState> {
+  const cached = stores.get(workspaceId);
+  if (cached) return cached;
+  const store = readable<WorkspaceWorkersState>(
+    { 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<typeof setTimeout> | 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: [],
+  };
+}