diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index f6f0cc0..eeb35af 100644
--- a/.github/workflows/test.yml
+++ b/.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
diff --git a/docs/usage/CLI.md b/docs/usage/CLI.md
index 5151b96..de42daf 100644
--- a/docs/usage/CLI.md
+++ b/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
+```
diff --git a/plugins/cell/pyproject.toml b/plugins/cell/pyproject.toml
index 782cd04..e01e202 100644
--- a/plugins/cell/pyproject.toml
+++ b/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"
diff --git a/plugins/console/pyproject.toml b/plugins/console/pyproject.toml
index 73ac56b..e59b93d 100644
--- a/plugins/console/pyproject.toml
+++ b/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"
diff --git a/plugins/kernel/pyproject.toml b/plugins/kernel/pyproject.toml
index 30a7dbd..9d7c2d3 100644
--- a/plugins/kernel/pyproject.toml
+++ b/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"
diff --git a/plugins/local_contents/pyproject.toml b/plugins/local_contents/pyproject.toml
index 6b8bacb..bc358f8 100644
--- a/plugins/local_contents/pyproject.toml
+++ b/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"
diff --git a/plugins/local_contents/txl_local_contents/main.py b/plugins/local_contents/txl_local_contents/main.py
index fa4f419..ac043ae 100644
--- a/plugins/local_contents/txl_local_contents/main.py
+++ b/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))
diff --git a/plugins/notebook_editor/pyproject.toml b/plugins/notebook_editor/pyproject.toml
index bd56d6f..8074cc1 100644
--- a/plugins/notebook_editor/pyproject.toml
+++ b/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"
diff --git a/plugins/remote_contents/pyproject.toml b/plugins/remote_contents/pyproject.toml
index a8bc11d..e0117eb 100644
--- a/plugins/remote_contents/pyproject.toml
+++ b/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"
diff --git a/plugins/remote_contents/txl_remote_contents/main.py b/plugins/remote_contents/txl_remote_contents/main.py
index 8ec7930..c6ee34e 100644
--- a/plugins/remote_contents/txl_remote_contents/main.py
+++ b/plugins/remote_contents/txl_remote_contents/main.py
@@ -5,9 +5,9 @@
 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
@@ -15,23 +15,6 @@
 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()
diff --git a/plugins/remote_kernels/pyproject.toml b/plugins/remote_kernels/pyproject.toml
index 4fb6db3..ba875b8 100644
--- a/plugins/remote_kernels/pyproject.toml
+++ b/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"
diff --git a/plugins/widgets/pyproject.toml b/plugins/widgets/pyproject.toml
index 4cca175..8d81269 100644
--- a/plugins/widgets/pyproject.toml
+++ b/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"
diff --git a/pyproject.toml b/pyproject.toml
index 880f27e..d7eb3a5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -52,7 +52,7 @@ dependencies = [
 [dependency-groups]
 test = [
     "pytest",
-    "jupyverse[jupyterlab,noauth]",
+    "jupyverse[jupyterlab,noauth,collaboration]",
 ]
 docs = [
     "mkdocs",
diff --git a/src/jpterm/cli.py b/src/jpterm/cli.py
index 6b9a758..b477911 100644
--- a/src/jpterm/cli.py
+++ b/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.
diff --git a/src/jpterm/write.py b/src/jpterm/write.py
index f031dc0..b83acb3 100644
--- a/src/jpterm/write.py
+++ b/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:
diff --git a/tests/test_server_cli.py b/tests/test_server_cli.py
index 1d6f6e7..05dc43e 100644
--- a/tests/test_server_cli.py
+++ b/tests/test_server_cli.py
@@ -1,4 +1,4 @@
-"""CLI integration checks against a real, non-collaborative Jupyverse server."""
+"""CLI integration checks against a real Jupyverse server, including RTC."""
 
 import json
 import os
@@ -9,9 +9,11 @@
 import time
 from types import SimpleNamespace
 
+import anyio
 import httpx2
 import pytest
 from rich.text import Text
+from txl_remote_contents.main import RemoteContents
 
 
 @pytest.fixture(scope="module")
@@ -54,6 +56,8 @@ def jupyverse_server(tmp_path_factory):
     }
     (server_root / "remote.ipynb").write_text(json.dumps(server.notebook), encoding="utf-8")
     (server_root / "writable.ipynb").write_text(json.dumps(server.notebook), encoding="utf-8")
+    (server_root / "rtc.ipynb").write_text(json.dumps(server.notebook), encoding="utf-8")
+    (server_root / "rtc.txt").write_text("Original shared text", encoding="utf-8")
     server.text = "SERVER LICENSE\n\nOnly available from the server.\n"
     # Write exact bytes so Windows does not translate LF into CRLF.
     (server_root / "LICENSE").write_bytes(server.text.encode("utf-8"))
@@ -110,7 +114,7 @@ def jupyverse_server(tmp_path_factory):
 @pytest.fixture
 def run_cli(jupyverse_server):
 
-    def run(*arguments, expected_code=0, command="show", input=None):
+    def run(*arguments, expected_code=0, command="show", input=None, collaborative=False):
         environment = dict(os.environ, PYTHONIOENCODING="utf-8", NO_COLOR="1", COLUMNS="80")
         result = subprocess.run(
             [
@@ -119,6 +123,7 @@ def run(*arguments, expected_code=0, command="show", input=None):
                 "from jpterm.cli import main; main()",
                 "--server",
                 jupyverse_server.url,
+                *(["--collaborative"] if collaborative else []),
                 command,
                 *arguments,
             ],
@@ -219,7 +224,7 @@ def test_show_markdown(run_cli, jupyverse_server):
 
 def test_missing_file(run_cli, jupyverse_server):
     result = run_cli("missing.ipynb", expected_code=1)
-    assert "404" in result.stderr
+    assert "File not found: missing.ipynb" in result.stderr
 
 
 @pytest.mark.parametrize("from_stdin", [False, True])
@@ -245,3 +250,58 @@ def test_write_server_cell(run_cli):
     result = run_cli(path, "--cell", "missing", "--source", "bad", command="write", expected_code=1)
     assert "Cell ID not found" in result.stderr
     assert json.loads(run_cli(path, "--json").stdout) == after
+
+
+@pytest.mark.parametrize("path", ["remote.ipynb", "LICENSE"])
+@pytest.mark.parametrize("as_json", [False, True])
+def test_collaborative_show(run_cli, path, as_json):
+    arguments = [path, *(["--json"] if as_json else [])]
+    expected = run_cli(*arguments)
+    actual = run_cli(*arguments, collaborative=True)
+    if as_json:
+        assert json.loads(actual.stdout) == json.loads(expected.stdout)
+    else:
+        assert actual.stdout + actual.stderr == expected.stdout + expected.stderr
+
+
+def test_collaborative_write_text(run_cli):
+    source = "Shared text π\nsecond line"
+    run_cli("rtc.txt", "--stdin", input=source, command="write", collaborative=True)
+    actual = json.loads(run_cli("rtc.txt", "--json", collaborative=True).stdout)
+    assert actual.replace("\r\n", "\n") == source
+
+
+def test_collaborative_write_reaches_connected_peer(run_cli, jupyverse_server):
+    async def check():
+        with anyio.fail_after(20):
+            async with anyio.create_task_group() as tasks:
+                contents = RemoteContents(
+                    jupyverse_server.url, {}, httpx2.Cookies(), True, tasks
+                )
+                document = await contents.get("rtc.ipynb", type="notebook")
+                peer_source = document.ycells[0]["source"]
+                target_source = document.ycells[1]["source"]
+                changed = anyio.Event()
+                subscription = target_source.observe(lambda event: changed.set())
+                peer_source.clear()
+                peer_source += "Edited by another client"
+
+                def write():
+                    run_cli(
+                        "rtc.ipynb", "--cell", "server-cell", "--source", "print(42)",
+                        command="write", collaborative=True,
+                    )
+
+                await anyio.to_thread.run_sync(write)
+                await changed.wait()
+                target_source.unobserve(subscription)
+                assert str(peer_source) == "Edited by another client"
+                assert str(target_source) == "print(42)"
+                assert document.source["cells"][1]["outputs"] == []
+                assert document.source["cells"][1]["execution_count"] is None
+                await contents.aclose()
+
+    anyio.run(check)
+    source = json.loads(run_cli("rtc.ipynb", "--json", collaborative=True).stdout)
+    assert source["cells"][0]["source"] == "Edited by another client"
+    assert source["cells"][1]["source"] == "print(42)"
diff --git a/tests/test_show.py b/tests/test_show.py
index 7134fc6..c58f673 100644
--- a/tests/test_show.py
+++ b/tests/test_show.py
@@ -10,7 +10,6 @@
 from fps import Module, get_root_module
 from jpterm.cli import jpterm_main, main
 from jpterm.show import load_document
-from pycrdt import YMessageType, YSyncMessageType
 from rich.console import Console
 from textual.app import App
 from textual.widgets import Static
@@ -19,7 +18,7 @@
 from txl_jpterm.main_area import MainArea
 from txl_markdown_viewer.main import MarkdownViewer, MarkdownViewerModule
 from txl_notebook_editor.main import NotebookEditor, NotebookEditorModule
-from txl_remote_contents.main import RemoteContents, SyncedWebsocket, ydocs
+from txl_remote_contents.main import RemoteContents, ydocs
 from txl_text_editor.main import TextEditor, TextEditorModule
 
 from txl.app import AppModule
@@ -274,10 +273,9 @@ async def check():
         async with anyio.create_task_group() as tasks:
             contents = RemoteContents("http://server", {}, httpx2.Cookies(), True, tasks)
 
-            async def provider(room, document, session, synced):
+            async def provider(room, document, session, *, task_status):
                 provider_started.set()
-                await allow_sync.wait()
-                synced.set()
+                task_status.started(SimpleNamespace(synced=allow_sync))
 
             contents.websocket_provider = provider
 
@@ -370,21 +368,3 @@ def test_error():
     with patch("jpterm.show.load_document", AsyncMock(side_effect=ValueError("bad notebook"))):
         code, _ = invoke(["show", "bad.ipynb"])
         assert code == 1
-
-
-def test_sync_signal_follows_application():
-
-    async def check():
-        synced = anyio.Event()
-        websocket = AsyncMock()
-        websocket.receive_bytes.side_effect = [
-            bytes([YMessageType.SYNC, YSyncMessageType.SYNC_STEP2]),
-            b"next",
-        ]
-        channel = SyncedWebsocket(websocket, "room", synced)
-        await channel.__anext__()
-        assert not synced.is_set()
-        await channel.__anext__()
-        assert synced.is_set()
-
-    anyio.run(check)
diff --git a/tests/test_websockets.py b/tests/test_websockets.py
index 0774c1e..cc91947 100644
--- a/tests/test_websockets.py
+++ b/tests/test_websockets.py
@@ -7,7 +7,8 @@
 import httpx2
 import pytest
 from httpx2.websockets import ASGIWebSocketTransport
-from pycrdt import Doc, Text, handle_sync_message
+from pycrdt import Doc, Text, YMessageType, YSyncMessageType, handle_sync_message
+from pycrdt.websocket.websocket import HttpxWebsocket
 from txl_remote_contents.main import RemoteContents
 from txl_remote_kernels.driver import KernelDriver
 from txl_remote_terminals.main import RemoteTerminals
@@ -28,7 +29,7 @@ def test_contents_websocket_sync(monkeypatch):
 
     async def app(scope, receive, send):
         assert scope["path"] == "/api/collaboration/room/test-room"
-        assert scope["query_string"] == b"sessionId=test-session"
+        assert scope["query_string"] == b"token=test-token&sessionId=test-session"
         assert b"session-cookie=yes" in dict(scope["headers"])[b"cookie"]
         assert (await receive())["type"] == "websocket.connect"
         await send({"type": "websocket.accept"})
@@ -45,20 +46,73 @@ async def app(scope, receive, send):
     async def check():
         client_doc = Doc()
         client_doc["source"] = Text()
-        synced = anyio.Event()
-        finished = anyio.Event()
-        monkeypatch.setattr("txl_remote_contents.main.sleep_forever", finished.wait)
         with anyio.fail_after(3):
             async with anyio.create_task_group() as tasks:
                 contents = RemoteContents(
-                    "http://server", {}, httpx2.Cookies({"session-cookie": "yes"}), True, tasks,
+                    "http://server", {"token": ["test-token"]},
+                    httpx2.Cookies({"session-cookie": "yes"}), True, tasks,
                 )
-                tasks.start_soon(
-                    contents.websocket_provider, "test-room", client_doc, "test-session", synced,
+                provider = await tasks.start(
+                    contents.websocket_provider, "test-room", client_doc, "test-session"
                 )
-                await synced.wait()
+                await provider.synced.wait()
                 assert str(client_doc["source"]) == "Synchronized through HTTPX2"
-                finished.set()
+                await contents.aclose()
+
+    anyio.run(check)
+
+
+def test_contents_shutdown_drains_pending_updates(monkeypatch):
+    server_doc = Doc()
+    server_doc["source"] = Text("original")
+
+    async def check():
+        send_started = anyio.Event()
+        allow_send = anyio.Event()
+        closed = anyio.Event()
+        slow = False
+
+        class SlowWebsocket(HttpxWebsocket):
+            async def send(self, message):
+                if slow and message[:2] == bytes([YMessageType.SYNC, YSyncMessageType.SYNC_UPDATE]):
+                    send_started.set()
+                    await allow_send.wait()
+                await super().send(message)
+
+        async def app(scope, receive, send):
+            await receive()
+            await send({"type": "websocket.accept"})
+            while True:
+                event = await receive()
+                if event["type"] == "websocket.disconnect":
+                    return
+                reply = handle_sync_message(event["bytes"][1:], server_doc)
+                if reply is not None:
+                    await send({"type": "websocket.send", "bytes": reply})
+
+        monkeypatch.setattr(httpx2, "AsyncClient", client_factory(app))
+        monkeypatch.setattr("txl_remote_contents.main.HttpxWebsocket", SlowWebsocket)
+        client_doc = Doc()
+        client_doc["source"] = Text()
+        with anyio.fail_after(5):
+            async with anyio.create_task_group() as tasks:
+                contents = RemoteContents("http://server", {}, httpx2.Cookies(), True, tasks)
+                provider = await tasks.start(contents.websocket_provider, "room", client_doc)
+                await provider.synced.wait()
+                slow = True
+                client_doc["source"].clear()
+                client_doc["source"] += "changed π"
+
+                async def close():
+                    await contents.aclose()
+                    closed.set()
+
+                tasks.start_soon(close)
+                await send_started.wait()
+                assert not closed.is_set()
+                allow_send.set()
+                await closed.wait()
+                assert str(server_doc["source"]) == "changed π"
 
     anyio.run(check)
 
diff --git a/tests/test_write.py b/tests/test_write.py
index 1e14431..9ea7116 100644
--- a/tests/test_write.py
+++ b/tests/test_write.py
@@ -10,7 +10,7 @@
 import pytest
 from fps import Module
 from jpterm.write import write_document
-from jupyter_ydoc import YNotebook
+from jupyter_ydoc import YFile, YNotebook
 from rich.text import Text
 from textual.app import App
 from txl_remote_contents.main import RemoteContents
@@ -81,6 +81,14 @@ def test_unknown_cell_does_not_save(notebook, write_with_contents):
     contents.save.assert_not_awaited()
 
 
+@pytest.mark.parametrize("error", [PermissionError("denied"), RuntimeError("connection failed")])
+def test_text_load_errors_do_not_save(write_with_contents, error):
+    contents = SimpleNamespace(get=AsyncMock(side_effect=error), save=AsyncMock())
+    with pytest.raises(type(error), match=str(error)):
+        write_with_contents(contents, "file.txt", "new source")
+    contents.save.assert_not_awaited()
+
+
 @pytest.fixture
 def run_write(tmp_path):
     def run(*arguments, input=None, env=None):
@@ -142,8 +150,7 @@ def test_write_local_notebook(run_write, tmp_path, notebook):
     (["write", "file.txt", "--source", "text", "--stdin"], "Provide exactly one"),
     (["write", "file.txt", "--cell", "code", "--source", "text"], "requires a notebook"),
     (["write", "test.ipynb", "--source", "text"], "requires --cell"),
-    (["--server", "http://server", "--collaborative", "write", "file.txt",
-      "--source", "text"], "does not support --collaborative"),
+    (["--collaborative", "write", "file.txt", "--source", "text"], "requires --server"),
 ])
 def test_invalid_write(run_write, tmp_path, arguments, message):
     result = run_write(*arguments)
@@ -166,3 +173,31 @@ async def check():
                 await contents.save("file.txt", SimpleNamespace(source="text"))
 
     anyio.run(check)
+
+
+def test_collaborative_save_uses_shared_document():
+    async def check():
+        document = YFile()
+        document.source = "new content"
+        document.dirty = True
+        contents = RemoteContents("http://server", {}, httpx2.Cookies(), True, None)
+        contents.document_id[document] = "room"
+        with patch("txl_remote_contents.main.httpx2.AsyncClient") as client:
+            await contents.save("file.txt", document)
+            assert document.dirty
+            client.assert_not_called()
+
+    anyio.run(check)
+
+
+def test_write_cell_preserves_unchanged_text(notebook, write_with_contents):
+    text = notebook.ycells[0]["source"]
+    text.clear()
+    text += "π😀 hello world"
+    events = []
+    subscription = text.observe(lambda event: events.append(event.delta))
+    contents = SimpleNamespace(get=AsyncMock(return_value=notebook), save=AsyncMock())
+    write_with_contents(contents, "test.ipynb", "π😀 hello there", "code")
+    assert str(text) == "π😀 hello there"
+    assert events[0][0] == {"retain": len("π😀 hello ".encode("utf-8"))}
+    text.unobserve(subscription)