optimistic-apply daemon support — Loro text-delta hot path for remote inserts

The GPU frontend's per-keystroke edits arrive as FrontendEvent::CrdtOp.
The old apply path materialized the whole document and diffed it per
op — O(file) per typed character on the daemon main thread.

- CrdtState: persistent text-projection subscription (capture gated by
  an AtomicBool so per-keystroke imports don't register/drop
  callbacks); import_updates_with_text_deltas returns Loro's deltas;
  unicode_to_utf8_pos converts the insert point.
- Buffer::apply_remote_crdt_op: the common single-insert delta applies
  straight to the rope; deletes/compound updates keep the conservative
  materialize+diff fallback. UTF-8 position regression test included.
- SyntaxRegistry::has_pending_parse_job_for: the main-thread
  "parse in flight" bit render producers need for settle-gating.
- TextView::pos_to_display: stack buffer for short line prefixes +
  valid_up_to() boundary trim — removes a per-cursor-move allocation.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/buffer.rs

@@ -684,16 +684,12 @@ impl Buffer {
     /// Used by the daemon's `FrontendEvent::CrdtOp` handler when a
     /// replica frontend forwards a CRDT op. The flow:
     ///
-    /// 1. Capture the rope's current bytes (rope ≡ CRDT projection
-    ///    invariant — both have the same content pre-import).
-    /// 2. `crdt.import_updates(op_bytes)` — integrates the remote op
-    ///    into the local CRDT state. CRDT convergence handles
-    ///    concurrent edits.
-    /// 3. Materialize the post-import CRDT content.
-    /// 4. Compute the diff between pre- and post-content as a single
-    ///    `Replace` `EditOp` (single insert/delete falls out as
-    ///    Replace with empty inserted or empty range).
-    /// 5. Apply the rope stages (rope mutation + mark adjustment +
+    /// 1. `crdt.import_updates_with_text_deltas(op_bytes)` integrates
+    ///    the remote op and captures Loro's text projection delta.
+    /// 2. Apply the common single-insert shape directly to the rope.
+    ///    Deletes and compound updates conservatively fall back to a
+    ///    post-import materialization + contiguous diff.
+    /// 3. Apply the rope stages (rope mutation + mark adjustment +
     ///    revision bump + modified flag + `on_edit` broadcast).
     ///    Skips the CRDT-application stage (already done in step 2)
     ///    AND the undo push (remote ops aren't locally undoable per
@@ -750,25 +746,37 @@ impl Buffer {
             });
         };
 
-        // Step 1: capture pre-import bytes (rope ≡ CRDT projection
-        // invariant means rope.slice == crdt.materialize_string here).
+        // Integrate the remote op and capture Loro's projection diff.
+        // Optimistic GUI typing produces one Insert delta, so handle
+        // that shape without copying or materializing the document.
+        let text_deltas = crdt
+            .import_updates_with_text_deltas(op_bytes)
+            .map_err(|e| BufferError::CrdtRejected {
+                reason: format!("import_updates: {e:?}"),
+            })?;
+        if let Some((unicode_pos, inserted)) = single_remote_text_insert(&text_deltas)
+            && let Some(byte_pos) = crdt.unicode_to_utf8_pos(unicode_pos)
+        {
+            let byte_pos = byte_pos as Position;
+            let mut views = std::mem::take(&mut self.views);
+            let result =
+                self.run_remote_rope_stages(&mut views, byte_pos, byte_pos, inserted.as_bytes());
+            self.views = views;
+            return result.map(Some);
+        }
+
+        // Conservative fallback for deletes, compound updates, and
+        // already-integrated ops. The rope is still the pre-import
+        // projection, so it remains the source for the old bytes.
         let old_len = self.rope.len();
         let mut old_bytes = vec![0u8; old_len as usize];
         if old_len > 0 {
             self.rope.slice(0, old_len, &mut old_bytes);
         }
-
-        // Step 2: integrate the remote op into the CRDT state.
-        crdt.import_updates(op_bytes)
-            .map_err(|e| BufferError::CrdtRejected {
-                reason: format!("import_updates: {e:?}"),
-            })?;
-
-        // Step 3: materialize the post-import content.
         let new_content = crdt.materialize_string();
         let new_bytes = new_content.as_bytes();
 
-        // Step 4: compute common prefix/suffix at byte level, then
+        // Compute common prefix/suffix at byte level, then
         // **back off to UTF-8 char boundaries** in both strings.
         //
         // # Post-audit-round-4 F25: char-boundary alignment
@@ -829,7 +837,7 @@ impl Buffer {
         let range_end = (old_bytes.len() - suffix) as Position;
         let inserted = &new_bytes[prefix..new_bytes.len() - suffix];
 
-        // Step 5: apply rope stages without re-applying to CRDT
+        // Apply rope stages without re-applying to CRDT
         // (CRDT was applied above in step 2) and without undo push
         // (remote ops aren't locally undoable per M10.4).
         let mut views = std::mem::take(&mut self.views);
@@ -1451,6 +1459,37 @@ impl Buffer {
 #[cfg(feature = "crdt")]
 type CrdtRoutingResult = (Option<Vec<u8>>, Option<Box<crate::rope::CrdtOp>>);
 
+/// Recognize the hot-path projection delta produced by one remote
+/// insertion. Loro's retain/delete lengths use Unicode scalar offsets;
+/// the caller converts the insertion point through the post-import
+/// text container before applying the UTF-8 bytes to the rope.
+#[cfg(feature = "crdt")]
+fn single_remote_text_insert(deltas: &[Vec<loro::TextDelta>]) -> Option<(usize, &str)> {
+    let [delta] = deltas else {
+        return None;
+    };
+    let mut cursor = 0usize;
+    let mut found = None;
+    for op in delta {
+        match op {
+            loro::TextDelta::Retain { retain, .. } => {
+                cursor = cursor.checked_add(*retain)?;
+            }
+            loro::TextDelta::Insert { insert, .. } if insert.is_empty() => {}
+            loro::TextDelta::Insert { insert, .. } => {
+                if found.is_some() {
+                    return None;
+                }
+                found = Some((cursor, insert.as_str()));
+                cursor = cursor.checked_add(insert.chars().count())?;
+            }
+            loro::TextDelta::Delete { delete } if *delete == 0 => {}
+            loro::TextDelta::Delete { .. } => return None,
+        }
+    }
+    found
+}
+
 /// T M10.4: derive a fine-grained `(range, inserted_len)` Edit
 /// description for the change from `old_rope` to `new_rope` via
 /// longest-common-prefix + longest-common-suffix trim.
@@ -2884,6 +2923,35 @@ mod tests {
         assert_eq!(count.get(), 1, "on_edit must fire for remote op");
     }
 
+    #[cfg(feature = "crdt")]
+    #[test]
+    fn apply_remote_crdt_op_insert_after_multibyte_char_uses_utf8_byte_position() {
+        let donor = crate::crdt::CrdtState::new(2).expect("donor");
+        donor.insert(0, "éx").expect("seed");
+
+        let mut buf = Buffer::new_with_crdt(BufferId::next(), "*utf8-insert*", 1).expect("buf");
+        let donor_snap = donor.export_snapshot().expect("snap");
+        buf.crdt
+            .as_ref()
+            .expect("crdt")
+            .import_snapshot(&donor_snap)
+            .expect("init from snap");
+        buf.rope = crate::rope::Rope::from_bytes("éx".as_bytes());
+
+        let v_before = donor.version();
+        donor.insert("é".len(), "!").expect("insert");
+        let op_bytes = donor.export_updates_since(&v_before).expect("export");
+        let edit = buf
+            .apply_remote_crdt_op(&op_bytes)
+            .expect("apply")
+            .expect("non-empty edit");
+
+        assert_eq!(rope_string(&buf), "é!x");
+        assert_eq!(edit.range, Range::new("é".len() as u64, "é".len() as u64));
+        assert_eq!(edit.inserted_len, 1);
+        assert_invariant(&buf);
+    }
+
     /// F25 (post-audit-round-4): a CRDT update that changes one
     /// codepoint into another with a shared leading UTF-8 byte
     /// must produce a char-boundary-aligned diff. Pre-fix, the

src/crdt.rs

@@ -40,7 +40,18 @@
 //! propagation, the optional `crdt_op` field on [`crate::rope::Edit`],
 //! and the convergence proptest.
 
-use loro::{ExportMode, LoroDoc, LoroEncodeError, LoroResult, UndoManager, VersionVector};
+use std::sync::{
+    Arc, Mutex,
+    atomic::{AtomicBool, Ordering},
+};
+
+use loro::{
+    ContainerTrait, ExportMode, LoroDoc, LoroEncodeError, LoroResult, TextDelta, UndoManager,
+    VersionVector,
+};
+
+type TextDeltaBatches = Arc<Mutex<Vec<Vec<TextDelta>>>>;
+type TextDeltaSubscription = (TextDeltaBatches, Arc<AtomicBool>, loro::Subscription);
 
 /// The CRDT-backed buffer state.
 ///
@@ -54,6 +65,12 @@ use loro::{ExportMode, LoroDoc, LoroEncodeError, LoroResult, UndoManager, Versio
 /// (foreground worker materializes the initial projection).
 pub struct CrdtState {
     doc: LoroDoc,
+    /// Text projection deltas captured only while a remote import is
+    /// active. Keeping the subscription alive avoids registering and
+    /// dropping one callback for every typed character.
+    text_delta_batches: TextDeltaBatches,
+    text_delta_capture_enabled: Arc<AtomicBool>,
+    _text_delta_subscription: loro::Subscription,
     /// T M10.4: per-peer undo machinery. Bound to `doc`'s `peer_id`
     /// at construction; produces inverse ops attributed to that peer.
     ///
@@ -90,13 +107,45 @@ impl CrdtState {
         // explicit get here ensures the container is registered before
         // any read or write.
         let _ = doc.get_text("body");
+        let (text_delta_batches, text_delta_capture_enabled, text_delta_subscription) =
+            Self::subscribe_text_deltas(&doc);
         let undo = Self::create_undo_manager(&doc);
         Ok(Self {
             doc,
+            text_delta_batches,
+            text_delta_capture_enabled,
+            _text_delta_subscription: text_delta_subscription,
             undo: std::cell::RefCell::new(undo),
         })
     }
 
+    fn subscribe_text_deltas(doc: &LoroDoc) -> TextDeltaSubscription {
+        let text = doc.get_text("body");
+        let batches = Arc::new(Mutex::new(Vec::<Vec<TextDelta>>::new()));
+        let capture_enabled = Arc::new(AtomicBool::new(false));
+        let captured_batches = Arc::clone(&batches);
+        let captured_enabled = Arc::clone(&capture_enabled);
+        let subscription = doc.subscribe(
+            &text.id(),
+            Arc::new(move |event| {
+                if !captured_enabled.load(Ordering::Relaxed) {
+                    return;
+                }
+                let mut guard = captured_batches
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner);
+                for event in event.events {
+                    if let Some(delta) = event.diff.as_text()
+                        && !delta.is_empty()
+                    {
+                        guard.push(delta.clone());
+                    }
+                }
+            }),
+        );
+        (batches, capture_enabled, subscription)
+    }
+
     /// T M10.4: construct a fresh `UndoManager` bound to the given doc.
     /// Extracted as a helper because `from_bytes` constructs it AFTER
     /// the initial seed insert (so the seed isn't observable as an
@@ -136,9 +185,14 @@ impl CrdtState {
         // The buffer's starting contents (from file load, scratch
         // initial text, etc.) shouldn't be undoable from the user's
         // perspective; only post-construction edits are.
+        let (text_delta_batches, text_delta_capture_enabled, text_delta_subscription) =
+            Self::subscribe_text_deltas(&doc);
         let undo = Self::create_undo_manager(&doc);
         Ok(Self {
             doc,
+            text_delta_batches,
+            text_delta_capture_enabled,
+            _text_delta_subscription: text_delta_subscription,
             undo: std::cell::RefCell::new(undo),
         })
     }
@@ -302,6 +356,41 @@ impl CrdtState {
         self.doc.import(bytes).map(|_| ())
     }
 
+    /// Import remote updates and capture Loro's text projection deltas.
+    ///
+    /// The import callback runs synchronously before `doc.import`
+    /// returns. Buffer's hot path uses the captured single-insert shape
+    /// to update its rope projection without materializing the whole
+    /// document.
+    pub fn import_updates_with_text_deltas(&self, bytes: &[u8]) -> LoroResult<Vec<Vec<TextDelta>>> {
+        self.text_delta_batches
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clear();
+        self.text_delta_capture_enabled
+            .store(true, Ordering::Relaxed);
+        let import_result = self.doc.import(bytes).map(|_| ());
+        self.text_delta_capture_enabled
+            .store(false, Ordering::Relaxed);
+        import_result?;
+        let mut guard = self
+            .text_delta_batches
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        Ok(std::mem::take(&mut *guard))
+    }
+
+    /// Convert a Unicode scalar offset in the current text projection
+    /// to its UTF-8 byte offset.
+    #[must_use]
+    pub fn unicode_to_utf8_pos(&self, pos: usize) -> Option<usize> {
+        self.doc.get_text("body").convert_pos(
+            pos,
+            loro::cursor::PosType::Unicode,
+            loro::cursor::PosType::Bytes,
+        )
+    }
+
     /// T M10.10 post-audit-round-4 F26 — validate that importing
     /// the wire bytes `bytes` would attribute every new op to
     /// `expected_peer_id`.

src/syntax.rs

@@ -579,6 +579,16 @@ impl SyntaxRegistry {
         self.parse_jobs.borrow().len()
     }
 
+    /// True when a dispatched parse job for `buffer` has not yet been
+    /// installed or drained. The syntax Lua glue records jobs here at
+    /// dispatch time and removes them in `_install_settled`, so this
+    /// is the main-thread "parse in flight" bit for render producers
+    /// that need to avoid stale whole-file work while typing.
+    #[must_use]
+    pub fn has_pending_parse_job_for(&self, buffer: BufferId) -> bool {
+        self.parse_jobs.borrow().values().any(|&bid| bid == buffer)
+    }
+
     /// Lazy-compile and cache the bundled `highlights.scm` query for
     /// `lang_name`. Returns `None` if the language is unknown, the
     /// language entry has an empty query (no highlights shipped),
@@ -869,4 +879,21 @@ mod tests {
         // Pending list cleared on make_request.
         assert_eq!(handle.pending_edit_count(), 0);
     }
+
+    #[test]
+    fn registry_tracks_inflight_parse_jobs_by_buffer() {
+        let registry = SyntaxRegistry::new();
+        let a = BufferId::next();
+        let b = BufferId::next();
+
+        registry.record_parse_job(11, a);
+        registry.record_parse_job(12, b);
+
+        assert!(registry.has_pending_parse_job_for(a));
+        assert!(registry.has_pending_parse_job_for(b));
+
+        assert_eq!(registry.take_parse_job(11), Some(a));
+        assert!(!registry.has_pending_parse_job_for(a));
+        assert!(registry.has_pending_parse_job_for(b));
+    }
 }

src/text_view.rs

@@ -34,6 +34,10 @@ use crate::view::{DisplayCoord, View, Viewport};
 /// that is a multiple of this value.
 const TAB_WIDTH: u32 = 8;
 
+/// Line-prefix lengths up to this many bytes are decoded on the stack in
+/// [`TextView::pos_to_display`]; longer prefixes fall back to a heap buffer.
+const STACK_CAP: usize = 256;
+
 /// Display width of `ch` when drawn starting at column `current_col`.
 ///
 /// Tabs expand to the next [`TAB_WIDTH`]-aligned column, so they need the
@@ -180,13 +184,27 @@ impl View for TextView {
         if take == 0 {
             return Some(DisplayCoord::new(row_idx as u32, 0));
         }
-        let mut bytes = vec![0u8; take];
-        buf.snapshot_rope().slice(line_start, pos, &mut bytes);
-        // Drop trailing bytes that don't form a complete codepoint.
-        while !bytes.is_empty() && std::str::from_utf8(&bytes).is_err() {
-            bytes.pop();
-        }
-        let s = std::str::from_utf8(&bytes).unwrap_or("");
+        // Copy [line_start, pos) into a stack buffer for the common short-line
+        // case, hitting the heap only for unusually long prefixes. This removes
+        // the per-call allocation that previously ran on every cursor move.
+        let mut stack_buf = [0u8; STACK_CAP];
+        let mut heap_buf: Vec<u8>;
+        let bytes: &mut [u8] = if take <= STACK_CAP {
+            &mut stack_buf[..take]
+        } else {
+            heap_buf = vec![0u8; take];
+            &mut heap_buf
+        };
+        buf.snapshot_rope().slice(line_start, pos, bytes);
+        // If `pos` fell inside a multi-byte codepoint, keep only the bytes up to
+        // the last complete codepoint. `valid_up_to()` gives that boundary in
+        // one step, replacing the old pop-one-byte-and-revalidate loop. (Only
+        // trailing bytes can be invalid here, since the slice is a prefix of
+        // valid UTF-8 cut at `pos`.)
+        let s = match std::str::from_utf8(bytes) {
+            Ok(valid) => valid,
+            Err(e) => std::str::from_utf8(&bytes[..e.valid_up_to()]).unwrap(),
+        };
         let mut col: u32 = 0;
         for ch in s.chars() {
             col += char_display_width(ch, col);