Merge pull request #32 from biggraph/darabos-crdt-updated

requirements.txt

@@ -6,6 +6,8 @@ orjson
 pandas
 scipy
 uvicorn[standard]
+pycrdt
+pycrdt-websocket
 # For llm_ops
 chromadb
 Jinja2

server/crdt.py

@@ -0,0 +1,131 @@
+'''CRDT is used to synchronize workspace state for backend and frontend(s).'''
+import asyncio
+import contextlib
+import fastapi
+import os.path
+import pycrdt
+import pycrdt_websocket
+
+import pycrdt_websocket.ystore
+
+router = fastapi.APIRouter()
+
+def ws_exception_handler(exception, log):
+    print('exception', exception)
+    log.exception(exception)
+    return True
+
+class WebsocketServer(pycrdt_websocket.WebsocketServer):
+    async def init_room(self, name):
+        ystore = pycrdt_websocket.ystore.FileYStore(f'crdt_data/{name}.crdt')
+        ydoc = pycrdt.Doc()
+        ydoc['workspace'] = ws = pycrdt.Map()
+        # Replay updates from the store.
+        try:
+            for update, timestamp in [(item[0], item[-1]) async for item in ystore.read()]:
+                ydoc.apply_update(update)
+        except pycrdt_websocket.ystore.YDocNotFound:
+            pass
+        if 'nodes' not in ws:
+            ws['nodes'] = pycrdt.Array()
+        if 'edges' not in ws:
+            ws['edges'] = pycrdt.Array()
+        if 'env' not in ws:
+            ws['env'] = 'unset'
+            try_to_load_workspace(ws, name)
+        room = pycrdt_websocket.YRoom(ystore=ystore, ydoc=ydoc)
+        room.ws = ws
+        def on_change(changes):
+            asyncio.create_task(workspace_changed(changes, ws))
+        ws.observe_deep(on_change)
+        return room
+
+    async def get_room(self, name: str) -> pycrdt_websocket.YRoom:
+        if name not in self.rooms:
+            self.rooms[name] = await self.init_room(name)
+        room = self.rooms[name]
+        await self.start_room(room)
+        return room
+
+websocket_server = WebsocketServer(exception_handler=ws_exception_handler, auto_clean_rooms=False)
+asgi_server = pycrdt_websocket.ASGIServer(websocket_server)
+
+last_ws_input = None
+def clean_input(ws_pyd):
+    for node in ws_pyd.nodes:
+        node.data.display = None
+        node.position.x = 0
+        node.position.y = 0
+        if node.model_extra:
+            for key in list(node.model_extra.keys()):
+                delattr(node, key)
+
+def crdt_update(crdt_obj, python_obj, boxes=set()):
+    if isinstance(python_obj, dict):
+        for key, value in python_obj.items():
+            if key in boxes:
+                crdt_obj[key] = value
+            elif isinstance(value, dict):
+                if crdt_obj.get(key) is None:
+                    crdt_obj[key] = pycrdt.Map()
+                crdt_update(crdt_obj[key], value, boxes)
+            elif isinstance(value, list):
+                if crdt_obj.get(key) is None:
+                    crdt_obj[key] = pycrdt.Array()
+                crdt_update(crdt_obj[key], value, boxes)
+            else:
+                crdt_obj[key] = value
+    elif isinstance(python_obj, list):
+        for i, value in enumerate(python_obj):
+            if isinstance(value, dict):
+                if i >= len(crdt_obj):
+                    crdt_obj.append(pycrdt.Map())
+                crdt_update(crdt_obj[i], value, boxes)
+            elif isinstance(value, list):
+                if i >= len(crdt_obj):
+                    crdt_obj.append(pycrdt.Array())
+                crdt_update(crdt_obj[i], value, boxes)
+            else:
+                if i >= len(crdt_obj):
+                    crdt_obj.append(value)
+                else:
+                    crdt_obj[i] = value
+    else:
+        raise ValueError('Invalid type:', python_obj)
+
+
+def try_to_load_workspace(ws, name):
+    from . import workspace
+    json_path = f'data/{name}'
+    if os.path.exists(json_path):
+        ws_pyd = workspace.load(json_path)
+        crdt_update(ws, ws_pyd.model_dump(), boxes={'display'})
+
+async def workspace_changed(e, ws_crdt):
+    global last_ws_input
+    from . import workspace
+    ws_pyd = workspace.Workspace.model_validate(ws_crdt.to_py())
+    clean_input(ws_pyd)
+    if ws_pyd == last_ws_input:
+        return
+    last_ws_input = ws_pyd.model_copy(deep=True)
+    await workspace.execute(ws_pyd)
+    for nc, np in zip(ws_crdt['nodes'], ws_pyd.nodes):
+        if 'data' not in nc:
+            nc['data'] = pycrdt.Map()
+        # Display is added as an opaque Box.
+        nc['data']['display'] = np.data.display
+        nc['data']['error'] = np.data.error
+
+@contextlib.asynccontextmanager
+async def lifespan(app):
+    async with websocket_server:
+        yield
+
+def sanitize_path(path):
+    return os.path.relpath(os.path.normpath(os.path.join("/", path)), "/")
+
+@router.websocket("/ws/crdt/{room_name}")
+async def crdt_websocket(websocket: fastapi.WebSocket, room_name: str):
+    room_name = sanitize_path(room_name)
+    await asgi_server({'path': room_name}, websocket._receive, websocket._send)

server/executors/one_by_one.py

@@ -74,7 +74,12 @@ def make_cache_key(obj):
 
 EXECUTOR_OUTPUT_CACHE = {}
 
-def execute(ws, catalog, cache=None):
+async def await_if_needed(obj):
+  if inspect.isawaitable(obj):
+    return await obj
+  return obj
+
+async def execute(ws, catalog, cache=None):
   nodes = {n.id: n for n in ws.nodes}
   contexts = {n.id: Context(node=n) for n in ws.nodes}
   edges = {n.id: [] for n in ws.nodes}
@@ -113,10 +118,10 @@ def execute(ws, catalog, cache=None):
           if cache is not None:
             key = make_cache_key((inputs, params))
             if key not in cache:
-              cache[key] = op(*inputs, **params)
+              cache[key] = await await_if_needed(op(*inputs, **params))
             result = cache[key]
           else:
-            result = op(*inputs, **params)
+            result = await await_if_needed(op(*inputs, **params))
         except Exception as e:
           traceback.print_exc()
           data.error = str(e)

server/lynxkite_ops.py

@@ -83,7 +83,7 @@ def disambiguate_edges(ws):
 
 
 @ops.register_executor('LynxKite')
-def execute(ws):
+async def execute(ws):
     catalog = ops.CATALOGS['LynxKite']
     # Nodes are responsible for interpreting/executing their child nodes.
     nodes = [n for n in ws.nodes if not n.parentId]
@@ -105,11 +105,6 @@ def execute(ws):
                 data = node.data
                 op = catalog[data.title]
                 params = {**data.params}
-                if op.sub_nodes:
-                    sub_nodes = children.get(node.id, [])
-                    sub_node_ids = [node.id for node in sub_nodes]
-                    sub_edges = [edge for edge in ws.edges if edge.source in sub_node_ids]
-                    params['sub_flow'] = {'nodes': sub_nodes, 'edges': sub_edges}
                 # Convert inputs.
                 for i, (x, p) in enumerate(zip(inputs, op.inputs.values())):
                   if p.type == nx.Graph and isinstance(x, Bundle):

server/main.py

@@ -3,6 +3,7 @@ import fastapi
 import importlib
 import pathlib
 import pkgutil
+from . import crdt
 from . import ops
 from . import workspace
 
@@ -14,8 +15,8 @@ for _, name, _ in pkgutil.iter_modules([str(here)]):
         name = f"server.{name}"
         lynxkite_modules[name] = importlib.import_module(name)
 
-app = fastapi.FastAPI()
-
+app = fastapi.FastAPI(lifespan=crdt.lifespan)
+app.include_router(crdt.router)
 
 
 @app.get("/api/catalog")
@@ -38,9 +39,9 @@ def save(req: SaveRequest):
 
 
 @app.post("/api/save")
-def save_and_execute(req: SaveRequest):
+async def save_and_execute(req: SaveRequest):
     save(req)
-    workspace.execute(req.ws)
+    await workspace.execute(req.ws)
     save(req)
     return req.ws
 

server/workspace.py

@@ -43,9 +43,9 @@ class Workspace(BaseConfig):
     edges: list[WorkspaceEdge] = dataclasses.field(default_factory=list)
 
 
-def execute(ws: Workspace):
+async def execute(ws: Workspace):
     if ws.env in ops.EXECUTORS:
-        ops.EXECUTORS[ws.env](ws)
+        await ops.EXECUTORS[ws.env](ws)
 
 
 def save(ws: Workspace, path: str):

web/package.json

@@ -10,27 +10,28 @@
     "check": "svelte-check --tsconfig ./tsconfig.json"
   },
   "devDependencies": {
-    "@sveltejs/vite-plugin-svelte": "^3.0.2",
-    "@tsconfig/svelte": "^5.0.4",
-    "sass": "^1.77.2",
-    "svelte": "^4.2.12",
-    "svelte-check": "^3.6.9",
+    "@sveltejs/vite-plugin-svelte": "3.1.2",
+    "@syncedstore/core": "0.6.0",
+    "@syncedstore/svelte": "0.6.0",
+    "@tsconfig/svelte": "5.0.4",
+    "sass": "1.79.5",
+    "svelte": "4.2.19",
+    "svelte-check": "4.0.5",
     "svelte-markdown": "^0.4.1",
-    "tslib": "^2.6.2",
-    "typescript": "^5.4.4",
-    "unplugin-icons": "^0.18.5",
-    "vite": "^5.2.8"
+    "tslib": "2.7.0",
+    "typescript": "5.6.3",
+    "unplugin-icons": "0.19.3",
+    "vite": "5.4.9",
+    "y-websocket": "2.0.4"
   },
   "dependencies": {
-    "@iconify-json/tabler": "^1.1.110",
-    "@popperjs/core": "^2.11.8",
-    "@sveltestack/svelte-query": "^1.6.0",
-    "@xyflow/svelte": "^0.1.3",
-    "bootstrap": "^5.3.3",
-    "deep-object-diff": "^1.1.9",
-    "echarts": "^5.5.0",
-    "fuse.js": "^7.0.0",
-    "svelte-echarts": "^1.0.0-rc1",
-    "tabulator-tables": "^6.2.1"
+    "@iconify-json/tabler": "1.2.5",
+    "@popperjs/core": "2.11.8",
+    "@sveltestack/svelte-query": "1.6.0",
+    "@xyflow/svelte": "0.1.21",
+    "bootstrap": "5.3.3",
+    "echarts": "5.5.1",
+    "fuse.js": "7.0.0",
+    "svelte-echarts": "^1.0.0-rc3"
   }
 }

web/src/LynxKiteFlow.svelte

@@ -1,12 +1,13 @@
 <script lang="ts">
-  import { diff } from 'deep-object-diff';
-  import { writable, derived } from 'svelte/store';
+  import { setContext } from 'svelte';
+  import { writable } from 'svelte/store';
   import {
     SvelteFlow,
     Controls,
     MiniMap,
     MarkerType,
     useSvelteFlow,
+    useUpdateNodeInternals,
     type XYPosition,
     type Node,
     type Edge,
@@ -16,7 +17,7 @@
   import ArrowBack from 'virtual:icons/tabler/arrow-back'
   import Backspace from 'virtual:icons/tabler/backspace'
   import Atom from 'virtual:icons/tabler/Atom'
-  import { useQuery, useMutation, useQueryClient } from '@sveltestack/svelte-query';
+  import { useQuery } from '@sveltestack/svelte-query';
   import NodeWithParams from './NodeWithParams.svelte';
   import NodeWithVisualization from './NodeWithVisualization.svelte';
   import NodeWithImage from './NodeWithImage.svelte';
@@ -26,27 +27,30 @@
   import NodeSearch from './NodeSearch.svelte';
   import EnvironmentSelector from './EnvironmentSelector.svelte';
   import '@xyflow/svelte/dist/style.css';
+  import { syncedStore, getYjsDoc } from "@syncedstore/core";
+  import { svelteSyncedStore } from "@syncedstore/svelte";
+  import { WebsocketProvider } from "y-websocket";
+  const updateNodeInternals = useUpdateNodeInternals();
+
+  function getCRDTStore(path) {
+    const sstore = syncedStore({ workspace: {} });
+    const doc = getYjsDoc(sstore);
+    const wsProvider = new WebsocketProvider("ws://localhost:8000/ws/crdt", path, doc);
+    return {store: svelteSyncedStore(sstore), sstore, doc};
+  }
+  $: connection = getCRDTStore(path);
+  $: store = connection.store;
+  $: store.subscribe((value) => {
+    if (!value?.workspace?.edges) return;
+    $nodes = [...value.workspace.nodes];
+    $edges = [...value.workspace.edges];
+    updateNodeInternals();
+  });
+  $: setContext('LynxKite store', store);
 
   export let path = '';
 
   const { screenToFlowPosition } = useSvelteFlow();
-  const queryClient = useQueryClient();
-  const backendWorkspace = useQuery(['workspace', path], async () => {
-    const res = await fetch(`/api/load?path=${path}`);
-    return res.json();
-  }, {staleTime: 10000, retry: false});
-  const mutation = useMutation(async(update) => {
-    const res = await fetch('/api/save', {
-      method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-      },
-      body: JSON.stringify(update),
-    });
-    return await res.json();
-  }, {
-    onSuccess: data => queryClient.setQueryData(['workspace', path], data),
-  });
 
   const nodeTypes: NodeTypes = {
     basic: NodeWithParams,
@@ -59,13 +63,6 @@
 
   const nodes = writable<Node[]>([]);
   const edges = writable<Edge[]>([]);
-  let doNotSave = true;
-  $: if ($backendWorkspace.isSuccess) {
-    doNotSave = true; // Change is coming from the backend.
-    nodes.set(JSON.parse(JSON.stringify($backendWorkspace.data?.nodes || [])));
-    edges.set(JSON.parse(JSON.stringify($backendWorkspace.data?.edges || [])));
-    doNotSave = false;
-  }
 
   function closeNodeSearch() {
     nodeSearchSettings = undefined;
@@ -78,37 +75,36 @@
     event.preventDefault();
     nodeSearchSettings = {
       pos: { x: event.clientX, y: event.clientY },
-      boxes: $catalog.data[$backendWorkspace.data?.env],
+      boxes: $catalog.data[$store.workspace.env],
     };
   }
   function addNode(e) {
     const meta = {...e.detail};
-    nodes.update((n) => {
-      const node = {
-        type: meta.type,
-        data: {
-          meta: meta,
-          title: meta.name,
-          params: Object.fromEntries(
-            Object.values(meta.params).map((p) => [p.name, p.default])),
-        },
-      };
-      node.position = screenToFlowPosition({x: nodeSearchSettings.pos.x, y: nodeSearchSettings.pos.y});
-      const title = node.data.title;
-      let i = 1;
+    const node = {
+      type: meta.type,
+      data: {
+        meta: meta,
+        title: meta.name,
+        params: Object.fromEntries(
+          Object.values(meta.params).map((p) => [p.name, p.default])),
+      },
+    };
+    node.position = screenToFlowPosition({x: nodeSearchSettings.pos.x, y: nodeSearchSettings.pos.y});
+    const title = node.data.title;
+    let i = 1;
+    node.id = `${title} ${i}`;
+    const nodes = $store.workspace.nodes;
+    while (nodes.find((x) => x.id === node.id)) {
+      i += 1;
       node.id = `${title} ${i}`;
-      while (n.find((x) => x.id === node.id)) {
-        i += 1;
-        node.id = `${title} ${i}`;
-      }
-      node.parentId = nodeSearchSettings.parentId;
-      if (node.parentId) {
-        node.extent = 'parent';
-        const parent = n.find((x) => x.id === node.parentId);
-        node.position = { x: node.position.x - parent.position.x, y: node.position.y - parent.position.y };
-      }
-      return [...n, node]
-    });
+    }
+    node.parentId = nodeSearchSettings.parentId;
+    if (node.parentId) {
+      node.extent = 'parent';
+      const parent = nodes.find((x) => x.id === node.parentId);
+      node.position = { x: node.position.x - parent.position.x, y: node.position.y - parent.position.y };
+    }
+    nodes.push(node);
     closeNodeSearch();
   }
   const catalog = useQuery(['catalog'], async () => {
@@ -122,46 +118,6 @@
     parentId: string,
   };
 
-  const graph = derived([nodes, edges], ([nodes, edges]) => ({ nodes, edges }));
-  // Like JSON.stringify, but with keys sorted.
-  function orderedJSON(obj: any) {
-    const allKeys = new Set();
-    JSON.stringify(obj, (key, value) => (allKeys.add(key), value));
-    return JSON.stringify(obj, Array.from(allKeys).sort());
-  }
-  graph.subscribe(async (g) => {
-    if (doNotSave) return;
-    const dragging = g.nodes.find((n) => n.dragging);
-    if (dragging) return;
-    const resizing = g.nodes.find((n) => n.data?.beingResized);
-    if (resizing) return;
-    scheduleSave(g);
-  });
-  let saveTimeout;
-  function scheduleSave(g) {
-    // A slight delay, so we don't send a million requests when a node is resized, for example.
-    clearTimeout(saveTimeout);
-    saveTimeout = setTimeout(() => save(g), 500);
-  }
-  function save(g) {
-    g = JSON.parse(JSON.stringify(g));
-    for (const node of g.nodes) {
-      delete node.measured;
-      delete node.selected;
-      delete node.dragging;
-      delete node.beingResized;
-    }
-    for (const node of g.edges) {
-      delete node.markerEnd;
-      delete node.selected;
-    }
-    g.env = $backendWorkspace.data?.env;
-    const ws = orderedJSON(g);
-    const bd = orderedJSON($backendWorkspace.data);
-    if (ws === bd) return;
-    console.log('changed', JSON.stringify(diff(g, $backendWorkspace.data), null, 2));
-    $mutation.mutate({ path, ws: g });
-  }
   function nodeClick(e) {
     const node = e.detail.node;
     const meta = node.data.meta;
@@ -176,11 +132,32 @@
       parentId: node.id,
     };
   }
+  function onConnect(params: Connection) {
+    const edge = {
+      id: `${params.source} ${params.target}`,
+      source: params.source,
+      sourceHandle: params.sourceHandle,
+      target: params.target,
+      targetHandle: params.targetHandle,
+    };
+    $store.workspace.edges.push(edge);
+  }
+  function onDelete(params) {
+    const { nodes, edges } = params;
+    for (const node of nodes) {
+      const index = $store.workspace.nodes.findIndex((x) => x.id === node.id);
+      if (index !== -1) $store.workspace.nodes.splice(index, 1);
+    }
+    for (const edge of edges) {
+      const index = $store.workspace.edges.findIndex((x) => x.id === edge.id);
+      if (index !== -1) $store.workspace.edges.splice(index, 1);
+    }
+  }
   $: parentDir = path.split('/').slice(0, -1).join('/');
-
 </script>
 
 <div class="page">
+  {#if $store.workspace !== undefined}
   <div class="top-bar">
     <div class="ws-name">
       <a href><img src="/favicon.ico"></a>
@@ -189,8 +166,10 @@
     <div class="tools">
       <EnvironmentSelector
         options={Object.keys($catalog.data || {})}
-        value={$backendWorkspace.data?.env}
-        onChange={(env) => $mutation.mutate({ path, ws: { ...$backendWorkspace.data, env } })}
+        value={$store.workspace.env}
+        onChange={(env) => {
+          $store.workspace.env = env;
+        }}
         />
       <a href><Atom /></a>
       <a href><Backspace /></a>
@@ -201,6 +180,8 @@
     <SvelteFlow {nodes} {edges} {nodeTypes} fitView
       on:paneclick={toggleNodeSearch}
       on:nodeclick={nodeClick}
+      onconnect={onConnect}
+      ondelete={onDelete}
       proOptions={{ hideAttribution: true }}
       maxZoom={3}
       minZoom={0.3}
@@ -213,6 +194,7 @@
       {/if}
     </SvelteFlow>
   </div>
+  {/if}
 </div>
 
 <style>

web/src/LynxKiteNode.svelte

@@ -1,4 +1,5 @@
 <script lang="ts">
+  import { getContext } from 'svelte';
   import { Handle, useSvelteFlow, useUpdateNodeInternals, type NodeProps, NodeResizeControl } from '@xyflow/svelte';
   import ChevronDownRight from 'virtual:icons/tabler/chevron-down-right';
 
@@ -10,6 +11,10 @@
   export let containerStyle = '';
   export let id: $$Props['id']; id;
   export let data: $$Props['data'];
+  export let deletable: $$Props['deletable'] = undefined; deletable;
+  export let draggable: $$Props['draggable'] = undefined; draggable;
+  export let parentId: $$Props['parentId'] = undefined; parentId;
+  export let selectable: $$Props['selectable'] = undefined; selectable;
   export let dragHandle: $$Props['dragHandle'] = undefined; dragHandle;
   export let type: $$Props['type']  = undefined; type;
   export let selected: $$Props['selected'] = undefined; selected;
@@ -24,11 +29,14 @@
   export let positionAbsoluteY: $$Props['positionAbsoluteY'] = undefined; positionAbsoluteY;
   export let onToggle = () => {};
 
+  $: store = getContext('LynxKite store');
   $: expanded = !data.collapsed;
   function titleClicked() {
-    updateNodeData(id, { collapsed: expanded });
-    data = data;
+    const i = $store.workspace.nodes.findIndex((n) => n.id === id);
+    $store.workspace.nodes[i].data.collapsed = expanded;
     onToggle({ expanded });
+    // Trigger update.
+    data = data;
     updateNodeInternals();
   }
   function asPx(n: number | undefined) {

web/src/NodeWithParams.svelte

@@ -1,21 +1,29 @@
 <script lang="ts">
-  import { type NodeProps, useSvelteFlow } from '@xyflow/svelte';
+  import { getContext } from 'svelte';
+  import { type NodeProps, useNodes } from '@xyflow/svelte';
   import LynxKiteNode from './LynxKiteNode.svelte';
   import NodeParameter from './NodeParameter.svelte';
   type $$Props = NodeProps;
   export let id: $$Props['id'];
   export let data: $$Props['data'];
-  const { updateNodeData } = useSvelteFlow();
   $: metaParams = data.meta?.params;
+  $: store = getContext('LynxKite store');
+  function setParam(name, newValue) {
+    const i = $store.workspace.nodes.findIndex((n) => n.id === id);
+    $store.workspace.nodes[i].data.params[name] = newValue;
+  }
+  $: params = $nodes && data?.params ? Object.entries(data.params) : [];
+  const nodes = useNodes(); // We don't properly get updates to "data". This is a hack.
+  $: props = $nodes && $$props;
 </script>
 
-<LynxKiteNode {...$$props}>
-  {#each Object.entries(data.params) as [name, value]}
+<LynxKiteNode {...props}>
+  {#each params as [name, value]}
     <NodeParameter
       {name}
       {value}
       meta={metaParams?.[name]}
-      onChange={(newValue) => updateNodeData(id, { params: { ...data.params, [name]: newValue } })}
+      onChange={(value) => setParam(name, value)}
       />
   {/each}
   <slot />

web/src/NodeWithTableView.svelte

@@ -1,12 +1,14 @@
 <script lang="ts">
-  import { type NodeProps } from '@xyflow/svelte';
+  import { useNodes, type NodeProps } from '@xyflow/svelte';
   import LynxKiteNode from './LynxKiteNode.svelte';
   import Table from './Table.svelte';
   import SvelteMarkdown from 'svelte-markdown'
   type $$Props = NodeProps;
   export let data: $$Props['data'];
+  const nodes = useNodes(); // We don't properly get updates to "data". This is a hack.
+  $: D = $nodes && data;
   const open = {};
-  $: single = data.display?.dataframes && Object.keys(data.display.dataframes).length === 1;
+  $: single = D.display?.value?.dataframes && Object.keys(D.display.value.dataframes).length === 1;
   function toMD(v) {
     if (typeof v === 'string') {
       return v;
@@ -19,8 +21,8 @@
 </script>
 
 <LynxKiteNode {...$$props}>
-  {#if data.display}
-    {#each Object.entries(data.display.dataframes || {}) as [name, df]}
+  {#if D?.display?.value}
+    {#each Object.entries(D.display.value.dataframes || {}) as [name, df]}
       {#if !single}<div class="df-head" on:click={() => open[name] = !open[name]}>{name}</div>{/if}
       {#if single || open[name]}
         {#if df.data.length > 1}
@@ -35,7 +37,7 @@
         {/if}
       {/if}
     {/each}
-    {#each Object.entries(data.display.others || {}) as [name, o]}
+    {#each Object.entries(D.display.value.others || {}) as [name, o]}
       <div class="df-head" on:click={() => open[name] = !open[name]}>{name}</div>
       {#if open[name]}
       <pre>{o}</pre>

web/src/NodeWithVisualization.svelte

@@ -1,16 +1,17 @@
 <script lang="ts">
-  import { type NodeProps } from '@xyflow/svelte';
+  import { useNodes, type NodeProps } from '@xyflow/svelte';
   import NodeWithParams from './NodeWithParams.svelte';
   import { Chart } from 'svelte-echarts';
   import { init } from 'echarts';
   type $$Props = NodeProps;
   export let data: $$Props['data'];
+
+  const nodes = useNodes(); // We don't properly get updates to "data". This is a hack.
+  $: D = $nodes && data;
 </script>
 
 <NodeWithParams {...$$props}>
-  {#if data.display}
-    <Chart {init} options={data.display} initOptions={{renderer: 'canvas', width: 250, height: 250}}/>
-  {/if}
+  <Chart {init} options={D?.display?.value || {}} initOptions={{renderer: 'canvas', width: 250, height: 250}}/>
 </NodeWithParams>
 <style>
 </style>

web/src/Table.svelte

@@ -1,38 +1,22 @@
 <script lang="ts">
-  import {TabulatorFull as Tabulator} from 'tabulator-tables';
-  import {onMount} from 'svelte';
-
   export let data, columns;
-
-  let tableComponent;
-  let tab;
-
-  // The rows in the data are arrays, but Tabulator expects objects.
-  const objs = [];
-  $: {
-    objs.splice();
-    for (const row of data) {
-      const obj = {};
-      for (let i = 0; i < columns.length; i++) {
-        let d = row[i];
-        if (typeof d !== 'string' && typeof d !== 'number') {
-          d = JSON.stringify(d);
-        }
-        obj[columns[i]] = d;
-      }
-      objs.push(obj);
-    }
-  }
-
-  onMount(() => {
-    tab = new Tabulator(tableComponent, {
-      data: objs,
-      columns: columns.map(c => ({title: c, field: c, widthGrow: 1})),
-      height: '311px',
-      reactiveData: true,
-      layout: "fitColumns",
-    });
-  });
 </script>
 
-<div bind:this={tableComponent}></div>
+<table>
+  <thead>
+    <tr>
+      {#each columns as column}
+        <th>{column}</th>
+      {/each}
+    </tr>
+  </thead>
+  <tbody>
+    {#each data as row}
+      <tr>
+        {#each columns as column}
+          <td>{row[column]}</td>
+        {/each}
+      </tr>
+    {/each}
+  </tbody>
+</table>

web/src/app.scss

@@ -1,7 +1,6 @@
 // Import all of Bootstrap's CSS
 $form-select-indicator-color: oklch(90% 0.01 55);
 @import "bootstrap/scss/bootstrap";
-@import "tabulator-tables";
 :root {
   --bs-border-color: oklch(90% 0.01 55);
 }

web/vite.config.ts

@@ -9,6 +9,13 @@ export default defineConfig({
     Icons({ compiler: 'svelte', defaultStyle: 'vertical-align: sub;' }),
   ],
   server: {
-    proxy: { '/api': 'http://127.0.0.1:8000' },
+    proxy: {
+      '/api': 'http://127.0.0.1:8000',
+      '/ws': {
+        target: 'ws://127.0.0.1:8000',
+        ws: true,
+        changeOrigin: true,
+      },
+    },
   },
 })