feat(runtimed): streaming notebook load with jiter, add_cell_full, and blob store outputs

Cargo.toml

@@ -38,6 +38,7 @@ schemars = "1"
 notify = "8"
 notify-debouncer-mini = "0.7"
 ts-rs = { version = "12", features = ["serde-compat"] }
+jiter = { version = "0.13", default-features = false, features = ["num-bigint"] }
 
 [workspace.lints.rust]
 dead_code = "warn"

crates/notebook-doc/src/lib.rs

@@ -372,6 +372,49 @@ impl NotebookDoc {
         Ok(())
     }
 
+    /// Insert a fully-populated cell at the given index in a single operation.
+    ///
+    /// Unlike calling `add_cell` + `update_source` + `set_outputs` +
+    /// `set_execution_count` sequentially, this method reuses the `ObjId`
+    /// returned by each Automerge insertion — no `find_cell_index` lookup
+    /// is needed. This eliminates the O(n) linear scan per operation that
+    /// makes sequential calls O(n²) during bulk loads.
+    pub fn add_cell_full(
+        &mut self,
+        index: usize,
+        cell_id: &str,
+        cell_type: &str,
+        source: &str,
+        outputs: &[String],
+        execution_count: &str,
+    ) -> Result<(), AutomergeError> {
+        let cells_id = self
+            .cells_list_id()
+            .ok_or_else(|| AutomergeError::InvalidObjId("cells list not found".into()))?;
+
+        let len = self.doc.length(&cells_id);
+        let index = index.min(len);
+
+        let cell_map = self.doc.insert_object(&cells_id, index, ObjType::Map)?;
+        self.doc.put(&cell_map, "id", cell_id)?;
+        self.doc.put(&cell_map, "cell_type", cell_type)?;
+
+        let source_id = self.doc.put_object(&cell_map, "source", ObjType::Text)?;
+        if !source.is_empty() {
+            self.doc.update_text(&source_id, source)?;
+        }
+
+        self.doc
+            .put(&cell_map, "execution_count", execution_count)?;
+
+        let outputs_id = self.doc.put_object(&cell_map, "outputs", ObjType::List)?;
+        for (i, output) in outputs.iter().enumerate() {
+            self.doc.insert(&outputs_id, i, output.as_str())?;
+        }
+
+        Ok(())
+    }
+
     /// Delete a cell by ID. Returns `true` if the cell was found and deleted.
     pub fn delete_cell(&mut self, cell_id: &str) -> Result<bool, AutomergeError> {
         let cells_id = match self.cells_list_id() {
@@ -387,6 +430,20 @@ impl NotebookDoc {
         }
     }
 
+    /// Remove all cells from the document.
+    ///
+    /// Used to clean up after a failed streaming load so the next
+    /// connection can retry from a clean state.
+    pub fn clear_all_cells(&mut self) {
+        if let Some(cells_id) = self.cells_list_id() {
+            let len = self.doc.length(&cells_id);
+            // Delete from the end to avoid index shifting
+            for i in (0..len).rev() {
+                let _ = self.doc.delete(&cells_id, i);
+            }
+        }
+    }
+
     // ── Source editing ───────────────────────────────────────────────
 
     /// Replace a cell's source text.

crates/runtimed/Cargo.toml

@@ -90,6 +90,9 @@ ts-rs = { workspace = true }
 notify = { workspace = true }
 notify-debouncer-mini = { workspace = true }
 
+# Fast JSON parsing for notebook loading
+jiter = { workspace = true }
+
 [target.'cfg(unix)'.dependencies]
 libc = "0.2"
 nix = { version = "0.30", features = ["signal", "process"] }

crates/runtimed/src/daemon.rs

@@ -952,6 +952,7 @@ impl Daemon {
                     working_dir_path,
                     initial_metadata,
                     false, // Send ProtocolCapabilities for legacy NotebookSync handshake
+                    None,  // No streaming load for legacy handshake
                 )
                 .await
             }
@@ -1003,58 +1004,32 @@ impl Daemon {
             )
         };
 
-        // Check-and-load atomically under write lock to prevent race conditions.
-        // Two concurrent opens could both see cell_count == 0 if we check with
-        // read lock first, leading to duplicate cells.
-        let cell_count = {
-            let mut doc = room.doc.write().await;
+        // Check whether this connection needs to stream-load the notebook
+        // from disk. The actual load is deferred to the sync loop so cells
+        // arrive progressively via Automerge sync messages.
+        let (cell_count, needs_load) = {
+            let doc = room.doc.read().await;
             let existing_count = doc.cell_count();
-            if existing_count == 0 {
-                // First connection - load from disk
-                match crate::notebook_sync_server::load_notebook_from_disk(
-                    &mut doc,
-                    &path_buf,
-                    &room.blob_store,
-                )
-                .await
-                {
-                    Ok(count) => {
-                        info!("[runtimed] Loaded {} cells from {} into room", count, path);
-                        count
-                    }
-                    Err(e) => {
-                        // Drop the doc lock before removing room
-                        drop(doc);
-                        // Remove the room to prevent stale trust state on retry
-                        {
-                            let mut rooms = self.notebook_rooms.lock().await;
-                            rooms.remove(&notebook_id);
-                            info!("[runtimed] Removed room {} after load failure", notebook_id);
-                        }
-                        // Send error response and return
-                        let (mut reader, mut writer) = tokio::io::split(stream);
-                        let response = NotebookConnectionInfo {
-                            protocol: PROTOCOL_V2.to_string(),
-                            protocol_version: Some(PROTOCOL_VERSION),
-                            daemon_version: Some(crate::daemon_version().to_string()),
-                            notebook_id: String::new(),
-                            cell_count: 0,
-                            needs_trust_approval: false,
-                            error: Some(format!("Failed to load notebook: {}", e)),
-                        };
-                        send_json_frame(&mut writer, &response).await?;
-                        // Drain any remaining data from reader to avoid broken pipe
-                        let _ = tokio::io::copy(&mut reader, &mut tokio::io::sink()).await;
-                        return Ok(());
-                    }
-                }
+            if existing_count == 0 && !room.is_loading.load(std::sync::atomic::Ordering::Acquire) {
+                // Room is empty and nobody is loading yet — this connection
+                // will do the streaming load inside the sync loop.
+                info!(
+                    "[runtimed] Room for {} is empty, deferring streaming load",
+                    path
+                );
+                (0, Some(path_buf.clone()))
             } else {
-                // Room already has cells from another connection
                 info!(
-                    "[runtimed] Room for {} already has {} cells (joining existing)",
-                    path, existing_count
+                    "[runtimed] Room for {} has {} cells (joining existing{})",
+                    path,
+                    existing_count,
+                    if room.is_loading.load(std::sync::atomic::Ordering::Acquire) {
+                        ", load in progress"
+                    } else {
+                        ""
+                    }
                 );
-                existing_count
+                (existing_count, None)
             }
         };
 
@@ -1089,7 +1064,9 @@ impl Daemon {
         // Drop the trust_state lock before continuing
         drop(trust_state);
 
-        // Continue with normal notebook sync (handles auto-launch internally)
+        // Continue with normal notebook sync (handles auto-launch internally).
+        // If needs_load is Some, the sync loop will stream cells from disk
+        // before entering the steady-state select loop.
         crate::notebook_sync_server::handle_notebook_sync_connection(
             reader,
             writer,
@@ -1102,6 +1079,7 @@ impl Daemon {
             working_dir_path,
             None, // No initial_metadata - doc is already populated
             true, // Skip ProtocolCapabilities - already sent in NotebookConnectionInfo
+            needs_load,
         )
         .await
     }
@@ -1230,6 +1208,7 @@ impl Daemon {
             working_dir_path,
             None, // No initial_metadata - doc is already populated
             true, // Skip ProtocolCapabilities - already sent in NotebookConnectionInfo
+            None, // No streaming load - doc was just created with empty cell
         )
         .await
     }