Load old files, few fixes for LynxScribe demo.

server/crdt.py

@@ -32,6 +32,7 @@ class WebsocketServer(pycrdt_websocket.WebsocketServer):
             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):
@@ -59,39 +60,47 @@ def clean_input(ws_pyd):
             for key in list(node.model_extra.keys()):
                 delattr(node, key)
 
-def crdt_update(crdt_obj, python_obj):
+def crdt_update(crdt_obj, python_obj, boxes=set()):
     if isinstance(python_obj, dict):
         for key, value in python_obj.items():
-            if isinstance(value, dict):
+            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)
+                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)
+                crdt_update(crdt_obj[key], value, boxes)
             else:
-                print('set', key, value)
                 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)
+                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)
+                crdt_update(crdt_obj[i], value, boxes)
             else:
                 if i >= len(crdt_obj):
                     crdt_obj.append(value)
                 else:
-                    print('set', i, value)
                     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
@@ -99,14 +108,14 @@ async def workspace_changed(e, ws_crdt):
     clean_input(ws_pyd)
     if ws_pyd == last_ws_input:
         return
-    print('ws changed')
     last_ws_input = ws_pyd.model_copy(deep=True)
-    workspace.execute(ws_pyd)
+    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):

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]

server/lynxscribe_ops.py

@@ -121,10 +121,10 @@ def mask(*, name='', regex='', exceptions='', mask_pattern=''):
 
 @ops.input_position(chat_api="bottom")
 @op("Test Chat API")
-def test_chat_api(message, chat_api, *, show_details=False):
+async def test_chat_api(message, chat_api, *, show_details=False):
   chat_api = chat_api[0]['chat_api']
   request = ChatAPIRequest(session_id="b43215a0-428f-11ef-9454-0242ac120002", question=message['text'], history=[])
-  response = asyncio.run(chat_api.answer(request))
+  response = await chat_api.answer(request)
   if show_details:
     return {**response.__dict__}
   else:

server/main.py

@@ -35,9 +35,9 @@ def save(req: SaveRequest):
     workspace.save(req.ws, path)
 
 @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/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>