feat: add asynchronous LSP plugin diagnostics and definitions

src/session.rs

@@ -5,15 +5,38 @@ use std::{
 };
 
 use crate::{
-    editor::{Editor, Input, Invocation},
+    editor::{Editor, Effect, Input, Invocation},
+    plugin::{
+        DocumentEvent, DocumentEventKind, DocumentSnapshot, PluginEvent, PluginUpdate,
+        TextPosition, UpdateEffect,
+    },
     ruby::{Config, RubyPlugin, load_config},
 };
 
+struct TrackedDocument {
+    snapshot: DocumentSnapshot,
+    save_revision: u64,
+}
+
+#[derive(Clone, Copy)]
+struct InvocationSource {
+    buffer_id: usize,
+    revision: u64,
+}
+
+#[derive(Default)]
+struct AsyncRequests {
+    pending: BTreeMap<u64, InvocationSource>,
+    latest: Option<(u64, InvocationSource)>,
+}
+
 /// Owns command eligibility and execution hosts together. Frontends can inspect
 /// the editor, but cannot independently mutate its plugin registry.
 pub struct EditorSession {
     editor: Editor,
     plugins: BTreeMap<String, RubyPlugin>,
+    documents: BTreeMap<usize, TrackedDocument>,
+    requests: BTreeMap<String, AsyncRequests>,
 }
 
 impl EditorSession {
@@ -61,8 +84,9 @@ impl EditorSession {
             ));
         }
         let mut plugins = BTreeMap::new();
-        for path in config.plugins {
-            match RubyPlugin::start(&path) {
+        for configured in config.plugins {
+            let path = configured.path;
+            match RubyPlugin::start(&path, &configured.options) {
                 Ok((plugin, feature)) => {
                     let name = feature.name.clone();
                     match editor.register(feature) {
@@ -88,7 +112,14 @@ impl EditorSession {
         } else {
             editor.set_message(notices.join(" | "));
         }
-        Ok(Self { editor, plugins })
+        let mut session = Self {
+            editor,
+            plugins,
+            documents: BTreeMap::new(),
+            requests: BTreeMap::new(),
+        };
+        session.synchronize_documents();
+        Ok(session)
     }
 
     pub fn editor(&self) -> &Editor {
@@ -96,34 +127,326 @@ impl EditorSession {
     }
 
     pub fn handle(&mut self, input: Input) {
-        if let Some(invocation) = self.editor.handle(input) {
+        let invocation = self.editor.handle(input);
+        self.synchronize_documents();
+        if let Some(invocation) = invocation {
             self.dispatch(invocation);
+            self.synchronize_documents();
         }
     }
 
     pub fn execute(&mut self, command: &str) {
-        if let Some(invocation) = self.editor.execute(command) {
+        let invocation = self.editor.execute(command);
+        self.synchronize_documents();
+        if let Some(invocation) = invocation {
             self.dispatch(invocation);
+            self.synchronize_documents();
         }
     }
 
     fn dispatch(&mut self, invocation: Invocation) {
+        let context = self.editor.context();
         let result = match self.plugins.get_mut(&invocation.plugin) {
+            Some(plugin) if plugin.is_async() => {
+                let requests = self.requests.entry(invocation.plugin.clone()).or_default();
+                plugin
+                    .queue_invoke(&invocation.command, &invocation.args, &context)
+                    .map(|id| {
+                        let source = InvocationSource {
+                            buffer_id: context.buffer_id,
+                            revision: context.revision,
+                        };
+                        requests.pending.insert(id, source);
+                        requests.latest = Some((id, source));
+                    })
+            }
             Some(plugin) => plugin
-                .invoke(
-                    &invocation.command,
-                    &invocation.args,
-                    &self.editor.context(),
-                )
+                .invoke(&invocation.command, &invocation.args, &context)
                 .and_then(|effects| self.editor.apply(effects)),
             None => Err("plugin host is unavailable".into()),
         };
         if let Err(error) = result {
-            // Drop also stops a live host whose protocol-valid edits were rejected.
-            drop(self.plugins.remove(&invocation.plugin));
-            self.editor.disable_plugin(&invocation.plugin, &error);
+            self.retire(&invocation.plugin, &error);
         }
     }
+
+    fn retire(&mut self, owner: &str, error: &str) {
+        // Drop also stops a live host whose protocol-valid effects were rejected.
+        drop(self.plugins.remove(owner));
+        self.requests.remove(owner);
+        self.editor.disable_plugin(owner, error);
+    }
+
+    fn synchronize_documents(&mut self) -> bool {
+        if !self.plugins.values().any(RubyPlugin::is_async) {
+            self.documents.clear();
+            return false;
+        }
+        let mut events = Vec::new();
+        if !self.editor.should_quit() {
+            for document in self.editor.documents() {
+                let path = document.path.map(Path::to_string_lossy);
+                let previous = self.documents.get(&document.id);
+                let renamed = previous
+                    .is_some_and(|previous| previous.snapshot.path.as_deref() != path.as_deref());
+                if renamed {
+                    events.push(DocumentEvent {
+                        event: DocumentEventKind::Close,
+                        document: previous.expect("renamed document existed").snapshot.clone(),
+                    });
+                }
+                let changed = previous.is_none_or(|previous| {
+                    previous.snapshot.revision != document.revision || renamed
+                });
+                if changed {
+                    let snapshot = DocumentSnapshot {
+                        id: document.id,
+                        revision: document.revision,
+                        path: path.map(|path| path.into_owned()),
+                        root: self.editor.project_root().to_string_lossy().into_owned(),
+                        text: document.lines.join("\n"),
+                    };
+                    events.push(DocumentEvent {
+                        event: if previous.is_none() || renamed {
+                            DocumentEventKind::Open
+                        } else {
+                            DocumentEventKind::Change
+                        },
+                        document: snapshot.clone(),
+                    });
+                    let save_revision = previous.map_or(0, |previous| previous.save_revision);
+                    self.documents.insert(
+                        document.id,
+                        TrackedDocument {
+                            snapshot,
+                            save_revision,
+                        },
+                    );
+                }
+                let tracked = self
+                    .documents
+                    .get_mut(&document.id)
+                    .expect("document tracked");
+                if tracked.save_revision != document.save_revision {
+                    events.push(DocumentEvent {
+                        event: DocumentEventKind::Save,
+                        document: tracked.snapshot.clone(),
+                    });
+                    tracked.save_revision = document.save_revision;
+                }
+            }
+        }
+        let closed: Vec<_> = self
+            .documents
+            .keys()
+            .filter(|id| self.editor.should_quit() || !self.editor.has_buffer(**id))
+            .copied()
+            .collect();
+        for id in closed {
+            let document = self.documents.remove(&id).expect("closed document existed");
+            events.push(DocumentEvent {
+                event: DocumentEventKind::Close,
+                document: document.snapshot,
+            });
+        }
+        let mut failures = Vec::new();
+        for (owner, plugin) in &mut self.plugins {
+            if !plugin.is_async() {
+                continue;
+            }
+            for event in &events {
+                if let Err(error) = plugin.notify_document(event) {
+                    failures.push((owner.clone(), error));
+                    break;
+                }
+            }
+        }
+        let changed = !failures.is_empty();
+        for (owner, error) in failures {
+            self.retire(&owner, &error);
+        }
+        changed
+    }
+
+    fn source_is_current(&self, source: InvocationSource) -> bool {
+        self.editor
+            .documents()
+            .any(|document| document.id == source.buffer_id && document.revision == source.revision)
+    }
+
+    fn navigation_is_current(
+        &self,
+        owner: &str,
+        request_id: Option<u64>,
+        source: InvocationSource,
+    ) -> bool {
+        self.editor.buffer_id() == source.buffer_id
+            && self.source_is_current(source)
+            && self
+                .requests
+                .get(owner)
+                .and_then(|requests| requests.latest)
+                .is_some_and(|(id, latest)| {
+                    request_id == Some(id)
+                        && latest.buffer_id == source.buffer_id
+                        && latest.revision == source.revision
+                })
+    }
+
+    fn apply_update(&mut self, owner: &str, update: PluginUpdate) -> Result<bool, String> {
+        // Validate the whole applicable annotation batch before changing any buffer.
+        for effect in &update.effects {
+            match effect {
+                UpdateEffect::Diagnostics {
+                    buffer_id,
+                    revision,
+                    items,
+                } => {
+                    self.editor
+                        .validate_diagnostics(*buffer_id, *revision, items)?;
+                }
+                UpdateEffect::Navigate {
+                    buffer_id,
+                    revision,
+                    path,
+                    ..
+                } => {
+                    let source = InvocationSource {
+                        buffer_id: *buffer_id,
+                        revision: *revision,
+                    };
+                    if self.navigation_is_current(owner, update.request_id, source)
+                        && (path.as_os_str().is_empty()
+                            || path.to_string_lossy().chars().any(char::is_control))
+                    {
+                        return Err(
+                            "Navigation path must be nonempty and contain no controls".into()
+                        );
+                    }
+                }
+                UpdateEffect::Message { .. } => {}
+            }
+        }
+        let mut changed = false;
+        for effect in update.effects {
+            match effect {
+                UpdateEffect::Diagnostics {
+                    buffer_id,
+                    revision,
+                    items,
+                } => {
+                    changed |= self
+                        .editor
+                        .replace_validated_diagnostics(owner, buffer_id, revision, items);
+                }
+                UpdateEffect::Navigate {
+                    buffer_id,
+                    revision,
+                    path,
+                    row,
+                    column,
+                } => {
+                    let source = InvocationSource {
+                        buffer_id,
+                        revision,
+                    };
+                    if self.navigation_is_current(owner, update.request_id, source) {
+                        if let Err(error) =
+                            self.editor.navigate_to(path, TextPosition { row, column })
+                        {
+                            self.editor
+                                .set_message(format!("Navigation failed: {error}"));
+                        }
+                        changed = true;
+                    }
+                }
+                UpdateEffect::Message { text } => {
+                    changed |= self.editor.message() != text;
+                    self.editor.set_message(text);
+                }
+            }
+        }
+        Ok(changed)
+    }
+
+    fn apply_response(
+        &mut self,
+        owner: &str,
+        id: u64,
+        effects: Vec<Effect>,
+    ) -> Result<bool, String> {
+        let source = self
+            .requests
+            .get_mut(owner)
+            .and_then(|requests| requests.pending.remove(&id));
+        let Some(source) = source else {
+            return Err("plugin response has an unknown invocation ID".into());
+        };
+        if !self.source_is_current(source) || self.editor.buffer_id() != source.buffer_id {
+            return Ok(false);
+        }
+        if effects
+            .iter()
+            .all(|effect| matches!(effect, Effect::Message(_)))
+        {
+            let mut changed = false;
+            for effect in effects {
+                if let Effect::Message(text) = effect {
+                    changed |= self.editor.message() != text;
+                    self.editor.set_message(text);
+                }
+            }
+            return Ok(changed);
+        }
+        let changed = !effects.is_empty();
+        self.editor.apply(effects)?;
+        Ok(changed)
+    }
+
+    /// Drains ready plugin work without waiting for command or server responses.
+    pub fn poll(&mut self) -> bool {
+        let mut changed = self.synchronize_documents();
+        let mut ready = Vec::new();
+        for (owner, plugin) in &mut self.plugins {
+            if plugin.is_async() {
+                let result = plugin.poll();
+                if match &result {
+                    Ok(events) => !events.is_empty(),
+                    Err(_) => true,
+                } {
+                    ready.push((owner.clone(), result));
+                }
+            }
+        }
+        for (owner, events) in ready {
+            let events = match events {
+                Ok(events) => events,
+                Err(error) => {
+                    self.retire(&owner, &error);
+                    changed = true;
+                    continue;
+                }
+            };
+            for event in events {
+                let result = match event {
+                    PluginEvent::Response { id, effects } => {
+                        self.apply_response(&owner, id, effects)
+                    }
+                    PluginEvent::Update(update) => self.apply_update(&owner, update),
+                };
+                match result {
+                    Ok(applied) => changed |= applied,
+                    Err(error) => {
+                        self.retire(&owner, &error);
+                        changed = true;
+                        break;
+                    }
+                }
+            }
+        }
+        changed |= self.synchronize_documents();
+        changed
+    }
 }
 
 #[cfg(test)]
@@ -200,6 +523,65 @@ Vamp.command("healthy") { |editor, args| editor.replace "healthy #{args}" }
 "#,
             );
         }
+        fn asynchronous(&self, file: Option<PathBuf>) -> EditorSession {
+            self.file(
+                "generic.rb",
+                r#"
+Vamp.plugin "generic_fixture", requires: ["text", "diagnostics"], async: true
+log_mutex = Mutex.new
+record = ->(value) { log_mutex.synchronize { File.open(Vamp.options.fetch("log"), "a") { |file| file.puts JSON.generate(value) } } }
+Vamp.on_document { |event| record.call(event) }
+Vamp.command("fixture_barrier") { |editor, args| editor.message args }
+Vamp.command("fixture_replace") { |editor, args| editor.replace args }
+Vamp.command("fixture_crash") { |_editor, _args| Process.exit!(7) }
+["fixture_publish", "fixture_hold"].each do |name|
+  Vamp.command(name) do |editor, args|
+    payload = JSON.parse(args)
+    effects = payload.fetch("effects")
+    effects.each do |effect|
+      next unless ["diagnostics", "navigate"].include?(effect["type"])
+      effect["buffer_id"] ||= editor.buffer_id
+      effect["revision"] ||= editor.revision
+    end
+    record.call({"invoke" => name, "buffer_id" => editor.buffer_id, "revision" => editor.revision,
+                 "row" => editor.row, "column" => editor.column, "request_id" => editor.request_id})
+    if name == "fixture_hold"
+      Thread.new do
+        gate = payload.fetch("gate")
+        sleep 0.002 until File.exist?(gate)
+        Vamp.publish(effects + [{"type" => "message", "text" => "delivered #{gate}"}], request_id: editor.request_id)
+      end
+    else
+      Vamp.publish(effects, request_id: editor.request_id)
+    end
+  end
+end
+"#,
+            );
+            let log = self.0.join("events.jsonl");
+            self.file("events.jsonl", "");
+            let config = self.file(
+                "async-config.rb",
+                &format!(
+                    "Vamp.configure {{ |config| config.clipboard_mode = \"separate\"; config.plugin \"generic.rb\", options: {{log: {}}} }}",
+                    serde_json::to_string(&log.to_string_lossy()).unwrap(),
+                ),
+            );
+            EditorSession::open(
+                file,
+                vec!["text".into(), "diagnostics".into()],
+                Some(&config),
+            )
+            .unwrap()
+        }
+
+        fn events(&self) -> Vec<serde_json::Value> {
+            fs::read_to_string(self.0.join("events.jsonl"))
+                .unwrap()
+                .lines()
+                .map(|line| serde_json::from_str(line).unwrap())
+                .collect()
+        }
     }
 
     impl Drop for Fixture {
@@ -1842,4 +2224,442 @@ end
             "x"
         );
     }
+
+    fn drain_until(session: &mut EditorSession, ready: impl Fn(&EditorSession) -> bool) {
+        // Process-I/O timeout, not a date-dependent assertion.
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
+        loop {
+            session.poll();
+            if ready(session) {
+                return;
+            }
+            assert!(
+                std::time::Instant::now() < deadline,
+                "plugin did not settle: {}",
+                session.editor().message()
+            );
+            std::thread::sleep(std::time::Duration::from_millis(2));
+        }
+    }
+
+    fn barrier(session: &mut EditorSession, message: &str) {
+        session.execute(&format!("fixture_barrier {message}"));
+        drain_until(session, |session| session.editor().message() == message);
+    }
+
+    fn diagnostic_effect(id: usize, revision: u64, message: &str) -> serde_json::Value {
+        serde_json::json!({
+            "type": "diagnostics", "buffer_id": id, "revision": revision,
+            "items": [{"start": {"row": 0, "column": 0}, "end": {"row": 0, "column": 1},
+                       "severity": "warning", "message": message}]
+        })
+    }
+
+    fn publish(session: &mut EditorSession, effects: Vec<serde_json::Value>) {
+        session.execute(&format!(
+            "fixture_publish {}",
+            serde_json::json!({"effects": effects})
+        ));
+    }
+
+    fn hold(session: &mut EditorSession, gate: &Path, effects: Vec<serde_json::Value>) {
+        session.execute(&format!(
+            "fixture_hold {}",
+            serde_json::json!({"gate": gate, "effects": effects})
+        ));
+    }
+
+    fn release(session: &mut EditorSession, gate: &Path) {
+        fs::write(gate, "").unwrap();
+        let delivered = format!("delivered {}", gate.display());
+        drain_until(session, |session| session.editor().message() == delivered);
+    }
+
+    #[test]
+    fn async_documents_include_unnamed_hidden_versions_real_saves_renames_and_closes() {
+        let fixture = Fixture::new();
+        let mut session = fixture.asynchronous(None);
+        barrier(&mut session, "initial");
+        let initial = fixture.events();
+        assert_eq!(initial[0]["event"], "open");
+        assert_eq!(initial[0]["document"]["id"], 1);
+        assert_eq!(initial[0]["document"]["revision"], 0);
+        assert_eq!(initial[0]["document"]["path"], serde_json::Value::Null);
+        assert_eq!(initial[0]["document"]["text"], "");
+        assert_eq!(
+            initial[0]["document"]["root"],
+            session.editor().project_root().to_string_lossy().as_ref()
+        );
+        session.handle(Input::Char('i'));
+        for input in [
+            Input::Char('😀'),
+            Input::Char('e'),
+            Input::Char('\u{301}'),
+            Input::Enter,
+        ] {
+            session.handle(input);
+        }
+        session.handle(Input::Escape);
+        let revision = session.editor().text_revision();
+        session.handle(Input::Char('h'));
+        session.handle(Input::Char('l'));
+        let renamed = fixture.0.join("saved unicode 😀.txt");
+        session.execute(&format!("w {}", renamed.display()));
+        session.execute("w");
+        session.execute(&format!("w {}", fixture.0.join("missing/failed").display()));
+        barrier(&mut session, "saved");
+        assert_eq!(session.editor().text_revision(), revision);
+        assert_eq!(
+            session.editor().documents().next().unwrap().save_revision,
+            2
+        );
+        assert_eq!(fs::read_to_string(&renamed).unwrap(), "😀e\u{301}\n");
+        let events = fixture.events();
+        assert_eq!(
+            events
+                .iter()
+                .filter(|event| event["event"] == "change")
+                .count(),
+            revision as usize
+        );
+        let kinds: Vec<_> = events
+            .iter()
+            .filter_map(|event| event["event"].as_str())
+            .collect();
+        assert_eq!(&kinds[kinds.len() - 4..], ["close", "open", "save", "save"]);
+        let reopened = events
+            .iter()
+            .find(|event| event["event"] == "open" && !event["document"]["path"].is_null())
+            .unwrap();
+        assert_eq!(reopened["document"]["revision"], revision);
+        assert_eq!(reopened["document"]["text"], "😀e\u{301}\n");
+        let target = fixture.file("target.txt", "target");
+        session.execute(&format!("edit {}", target.display()));
+        session.handle(Input::Char('x'));
+        session.execute("buffer 1");
+        let documents: Vec<_> = session.editor().documents().collect();
+        assert_eq!(
+            documents
+                .iter()
+                .map(|document| document.id)
+                .collect::<Vec<_>>(),
+            [1, 2]
+        );
+        assert_eq!(documents[1].lines, ["arget"]);
+        assert_eq!(documents[1].revision, 1);
+        assert_eq!(documents[1].save_revision, 0);
+        session.execute("bdelete! 2");
+        barrier(&mut session, "closed");
+        let events = fixture.events();
+        let closed = events
+            .iter()
+            .find(|event| event["event"] == "close" && event["document"]["id"] == 2)
+            .unwrap();
+        assert_eq!(closed["document"]["text"], "arget");
+        assert_eq!(closed["document"]["revision"], 1);
+        session.execute("q!");
+        barrier(&mut session, "quit flushed");
+        assert!(session.editor().should_quit());
+        let events = fixture.events();
+        let final_close = events
+            .iter()
+            .rev()
+            .find(|event| event["event"] == "close")
+            .unwrap();
+        assert_eq!(final_close["document"]["id"], 1);
+        assert_eq!(
+            final_close["document"]["path"],
+            renamed.to_string_lossy().as_ref()
+        );
+    }
+
+    #[test]
+    fn async_annotations_and_messages_preserve_insert_group_and_ignore_stale_closed_sources() {
+        let fixture = Fixture::new();
+        let file = fixture.file("buffer.txt", "abc");
+        let mut session = fixture.asynchronous(Some(file));
+        let gate = fixture.0.join("annotations-ready");
+        hold(&mut session, &gate, vec![diagnostic_effect(1, 1, "live")]);
+        session.handle(Input::Char('i'));
+        session.handle(Input::Char('x'));
+        release(&mut session, &gate);
+        assert_eq!(session.editor().mode(), crate::editor::Mode::Insert);
+        assert_eq!(
+            session
+                .editor()
+                .diagnostic_at_cursor()
+                .map(|diagnostic| diagnostic.message.as_str()),
+            None
+        );
+        assert_eq!(session.editor().diagnostics().count(), 1);
+        session.handle(Input::Char('y'));
+        assert_eq!(session.editor().diagnostics().count(), 0);
+        session.handle(Input::Escape);
+        session.handle(Input::Char('u'));
+        assert_eq!(session.editor().lines(), ["abc"]);
+        let current = session.editor().text_revision();
+        let mut stale = diagnostic_effect(1, 0, "stale");
+        stale["items"][0]["end"]["row"] = serde_json::json!(999);
+        publish(
+            &mut session,
+            vec![
+                stale,
+                diagnostic_effect(999, current, "closed"),
+                serde_json::json!({"type":"message","text":"stale checked"}),
+            ],
+        );
+        drain_until(&mut session, |session| {
+            session.editor().message() == "stale checked"
+        });
+        assert_eq!(session.editor().diagnostics().count(), 0);
+        assert!(session.plugins.contains_key("generic_fixture"));
+        let target = fixture.file("closed.txt", "closed");
+        session.execute(&format!("edit {}", target.display()));
+        let gate = fixture.0.join("closed-ready");
+        hold(
+            &mut session,
+            &gate,
+            vec![diagnostic_effect(2, 0, "late close")],
+        );
+        session.execute("bdelete 2");
+        release(&mut session, &gate);
+        assert_eq!(session.editor().buffer_id(), 1);
+        assert_eq!(session.editor().diagnostics().count(), 0);
+        assert!(session.plugins.contains_key("generic_fixture"));
+    }
+
+    #[test]
+    fn invalid_annotation_batches_retire_owner_and_clear_hidden_annotations_only() {
+        let fixture = Fixture::new();
+        let file = fixture.file("buffer.txt", "source");
+        let target = fixture.file("target.txt", "target");
+        let mut session = fixture.asynchronous(Some(file));
+        session.execute(&format!("edit {}", target.display()));
+        session.execute("buffer 1");
+        publish(
+            &mut session,
+            vec![
+                diagnostic_effect(1, 0, "source"),
+                diagnostic_effect(2, 0, "hidden"),
+            ],
+        );
+        drain_until(&mut session, |session| {
+            session.editor().diagnostics().count() == 1
+        });
+        let other = crate::plugin::Diagnostic {
+            start: TextPosition { row: 0, column: 0 },
+            end: TextPosition { row: 0, column: 1 },
+            severity: crate::plugin::DiagnosticSeverity::Hint,
+            message: "other owner".into(),
+        };
+        session
+            .editor
+            .replace_diagnostics("other", 1, 0, vec![other])
+            .unwrap();
+        let mut invalid = diagnostic_effect(1, 0, "bad range");
+        invalid["items"][0]["end"]["column"] = serde_json::json!(999);
+        publish(
+            &mut session,
+            vec![diagnostic_effect(2, 0, "must not commit"), invalid],
+        );
+        drain_until(&mut session, |session| {
+            !session.plugins.contains_key("generic_fixture")
+        });
+        assert!(
+            session
+                .editor()
+                .message()
+                .contains("Diagnostic position is outside")
+        );
+        assert_eq!(session.editor().diagnostics().count(), 1);
+        assert_eq!(
+            session.editor().diagnostic_at_cursor().unwrap().message,
+            "other owner"
+        );
+        session.execute("buffer 2");
+        assert_eq!(session.editor().diagnostics().count(), 0);
+        session.execute("fixture_publish {}");
+        assert!(session.editor().message().contains("Unknown command"));
+        session.handle(Input::Char('x'));
+        assert_eq!(session.editor().lines(), ["arget"]);
+        session.handle(Input::Char('u'));
+        assert_eq!(session.editor().lines(), ["target"]);
+    }
+
+    fn navigation_effect(path: &Path) -> serde_json::Value {
+        serde_json::json!({"type":"navigate", "path":path, "row":0, "column":1})
+    }
+
+    #[test]
+    fn async_navigation_reuses_dirty_target_and_reports_open_failure_without_retiring() {
+        let fixture = Fixture::new();
+        let source = fixture.file("buffer.txt", "source");
+        let target = fixture.file("dirty 😀 target.txt", "a😀e\u{301}");
+        let mut session = fixture.asynchronous(Some(source));
+        session.execute(&format!("edit {}", target.display()));
+        session.handle(Input::Char('i'));
+        session.handle(Input::Char('!'));
+        session.handle(Input::Escape);
+        session.execute("buffer 1");
+        publish(&mut session, vec![navigation_effect(&target)]);
+        drain_until(&mut session, |session| session.editor().buffer_id() == 2);
+        assert_eq!(session.editor().lines(), ["!a😀e\u{301}"]);
+        assert_eq!(
+            session.editor().cursor(),
+            crate::editor::Cursor { row: 0, column: 1 }
+        );
+        assert_eq!(session.editor().mode(), crate::editor::Mode::Normal);
+        assert_eq!(session.editor().buffer_count(), 2);
+        assert_eq!(fs::read_to_string(&target).unwrap(), "a😀e\u{301}");
+        let gate = fixture.0.join("missing-ready");
+        hold(
+            &mut session,
+            &gate,
+            vec![navigation_effect(&fixture.0.join("missing"))],
+        );
+        fs::write(&gate, "").unwrap();
+        drain_until(&mut session, |session| {
+            session
+                .requests
+                .get("generic_fixture")
+                .is_some_and(|requests| requests.pending.is_empty())
+                && session.editor().message().starts_with("delivered ")
+        });
+        assert_eq!(session.editor().buffer_id(), 2);
+        assert_eq!(session.editor().lines(), ["!a😀e\u{301}"]);
+        assert!(session.plugins.contains_key("generic_fixture"));
+        // A standalone failure message remains visible when no later plugin message overwrites it.
+        publish(
+            &mut session,
+            vec![navigation_effect(&fixture.0.join("missing"))],
+        );
+        drain_until(&mut session, |session| {
+            session.editor().message().starts_with("Navigation failed:")
+        });
+        assert_eq!(session.editor().buffer_id(), 2);
+        assert_eq!(session.editor().buffer_count(), 2);
+        assert!(session.plugins.contains_key("generic_fixture"));
+    }
+
+    #[test]
+    fn async_navigation_drops_superseded_changed_or_inactive_source_replies() {
+        for guard in ["superseded", "changed", "inactive", "closed"] {
+            let fixture = Fixture::new();
+            let source = fixture.file("buffer.txt", "source");
+            let target = fixture.file("target.txt", "target");
+            let mut session = fixture.asynchronous(Some(source));
+            let gate = fixture.0.join("navigation-ready");
+            hold(&mut session, &gate, vec![navigation_effect(&target)]);
+            match guard {
+                "superseded" => barrier(&mut session, "new request"),
+                "changed" => {
+                    session.handle(Input::Char('i'));
+                    session.handle(Input::Char('x'));
+                }
+                "inactive" => {
+                    let other = fixture.file("other.txt", "other");
+                    session.execute(&format!("edit {}", other.display()));
+                }
+                "closed" => session.execute("bdelete 1"),
+                _ => unreachable!(),
+            }
+            let active = session.editor().buffer_id();
+            let text = session.editor().context().text;
+            release(&mut session, &gate);
+            assert_eq!(session.editor().buffer_id(), active, "{guard}");
+            assert_eq!(session.editor().context().text, text, "{guard}");
+            assert!(session.plugins.contains_key("generic_fixture"), "{guard}");
+            assert!(
+                !session
+                    .editor()
+                    .buffers()
+                    .any(|buffer| buffer.path == Some(target.as_path())),
+                "{guard}"
+            );
+            let invokes: Vec<_> = fixture
+                .events()
+                .into_iter()
+                .filter(|event| event["invoke"] == "fixture_hold")
+                .collect();
+            assert_eq!(invokes.len(), 1);
+            assert_eq!(invokes[0]["buffer_id"], 1);
+            assert_eq!(invokes[0]["revision"], 0);
+            assert_eq!(invokes[0]["row"], 1);
+            assert_eq!(invokes[0]["column"], 1);
+        }
+    }
+
+    #[test]
+    fn stale_async_text_response_and_dead_host_never_break_native_typing() {
+        let fixture = Fixture::new();
+        let file = fixture.file("buffer.txt", "source");
+        let mut session = fixture.asynchronous(Some(file));
+        session.execute("fixture_replace BAD");
+        session.handle(Input::Char('i'));
+        session.handle(Input::Char('x'));
+        drain_until(&mut session, |session| {
+            session
+                .requests
+                .get("generic_fixture")
+                .is_some_and(|requests| requests.pending.is_empty())
+        });
+        assert_eq!(session.editor().lines(), ["xsource"]);
+        assert_eq!(session.editor().mode(), crate::editor::Mode::Insert);
+        session.handle(Input::Char('y'));
+        session.handle(Input::Escape);
+        session.handle(Input::Char('u'));
+        assert_eq!(session.editor().lines(), ["source"]);
+        let revision = session.editor().text_revision();
+        publish(
+            &mut session,
+            vec![diagnostic_effect(1, revision, "before crash")],
+        );
+        drain_until(&mut session, |session| {
+            session.editor().diagnostics().count() == 1
+        });
+        session.execute("fixture_crash");
+        session.handle(Input::Char('i'));
+        session.handle(Input::Char('z'));
+        drain_until(&mut session, |session| {
+            !session.plugins.contains_key("generic_fixture")
+        });
+        assert_eq!(session.editor().diagnostics().count(), 0);
+        assert_eq!(session.editor().mode(), crate::editor::Mode::Insert);
+        session.handle(Input::Char('q'));
+        session.handle(Input::Escape);
+        session.handle(Input::Char('u'));
+        assert_eq!(session.editor().lines(), ["source"]);
+        assert!(!session.editor().should_quit());
+        session.execute("plugins");
+        assert!(session.editor().message().contains("core [core]"));
+    }
+
+    #[test]
+    fn async_navigation_invokes_once_at_primary_with_multiple_carets() {
+        let fixture = Fixture::new();
+        let source = fixture.file("buffer.txt", "abc\nabc");
+        let target = fixture.file("target.txt", "target");
+        let mut session = fixture.asynchronous(Some(source));
+        session.handle(Input::Char('l'));
+        session.handle(Input::Ctrl('e'));
+        assert_eq!(session.editor().selection_count(), 2);
+        assert_eq!(
+            session.editor().cursor(),
+            crate::editor::Cursor { row: 1, column: 1 }
+        );
+        let gate = fixture.0.join("primary-ready");
+        hold(&mut session, &gate, vec![navigation_effect(&target)]);
+        release(&mut session, &gate);
+        assert_eq!(session.editor().path(), Some(target.as_path()));
+        assert_eq!(session.editor().selection_count(), 1);
+        let invokes: Vec<_> = fixture
+            .events()
+            .into_iter()
+            .filter(|event| event["invoke"] == "fixture_hold")
+            .collect();
+        assert_eq!(invokes.len(), 1);
+        assert_eq!(invokes[0]["row"], 2);
+        assert_eq!(invokes[0]["column"], 2);
+        assert_eq!(invokes[0]["revision"], 0);
+    }
 }