Support collaborative CLI document writes

.github/workflows/test.yml

@@ -38,10 +38,6 @@ jobs:
       matrix:
         os: [ubuntu-latest, macos-latest, windows-latest]
         python-version: [ '3.11', '3.12', '3.13', '3.14', '3.15' ]
-        exclude:
-          # pywinpty's Python 3.15 source build fails to load a DLL on Windows.
-          - os: windows-latest
-            python-version: '3.15'
 
     steps:
       - name: Checkout repository

docs/usage/CLI.md

@@ -52,3 +52,23 @@ Provide exactly one of `--source` or `--stdin`; `--source ''` clears the content
 When a code cell's source changes, its saved outputs and execution count are cleared.
 Other cells are preserved, and no cells are executed. Writing a whole notebook without
 `--cell` is not supported.
+
+For a server with real-time collaboration enabled, add `--collaborative` before
+the command to read and write the live shared document:
+
+```bash
+jpterm --server http://127.0.0.1:8000 --collaborative show notebook.ipynb --show-cell-ids
+jpterm --server http://127.0.0.1:8000 --collaborative write notebook.ipynb --cell CELL_ID --source 'print(42)'
+cat script.py | jpterm --server http://127.0.0.1:8000 --collaborative write script.py --stdin
+```
+
+Collaborative commands wait for the initial synchronization before reading or
+editing. Writes update the shared CRDT. Saving to disk follows
+the server's autosave policy.
+
+For Jupyverse, install the collaboration plugins in the server environment:
+
+```bash
+pip install 'jupyverse[jupyterlab,noauth,collaboration]'
+jupyverse --timeout 10
+```

plugins/cell/pyproject.toml

@@ -11,7 +11,7 @@ requires-python = ">=3.11"
 license = "MIT"
 keywords = []
 classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",]
-dependencies = [ "txl ==0.3.4", "pycrdt >=0.14.1,<0.15.0",]
+dependencies = [ "txl ==0.3.4", "pycrdt >=0.14.10,<0.15.0",]
 [[project.authors]]
 name = "David Brochart"
 email = "david.brochart@gmail.com"

plugins/console/pyproject.toml

@@ -11,7 +11,7 @@ requires-python = ">=3.11"
 license = "MIT"
 keywords = []
 classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",]
-dependencies = [ "txl ==0.3.4", "jupyter-ydoc >=4.1.1,<5.0.0",]
+dependencies = [ "txl ==0.3.4", "jupyter-ydoc >=4.2.0,<5.0.0",]
 [[project.authors]]
 name = "David Brochart"
 email = "david.brochart@gmail.com"

plugins/kernel/pyproject.toml

@@ -11,7 +11,7 @@ requires-python = ">=3.11"
 license = "MIT"
 keywords = []
 classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",]
-dependencies = [ "txl ==0.3.4", "python-dateutil >=2.8.2", "pycrdt >=0.14.1,<0.15.0",]
+dependencies = [ "txl ==0.3.4", "python-dateutil >=2.8.2", "pycrdt >=0.14.10,<0.15.0",]
 [[project.authors]]
 name = "David Brochart"
 email = "david.brochart@gmail.com"

plugins/local_contents/pyproject.toml

@@ -11,7 +11,7 @@ requires-python = ">=3.11"
 license = "MIT"
 keywords = []
 classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",]
-dependencies = [ "txl ==0.3.4", "anyio >=4.14.2,<5", "jupyter-ydoc >=4.1.1,<5.0.0",]
+dependencies = [ "txl ==0.3.4", "anyio >=4.14.2,<5", "jupyter-ydoc >=4.2.0,<5.0.0",]
 [[project.authors]]
 name = "David Brochart"
 email = "david.brochart@gmail.com"

plugins/local_contents/txl_local_contents/main.py

@@ -21,6 +21,8 @@ async def get(
         format: str | None = None,
     ) -> Union[List, Doc]:
         p = Path(path)
+        if not await p.exists():
+            raise FileNotFoundError(f"File not found: {path}")
         assert (await p.is_dir()) == is_dir
         if await p.is_dir():
             return sorted(list(scandir(path)), key=lambda entry: (not entry.is_dir(), entry.name))

plugins/notebook_editor/pyproject.toml

@@ -11,7 +11,7 @@ requires-python = ">=3.11"
 license = "MIT"
 keywords = []
 classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",]
-dependencies = [ "httpx2>=2.6.0", "txl ==0.3.4", "pycrdt >=0.14.1,<0.15.0", "jupyter-ydoc >=4.1.1,<5.0.0",]
+dependencies = [ "httpx2>=2.6.0", "txl ==0.3.4", "pycrdt >=0.14.10,<0.15.0", "jupyter-ydoc >=4.2.0,<5.0.0",]
 [[project.authors]]
 name = "David Brochart"
 email = "david.brochart@gmail.com"

plugins/remote_contents/pyproject.toml

@@ -11,7 +11,7 @@ requires-python = ">=3.11"
 license = "MIT"
 keywords = []
 classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",]
-dependencies = [ "txl ==0.3.4", "httpx2[ws]>=2.6.0", "pycrdt >=0.14.1,<0.15.0", "pycrdt-websocket >=0.16.4,<0.17.0", "jupyter-ydoc >=4.1.1,<5.0.0",]
+dependencies = [ "txl ==0.3.4", "httpx2[ws]>=2.6.0", "pycrdt >=0.14.10,<0.15.0", "pycrdt-websocket >=0.16.4,<0.17.0", "jupyter-ydoc >=4.2.0,<5.0.0",]
 [[project.authors]]
 name = "David Brochart"
 email = "david.brochart@gmail.com"

plugins/remote_contents/txl_remote_contents/main.py

@@ -5,33 +5,16 @@
 from urllib import parse
 
 import httpx2
-from anyio import Event, create_task_group, sleep_forever
+from anyio import TASK_STATUS_IGNORED, Event, create_task_group, sleep_forever
 from fps import Module
-from pycrdt import Doc, Provider, YMessageType, YSyncMessageType
+from pycrdt import Doc, Provider
 from pycrdt.websocket.websocket import HttpxWebsocket
 
 from txl.base import Contents
 
 ydocs = {ep.name: ep.load() for ep in entry_points(group="jupyter_ydoc")}
 
 
-class SyncedWebsocket(HttpxWebsocket):
-    """Signal after the provider has applied the initial sync response."""
-
-    def __init__(self, websocket, path, synced=None):
-        super().__init__(websocket, path)
-        self.synced = Event() if synced is None else synced
-        self.received_sync = False
-
-    async def __anext__(self):
-        if self.received_sync:
-            self.synced.set()
-        message = await super().__anext__()
-        if message[:2] == bytes([YMessageType.SYNC, YSyncMessageType.SYNC_STEP2]):
-            self.received_sync = True
-        return message
-
-
 class Entry:
     """Provide a scandir-like API"""
 
@@ -71,6 +54,7 @@ def __init__(
         i = base_url.find(":")
         self.ws_url = ("wss" if base_url[i - 1] == "s" else "ws") + base_url[i:]
         self.document_id = {}
+        self._connections = {}
 
     async def get(
         self,
@@ -88,6 +72,8 @@ async def get(
                     cookies=self.cookies,
                 )
             self.cookies.update(r.cookies)
+            if r.status_code == 404:
+                raise FileNotFoundError(f"File not found: {path}")
             r.raise_for_status()
             model = r.json()
             if model["type"] == "directory":
@@ -105,17 +91,18 @@ async def get(
         else:
             if format is not None:
                 pass
+            elif type == "notebook":
+                format = "json"
             elif type == "blob":
                 format = "base64"
             else:
                 format = "text"
-            doc_type = type  # if type == "notebook" else "file"
-            async with httpx2.AsyncClient() as client:
+            doc_type = type
+            async with httpx2.AsyncClient(cookies=self.cookies) as client:
                 response = await client.put(
                     f"{self.base_url}/api/collaboration/session/{path}",
                     json={"format": format, "type": doc_type},
                     params={**self.query_params},
-                    cookies=self.cookies,
                 )
             self.cookies.update(response.cookies)
             response.raise_for_status()
@@ -124,9 +111,10 @@ async def get(
             session_id = r["sessionId"]
             ydoc = Doc()
             jupyter_ydoc = ydocs[type](ydoc)
-            synced = Event()
-            self.task_group.start_soon(self.websocket_provider, room_id, ydoc, session_id, synced)
-            await synced.wait()
+            provider = await self.task_group.start(
+                self.websocket_provider, room_id, ydoc, session_id
+            )
+            await provider.synced.wait()
             self.document_id[jupyter_ydoc] = room_id
             return jupyter_ydoc
 
@@ -135,6 +123,10 @@ async def save(
         path: str,
         jupyter_ydoc: Doc,
     ) -> None:
+        if self.collaborative:
+            if jupyter_ydoc not in self.document_id:
+                raise ValueError("Open the document before saving it collaboratively.")
+            return
         source = jupyter_ydoc.source
         if isinstance(source, dict):
             cnt = source
@@ -163,15 +155,32 @@ async def save(
         self.cookies.update(r.cookies)
         r.raise_for_status()
 
-    async def websocket_provider(self, room_id, ydoc, session_id=None, synced=None):
+    async def websocket_provider(
+        self, room_id, ydoc, session_id=None, *, task_status=TASK_STATUS_IGNORED
+    ):
         ws_url = f"{self.ws_url}/api/collaboration/room/{room_id}"
-        params = None if session_id is None else {"sessionId": session_id}
-        async with httpx2.AsyncClient(cookies=self.cookies) as client, client.websocket(
-            ws_url, params=params
-        ) as websocket:
-            channel = SyncedWebsocket(websocket, room_id, synced)
-            async with Provider(ydoc, channel):
-                await sleep_forever()
+        params = {**self.query_params}
+        if session_id is not None:
+            params["sessionId"] = session_id
+        closing, closed = Event(), Event()
+        self._connections[ydoc] = closing, closed
+        try:
+            async with httpx2.AsyncClient(cookies=self.cookies) as client, client.websocket(
+                ws_url, params=params
+            ) as websocket:
+                async with Provider(ydoc, HttpxWebsocket(websocket, room_id)) as provider:
+                    task_status.started(provider)
+                    await closing.wait()
+        finally:
+            del self._connections[ydoc]
+            closed.set()
+
+    async def aclose(self):
+        connections = list(self._connections.values())
+        for closing, _ in connections:
+            closing.set()
+        for _, closed in connections:
+            await closed.wait()
 
 
 class RemoteContentsModule(Module):
@@ -187,6 +196,10 @@ async def start(self) -> None:
         cookies = httpx2.Cookies()
         async with create_task_group() as tg:
             contents = RemoteContents(base_url, query_params, cookies, self.collaborative, tg)
+            self.contents = contents
             self.put(contents, Contents)
             self.done()
             await sleep_forever()
+
+    async def stop(self) -> None:
+        await self.contents.aclose()

plugins/remote_kernels/pyproject.toml

@@ -11,7 +11,7 @@ requires-python = ">=3.11"
 license = "MIT"
 keywords = []
 classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",]
-dependencies = [ "txl ==0.3.4", "txl_kernel >=0.3.10,<0.4.0", "httpx2[ws]>=2.6.0", "pycrdt >=0.14.1,<0.15.0",]
+dependencies = [ "txl ==0.3.4", "txl_kernel >=0.3.10,<0.4.0", "httpx2[ws]>=2.6.0", "pycrdt >=0.14.10,<0.15.0",]
 [[project.authors]]
 name = "David Brochart"
 email = "david.brochart@gmail.com"

plugins/widgets/pyproject.toml

@@ -11,7 +11,7 @@ requires-python = ">=3.11"
 license = "MIT"
 keywords = []
 classifiers = [ "Development Status :: 4 - Beta", "Programming Language :: Python", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy",]
-dependencies = [ "txl ==0.3.4", "ypywidgets >=0.9.9,<0.10.0", "ypywidgets-textual >=0.5.5,<0.6.0", "pycrdt >=0.14.1,<0.15.0",]
+dependencies = [ "txl ==0.3.4", "ypywidgets >=0.9.9,<0.10.0", "ypywidgets-textual >=0.5.5,<0.6.0", "pycrdt >=0.14.10,<0.15.0",]
 [[project.authors]]
 name = "David Brochart"
 email = "david.brochart@gmail.com"

pyproject.toml

@@ -52,7 +52,7 @@ dependencies = [
 [dependency-groups]
 test = [
     "pytest",
-    "jupyverse[jupyterlab,noauth]",
+    "jupyverse[jupyterlab,noauth,collaboration]",
 ]
 docs = [
     "mkdocs",

src/jpterm/cli.py

@@ -79,8 +79,6 @@ def command(ctx, **kwargs):
         else:
             if kwargs["collaborative"] and not kwargs["server"]:
                 raise click.UsageError("--collaborative requires --server for CLI commands.")
-            if kwargs["collaborative"] and ctx.invoked_subcommand == "write":
-                raise click.UsageError("write does not support --collaborative yet.")
             ctx.obj = jpterm_main(kwargs)
 
     # Keep the existing TUI options on the root command.

src/jpterm/write.py

@@ -11,7 +11,7 @@
 from .app import create_app
 
 
-async def write_document(path, source, cell_id, settings, timeout):
+async def write_document(path, source, cell_id, settings, timeout, collaborative=False):
     notebook = PurePosixPath(path).suffix == ".ipynb"
     if notebook and cell_id is None:
         raise ValueError("Writing a notebook requires --cell CELL_ID.")
@@ -23,24 +23,36 @@ async def write_document(path, source, cell_id, settings, timeout):
             contents = await module.get(Contents)
             if notebook:
                 document = await contents.get(path, type="notebook")
-                notebook_source = document.source
-                cell = next(
-                    (cell for cell in notebook_source["cells"] if cell.get("id") == cell_id), None
+                index = next(
+                    (i for i, cell in enumerate(document.source["cells"])
+                     if cell.get("id") == cell_id), None
                 )
-                if cell is None:
+                if index is None:
                     raise ValueError(f"Cell ID not found: {cell_id}")
-                if cell["source"] == source:
+                cell = document.ycells[index]
+                if str(cell["source"]) == source:
                     return
-                cell["source"] = source
-                if cell["cell_type"] == "code":
-                    cell["outputs"] = []
-                    cell["execution_count"] = None
-                document.source = notebook_source
+                with document.ydoc.transaction():
+                    document.set_cell(index, source=source)
+                    if cell["cell_type"] == "code":
+                        cell["outputs"].clear()
+                        cell["execution_count"] = None
+                    document.dirty = True
             else:
-                factory = next(iter(entry_points(group="jupyter_ydoc", name="file"))).load()
-                document = factory()
-                document.source = source
-            await contents.save(path, document)
+                try:
+                    document = await contents.get(path, type="file", format="text")
+                except FileNotFoundError:
+                    if collaborative:
+                        raise
+                    factory = next(iter(entry_points(group="jupyter_ydoc", name="file"))).load()
+                    document = factory()
+                if collaborative and document.source == source:
+                    return
+                with document.ydoc.transaction():
+                    document.source = source
+                    document.dirty = True
+            if not collaborative:
+                await contents.save(path, document)
 
 
 @click.command()
@@ -65,6 +77,7 @@ def write(ctx, path, cell_id, source, from_stdin, timeout):
     try:
         anyio.run(
             write_document, path, source, cell_id, options["set_"], timeout,
+            ctx.parent.params["collaborative"],
             backend=options["backend"],
         )
     except Exception as exc: