Remove duplicate inputs with CRDT.

lynxkite-app/src/lynxkite_app/crdt.py

@@ -197,7 +197,6 @@ async def workspace_changed(name: str, changes: pycrdt.MapEvent, ws_crdt: pycrdt
         getattr(change, "keys", {}).get("__execution_delay", {}).get("newValue", 0)
         for change in changes
     )
-    print(f"Running {name} in {ws_pyd.env}...")
     if delay:
         task = asyncio.create_task(execute(name, ws_crdt, ws_pyd, delay))
         delayed_executions[name] = task
@@ -221,10 +220,12 @@ async def execute(
             await asyncio.sleep(delay)
         except asyncio.CancelledError:
             return
+    print(f"Running {name} in {ws_pyd.env}...")
     path = config.DATA_PATH / name
     assert path.is_relative_to(config.DATA_PATH), "Provided workspace path is invalid"
     # Save user changes before executing, in case the execution fails.
     workspace.save(ws_pyd, path)
+    ws_pyd._crdt = ws_crdt
     with ws_crdt.doc.transaction():
         for nc, np in zip(ws_crdt["nodes"], ws_pyd.nodes):
             if "data" not in nc:
@@ -234,6 +235,7 @@ async def execute(
             np._crdt = nc
     await workspace.execute(ws_pyd)
     workspace.save(ws_pyd, path)
+    print(f"Finished running {name} in {ws_pyd.env}.")
 
 
 @contextlib.asynccontextmanager

lynxkite-core/src/lynxkite/core/workspace.py

@@ -92,6 +92,7 @@ class Workspace(BaseConfig):
     env: str = ""
     nodes: list[WorkspaceNode] = dataclasses.field(default_factory=list)
     edges: list[WorkspaceEdge] = dataclasses.field(default_factory=list)
+    _crdt: pycrdt.Map
 
 
 async def execute(ws: Workspace):

lynxkite-graph-analytics/src/lynxkite_graph_analytics/core.py

@@ -135,12 +135,15 @@ def nx_node_attribute_func(name):
     return decorator
 
 
-def disambiguate_edges(ws):
+def disambiguate_edges(ws: workspace.Workspace):
     """If an input plug is connected to multiple edges, keep only the last edge."""
     seen = set()
     for edge in reversed(ws.edges):
         if (edge.target, edge.targetHandle) in seen:
-            ws.edges.remove(edge)
+            i = ws.edges.index(edge)
+            del ws.edges[i]
+            if hasattr(ws, "_crdt"):
+                del ws._crdt["edges"][i]
         seen.add((edge.target, edge.targetHandle))