use std::{ collections::BTreeMap, io, path::{Path, PathBuf}, sync::{ Arc, atomic::{AtomicBool, AtomicU64, Ordering}, mpsc::{self, Receiver, Sender}, }, thread::{self, JoinHandle}, time::{Duration, Instant}, }; use crate::{ blame::BlameWorker, diff::{Focus, Operation as DiffOperation, Outcome as DiffOutcome}, editor::{ BlameSource, Cursor, Editor, Effect, Input, Invocation, Mode, ResolverOutcome, ResolverRequest, }, lsp::Lsp, plugin::{ DocumentEvent, DocumentEventKind, DocumentSnapshot, PluginEvent, PluginUpdate, TextPosition, UpdateEffect, }, repository::{CheckoutRefresh, Repository}, ruby::{Config, RubyPlugin, load_config}, }; type DiffResult = (u64, Result); struct ResolverWorker { requests: Option>, results: Receiver, stopped: Arc, thread: Option>, } impl ResolverWorker { fn start() -> Result { let (requests, incoming) = mpsc::channel::(); let (outgoing, results) = mpsc::channel(); let stopped = Arc::new(AtomicBool::new(false)); let stop = Arc::clone(&stopped); let thread = thread::Builder::new() .name("vamp-resolve".into()) .spawn(move || { while let Ok(request) = incoming.recv() { if stop.load(Ordering::Acquire) { break; } let cancelled = || stop.load(Ordering::Acquire); let outcome = match request { ResolverRequest::Load { tab, root } => ResolverOutcome::Loaded { tab, result: crate::resolve::load(&root, &cancelled), }, ResolverRequest::Save { tab, file, buffer, workspace, resolution, quit, } => ResolverOutcome::Saved { tab, file, buffer, quit, result: crate::resolve::save(&workspace, file, resolution, &cancelled), }, }; if outgoing.send(outcome).is_err() { break; } } }) .map_err(|error| format!("Cannot start resolver worker: {error}"))?; Ok(Self { requests: Some(requests), results, stopped, thread: Some(thread), }) } } impl Drop for ResolverWorker { fn drop(&mut self) { self.stopped.store(true, Ordering::Release); self.requests.take(); if let Some(thread) = self.thread.take() { let _ = thread.join(); } } } #[derive(Clone, Copy)] struct PullRequestJob { page: u32, busy: bool, complete: bool, failed: bool, retry: u64, } struct PullRequestWorker { id: u64, requests: Vec>, threads: Vec>, cancelled: Arc, jobs: [PullRequestJob; 2], priority: usize, } impl PullRequestWorker { fn start( id: u64, root: &Path, remote: String, outgoing: Sender, ) -> Result { let cancelled = Arc::new(AtomicBool::new(false)); let fetches = std::array::from_fn(|index| { let root = root.to_path_buf(); let remote = remote.clone(); let cancelled = Arc::clone(&cancelled); let mut repository = None; move |page| { if repository.is_none() { repository = Some(Repository::open(&root, Arc::clone(&cancelled))?); } repository .as_mut() .expect("repository opened") .pull_request_page(&remote, index == 1, page) } }); Self::start_with(id, outgoing, cancelled, fetches) } fn start_with( id: u64, outgoing: Sender, cancelled: Arc, fetches: [impl FnMut(u32) -> Result + Send + 'static; 2], ) -> Result { let mut worker = Self { id, requests: Vec::with_capacity(2), threads: Vec::with_capacity(2), cancelled, jobs: [PullRequestJob { page: 1, busy: false, complete: false, failed: false, retry: 0, }; 2], priority: 0, }; for (index, mut fetch) in fetches.into_iter().enumerate() { let (requests, incoming) = mpsc::channel(); let outgoing = outgoing.clone(); let cancelled = Arc::clone(&worker.cancelled); let thread = thread::Builder::new() .name(format!("vamp-pull-requests-{index}")) .spawn(move || { while let Ok(page) = incoming.recv() { if cancelled.load(Ordering::Acquire) { break; } let result = fetch(page); if cancelled.load(Ordering::Acquire) || outgoing .send(( id, Ok(DiffOutcome::PullRequestPage { closed: index == 1, result, }), )) .is_err() { break; } } }) .map_err(|error| format!("Cannot start pull request worker: {error}"))?; worker.requests.push(requests); worker.threads.push(thread); } worker.schedule()?; Ok(worker) } fn update(&mut self, category: usize, retries: [u64; 2]) -> Result<(), String> { self.priority = usize::from(category == 1); for (job, retry) in self.jobs.iter_mut().zip(retries) { if job.retry != retry { job.retry = retry; job.failed = false; } } self.schedule() } fn complete(&mut self, closed: bool, result: &Result) { let job = &mut self.jobs[usize::from(closed)]; job.busy = false; match result { Ok(page) => { job.complete = !page.has_more; job.page += 1; } Err(_) => job.failed = true, } } fn schedule(&mut self) -> Result<(), String> { for index in [self.priority, 1 - self.priority] { let job = &mut self.jobs[index]; if !job.busy && !job.complete && !job.failed { self.requests[index] .send(job.page) .map_err(|_| "Pull request worker stopped unexpectedly")?; job.busy = true; } } Ok(()) } } impl Drop for PullRequestWorker { fn drop(&mut self) { self.cancelled.store(true, Ordering::Release); self.requests.clear(); for thread in self.threads.drain(..) { let _ = thread.join(); } } } struct DiffWorker { requests: Option>, results: Option>, cancelled: Arc, stopped: Arc, cancelled_request: Arc, thread: Option>, root: PathBuf, outgoing: Sender, pull_requests: Option, pending: Option, } impl DiffWorker { fn start(root: PathBuf) -> Result { let (requests, incoming) = mpsc::channel::<(u64, DiffOperation)>(); let (outgoing, results) = mpsc::channel(); let sender = outgoing.clone(); let worker_root = root.clone(); let cancelled = Arc::new(AtomicBool::new(false)); let stopped = Arc::new(AtomicBool::new(false)); let cancelled_request = Arc::new(AtomicU64::new(0)); let cancelled_id = Arc::clone(&cancelled_request); let stop = Arc::clone(&stopped); let cancellation = Arc::clone(&cancelled); let thread = thread::Builder::new() .name("vamp-diff".into()) .spawn(move || { let mut repository = None; while let Ok((id, operation)) = incoming.recv() { if stop.load(Ordering::Acquire) { break; } cancellation.store(false, Ordering::Release); if stop.load(Ordering::Acquire) { break; } if cancelled_id.load(Ordering::Acquire) >= id { continue; } let result = (|| { if repository.is_none() { repository = Some(Repository::open(&worker_root, Arc::clone(&cancellation))?); } let repository = repository.as_mut().expect("repository opened"); match operation { DiffOperation::Discover => { repository.catalog().map(DiffOutcome::Catalog) } DiffOperation::PullRequests(_) => { unreachable!("listing uses category workers") } DiffOperation::ResolvePullRequest(revision) => repository .resolve_pull_request(revision) .map(|(before, after)| DiffOutcome::ResolvedPullRequest { before, after, }), DiffOperation::ListRemote(remote) => { repository.list_remote(&remote).map(DiffOutcome::Catalog) } DiffOperation::FetchRemote(remote) => { repository.fetch_remote(&remote).map(DiffOutcome::Catalog) } DiffOperation::Compare { before, after, mode, buffers, } => repository .compare(id, &before, &after, mode, &buffers) .map(DiffOutcome::Comparison), DiffOperation::RefreshCheckout { comparison_id, follow_working, buffers, } => repository .refresh_checkout(comparison_id, id, follow_working, &buffers) .map(|refresh| match refresh { Some(CheckoutRefresh::Bookmark(comparison)) => { DiffOutcome::BookmarkRefresh(comparison) } Some(CheckoutRefresh::Checkout(comparison)) => { DiffOutcome::CheckoutRefresh(Some(comparison)) } None => DiffOutcome::CheckoutRefresh(None), }), DiffOperation::LoadFile { comparison_id, file_index, } => repository.load_file(comparison_id, file_index).map(|file| { DiffOutcome::File { comparison_id, file_index, file, } }), } })(); if stop.load(Ordering::Acquire) || outgoing.send((id, result)).is_err() { break; } } }) .map_err(|error| format!("Cannot start diff worker: {error}"))?; Ok(Self { requests: Some(requests), results: Some(results), cancelled, stopped, cancelled_request, thread: Some(thread), root, outgoing: sender, pull_requests: None, pending: None, }) } fn request(&mut self, id: u64, operation: DiffOperation) -> Result<(), String> { self.pull_requests = None; if let Some(previous) = self.pending.take() { self.cancelled_request.store(previous, Ordering::Release); self.cancelled.store(true, Ordering::Release); } if let DiffOperation::PullRequests(remote) = operation { self.pull_requests = Some(PullRequestWorker::start( id, &self.root, remote, self.outgoing.clone(), )?); self.pending = Some(id); return Ok(()); } self.requests .as_ref() .expect("live diff worker has a sender") .send((id, operation)) .map_err(|_| "Diff worker stopped unexpectedly".to_owned())?; self.pending = Some(id); Ok(()) } } impl Drop for DiffWorker { fn drop(&mut self) { self.stopped.store(true, Ordering::Release); self.pull_requests = None; self.cancelled.store(true, Ordering::Release); self.requests.take(); self.results.take(); if let Some(thread) = self.thread.take() { let _ = thread.join(); } } } struct TrackedDocument { snapshot: DocumentSnapshot, save_revision: u64, } #[derive(Clone, Copy)] struct InvocationSource { buffer_id: usize, pane_id: usize, revision: u64, } #[derive(Default)] struct AsyncRequests { pending: BTreeMap, latest: Option<(u64, InvocationSource)>, } struct GutterSource { buffer: usize, revision: u64, save_revision: u64, path: PathBuf, dirty: bool, contents: Arc, } /// 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, documents: BTreeMap, requests: BTreeMap, lsp: Lsp, lsp_request_id: u64, lsp_source_cursor: Option, lsp_source_mode: Mode, lsp_source_view: Option<(usize, Option)>, pending_hover: bool, pending_completion: bool, hover_owner: Option, diff_workers: BTreeMap, resolver_worker: Option, blame_worker: Option, blame_source: Option, blame_lines: Option>>>, blame_request_id: u64, gutter_worker: Option, gutter_sources: Vec, gutter_request_id: u64, gutter_pending: bool, last_gutter_refresh: Option, picker_source: Option, battery: Option, battery_enabled: bool, last_file_refresh: Instant, last_diff_refresh: Instant, pub(crate) diff_follow_working_copy: bool, } impl EditorSession { /// File errors fail opening; rejected configuration leaves core editing /// available and presents a persistent startup diagnostic. pub fn open( file: Option, capabilities: Vec, config: Option<&Path>, ) -> io::Result { let mut editor = Editor::open(file, capabilities)?; let mut notices = Vec::new(); let config = match config { Some(path) => load_config(path).unwrap_or_else(|error| { editor.show_startup_diagnostic(format!( "{}\n\nConfiguration was rejected; using default settings and bindings.\n\n{error}", path.display() )); Config::default() }), None => Config::default(), }; editor.set_review_directory(match config.review_directory { Some(directory) => directory, None => crate::review::default_directory().map_err(io::Error::other)?, }); if let Err(error) = editor.set_leader(&config.leader) { notices.push(format!("leader {:?}: {error}", config.leader)); } if let Err(error) = editor.set_file_explorer_side(&config.file_explorer_side) { notices.push(format!( "file explorer side {:?}: {error}", config.file_explorer_side )); } if let Err(error) = editor.set_file_picker_view(&config.file_picker_view) { notices.push(format!( "file picker view {:?}: {error}", config.file_picker_view )); } if let Err(error) = editor.set_file_explorer_view(&config.file_explorer_view) { notices.push(format!( "file explorer view {:?}: {error}", config.file_explorer_view )); } if let Err(error) = editor.set_clipboard_mode(&config.clipboard_mode) { notices.push(format!( "clipboard mode {:?}: {error}", config.clipboard_mode )); } editor.set_git_blame(config.git_blame); editor.set_gutter_style(config.git_gutter); editor.set_scroll_distance(config.scroll_distance); editor.set_scroll_step_lines(config.scroll_step_lines); editor.set_highlight_trailing_whitespace(config.highlight_trailing_whitespace); editor.set_whitespace_characters(config.whitespace_characters); editor.set_whitespace_scope(config.whitespace_scope); editor.set_tab_spaces(config.tab_spaces, config.language_tab_spaces); editor.set_diff_context_options(config.diff_context_lines, config.diff_expand_lines); editor.set_diff_display_options(config.diff_view_mode, config.diff_split_orientation); editor.resolve_history_layout = config.resolve_history_layout; if let Err(error) = editor.set_chrome_config(config.top_bar, config.bottom_bar, config.labels) { notices.push(format!("chrome: {error}")); } editor.set_popup_config(config.popup); let mut plugins = BTreeMap::new(); 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) { Ok(()) => { plugins.insert(name, plugin); } Err(error) => notices.push(format!("{}: {error}", path.display())), } } Err(error) => notices.push(format!("{}: {error}", path.display())), } } for binding in config.bindings { if let Err(error) = editor.bind(&binding.mode, &binding.key, binding.command) { notices.push(format!( "binding {} {:?}: {error}", binding.mode, binding.key )); } } if !notices.is_empty() { editor.set_message(notices.join(" | ")); } let mut session = Self { editor, plugins, documents: BTreeMap::new(), requests: BTreeMap::new(), lsp: Lsp::new(config.lsp), lsp_request_id: 0, lsp_source_cursor: None, lsp_source_mode: Mode::Normal, lsp_source_view: None, pending_hover: false, pending_completion: false, hover_owner: None, diff_workers: BTreeMap::new(), resolver_worker: None, blame_worker: None, blame_source: None, blame_lines: None, blame_request_id: 0, gutter_worker: None, gutter_sources: Vec::new(), gutter_request_id: 0, gutter_pending: false, last_gutter_refresh: None, picker_source: None, battery: None, battery_enabled: false, last_file_refresh: Instant::now(), last_diff_refresh: Instant::now(), diff_follow_working_copy: config.diff_follow_working_copy, }; session.synchronize_documents(); session.poll_battery(); session.poll_gutter_at(Instant::now()); Ok(session) } pub fn editor(&self) -> &Editor { &self.editor } /// Rendering-only mutable access; document synchronization stays in session dispatch. pub(crate) fn editor_mut(&mut self) -> &mut Editor { &mut self.editor } pub(crate) fn set_diff_viewport( &mut self, comparison_id: u64, file: usize, top: usize, left_before: usize, left_after: usize, ) { self.editor .set_diff_viewport(comparison_id, file, top, left_before, left_after); } pub(crate) fn scroll_diff(&mut self, focus: Focus, down: bool) { self.editor.scroll_diff(focus, down); self.synchronize_documents(); } pub(crate) fn scroll_diff_horizontal(&mut self, focus: Focus, right: bool, width: usize) { self.editor.scroll_diff_horizontal(focus, right, width); self.synchronize_documents(); } pub fn handle(&mut self, input: Input) { if input == Input::Ctrl('r') && self .editor .normal_file_picker() .is_some_and(|picker| picker.focused) { self.last_gutter_refresh = None; } self.cancel_pending_popup_request(); let invocation = self.editor.handle(input); self.synchronize_documents(); if let Some(invocation) = invocation { self.dispatch(invocation); self.synchronize_documents(); } self.start_diff_request(); self.start_resolver_requests(); } pub fn execute(&mut self, command: &str) { self.cancel_pending_popup_request(); let invocation = self.editor.execute(command); self.synchronize_documents(); if let Some(invocation) = invocation { self.dispatch(invocation); self.synchronize_documents(); } self.start_diff_request(); self.start_resolver_requests(); } fn start_resolver_requests(&mut self) { while let Some(request) = self.editor.take_resolver_request() { if self.resolver_worker.is_none() { match ResolverWorker::start() { Ok(worker) => self.resolver_worker = Some(worker), Err(error) => { self.fail_resolver_request(request, error); continue; } } } if let Err(error) = self .resolver_worker .as_ref() .unwrap() .requests .as_ref() .unwrap() .send(request) { self.fail_resolver_request(error.0, "Resolver worker stopped unexpectedly".into()); self.resolver_worker = None; } } } fn fail_resolver_request(&mut self, request: ResolverRequest, error: String) { let outcome = match request { ResolverRequest::Load { tab, .. } => ResolverOutcome::Loaded { tab, result: Err(error), }, ResolverRequest::Save { tab, file, buffer, quit, .. } => ResolverOutcome::Saved { tab, file, buffer, quit, result: Err(error), }, }; self.editor.apply_resolver_outcome(outcome); } fn poll_resolver(&mut self) -> bool { self.start_resolver_requests(); let mut changed = false; if let Some(worker) = &self.resolver_worker { while let Ok(outcome) = worker.results.try_recv() { changed |= self.editor.apply_resolver_outcome(outcome); } } changed } fn start_diff_request(&mut self) { self.diff_workers.retain(|tab, worker| { let Some(diff) = self.editor.tab_diff(*tab) else { return false; }; if let Some(listing) = &mut worker.pull_requests { if let Err(error) = listing.update(diff.category, diff.pull_request_retries) { if let Some(id) = worker.pending.take() { self.editor.apply_diff_result(id, Err(error)); } return false; } return true; } if !diff.loading && !diff.refresh_pending() && let Some(id) = worker.pending.take() { worker.cancelled_request.store(id, Ordering::Release); worker.cancelled.store(true, Ordering::Release); } true }); while let Some((tab, id, operation)) = self.editor.take_next_diff_request() { if let std::collections::btree_map::Entry::Vacant(entry) = self.diff_workers.entry(tab) { match DiffWorker::start(self.editor.diff_root(tab)) { Ok(worker) => { entry.insert(worker); } Err(error) => { self.editor.apply_diff_result(id, Err(error)); continue; } } } if let Err(error) = self .diff_workers .get_mut(&tab) .expect("diff worker started") .request(id, operation) { self.editor.apply_diff_result(id, Err(error)); self.diff_workers.remove(&tab); } } } fn poll_diff(&mut self) -> bool { self.start_diff_request(); let mut changed = false; self.diff_workers.retain(|_, worker| { loop { let result = worker .results .as_ref() .expect("live diff worker has results") .try_recv(); match result { Ok((id, result)) => { if let Ok(DiffOutcome::PullRequestPage { closed, result }) = &result { if let Some(listing) = worker .pull_requests .as_mut() .filter(|listing| listing.id == id) { listing.complete(*closed, result); } } else if worker.pending == Some(id) { worker.pending = None; } changed |= self.editor.apply_diff_result(id, result); } Err(mpsc::TryRecvError::Empty) => { if let Some(listing) = &mut worker.pull_requests && let Err(error) = listing.schedule() { changed |= self.editor.apply_diff_result(listing.id, Err(error)); return false; } return true; } Err(mpsc::TryRecvError::Disconnected) => { if let Some(id) = worker.pending.take() { changed |= self.editor.apply_diff_result( id, Err("Diff worker stopped unexpectedly".into()), ); } return false; } } } }); changed } fn dispatch(&mut self, invocation: Invocation) { if invocation.plugin == "core" { if !self.editor.document_actions_available() || (invocation.command == "lsp.completion" && self.editor.mutation_blocked()) { return; } let result = match invocation.command.as_str() { "lsp.definition" | "lsp.hover" | "lsp.completion" => { if let Some(id) = self.lsp_request_id.checked_add(1) { self.lsp_request_id = id; let context = self.editor.lsp_context(); let source = InvocationSource { buffer_id: context.buffer_id, pane_id: self.editor.pane_id(), revision: context.revision, }; self.requests.entry("core".into()).or_default().latest = Some((id, source)); self.lsp_source_cursor = Some(self.editor.cursor()); self.lsp_source_mode = self.editor.mode(); self.lsp_source_view = Some(( self.editor.tab_id(), self.editor.diff().and_then(|diff| diff.comparison_id), )); self.pending_hover = invocation.command == "lsp.hover"; self.pending_completion = invocation.command == "lsp.completion"; match invocation.command.as_str() { "lsp.hover" => self.lsp.hover(&context, id), "lsp.completion" => self.lsp.completion(&context, id), _ => self.lsp.definition(&context, id), } } else { Err("LSP request IDs exhausted".into()) } } "lsp.diagnostics" => self.lsp.diagnostics(self.editor.buffer_id()), _ => unreachable!("session-dispatched core command"), }; if let Err(error) = result { self.cancel_pending_popup_request(); self.editor.set_message(format!("LSP: {error}")); } return; } let context = self.editor.context(); let read_only = self.editor.mutation_blocked(); 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, pane_id: self.editor.pane_id(), revision: context.revision, }; requests.pending.insert(id, source); requests.latest = Some((id, source)); }) } Some(plugin) => plugin .invoke(&invocation.command, &invocation.args, &context) .and_then(|effects| { if read_only && effects .iter() .any(|effect| matches!(effect, Effect::Insert(_) | Effect::Replace(_))) { self.editor.set_message("Diff pane is read-only"); Ok(()) } else { self.editor.apply(effects) } }), None => Err("plugin host is unavailable".into()), }; if let Err(error) = result { self.retire(&invocation.plugin, &error); } } fn cancel_pending_popup_request(&mut self) { if self.pending_hover || self.pending_completion { if let Some(requests) = self.requests.get_mut("core") { requests.latest = None; } self.pending_hover = false; self.pending_completion = false; } } 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); if self.hover_owner.as_deref() == Some(owner) { self.editor.clear_hover(); self.hover_owner = None; } } fn synchronize_documents(&mut self) -> bool { self.editor.refresh_resolver_comparison(); let blame_changed = self.poll_blame(); // Leaving a request's source invalidates it permanently, even if history returns there. if let Some(requests) = self.requests.get_mut("core") && requests.latest.is_some_and(|(_, source)| { source.buffer_id != self.editor.buffer_id() || source.pane_id != self.editor.pane_id() || source.revision != self.editor.text_revision() || self.lsp_source_cursor != Some(self.editor.cursor()) || self.editor.mode() != self.lsp_source_mode || self.lsp_source_view != Some(( self.editor.tab_id(), self.editor.diff().and_then(|diff| diff.comparison_id), )) || !self.editor.document_actions_available() }) { requests.latest = None; self.lsp_source_cursor = None; self.lsp_source_view = None; self.pending_hover = false; self.pending_completion = false; } if !self.lsp.is_enabled() && !self.plugins.values().any(RubyPlugin::is_async) { self.documents.clear(); return self.refresh_lsp_names() | blame_changed; } let mut events = Vec::new(); if !self.editor.should_quit() { for document in self.editor.interaction_documents() { let path = document.path.map(Path::to_string_lossy); let root = self .editor .interaction_document_root(document.id) .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() || previous.snapshot.root.as_str() != root.as_ref() }); 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: root.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(); let mut changed = blame_changed; for event in &events { if let Err(error) = self.lsp.notify_document(event) { self.editor.set_message(format!("LSP: {error}")); changed = true; } } 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; } } } changed |= !failures.is_empty(); for (owner, error) in failures { self.retire(&owner, &error); } changed | self.refresh_lsp_names() } fn refresh_lsp_names(&mut self) -> bool { let buffer_id = self.editor.buffer_id(); if self.lsp.active_server_names(buffer_id).eq(self .editor .lsp_names() .iter() .map(String::as_str)) { return false; } self.editor.set_lsp_names( self.lsp .active_server_names(buffer_id) .map(str::to_owned) .collect(), ) } fn source_is_current(&self, source: InvocationSource) -> bool { self.editor .interaction_documents() .any(|document| document.id == source.buffer_id && document.revision == source.revision) } fn request_is_current( &self, owner: &str, request_id: Option, source: InvocationSource, ) -> bool { if !self.editor.document_actions_available() { return false; } self.editor.buffer_id() == source.buffer_id && self.source_is_current(source) && (owner != "core" || (self.lsp_source_cursor == Some(self.editor.cursor()) && self.editor.mode() == self.lsp_source_mode && self.lsp_source_view == Some(( self.editor.tab_id(), self.editor.diff().and_then(|diff| diff.comparison_id), )))) && 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.pane_id == self.editor.pane_id() && latest.revision == source.revision }) } fn hover_is_current( &self, owner: &str, request_id: Option, source: InvocationSource, ) -> bool { self.request_is_current(owner, request_id, source) && self.editor.mode() == Mode::Normal && self.editor.palette().is_none() && self .editor .file_picker() .is_none_or(|picker| !picker.focused) } fn completion_is_current(&self, request_id: Option, source: InvocationSource) -> bool { self.pending_completion && !self.editor.mutation_blocked() && self.request_is_current("core", request_id, source) && self.editor.mode() == Mode::Insert && self.editor.palette().is_none() && self .editor .file_picker() .is_none_or(|picker| !picker.focused) && self.editor.startup_diagnostic().is_none() } fn apply_update(&mut self, owner: &str, mut update: PluginUpdate) -> Result { if owner == "core" && let Some(id) = update.request_id && self .requests .get(owner) .and_then(|requests| requests.latest) .map(|latest| latest.0) != Some(id) { update .effects .retain(|effect| matches!(effect, UpdateEffect::Label { .. })); } for effect in &update.effects { if let UpdateEffect::Completion { buffer_id, revision, items, } = effect { if owner != "core" { return Err("Completion is a native LSP update only".into()); } let source = InvocationSource { buffer_id: *buffer_id, pane_id: self.editor.pane_id(), revision: *revision, }; if self.completion_is_current(update.request_id, source) && let Err(error) = self.editor.validate_completion(items) { self.pending_completion = false; self.editor .set_message(format!("LSP completion failed: {error}")); return Ok(true); } } } // Validate the whole applicable annotation batch before changing any buffer. let validation = (|| -> Result<(), String> { let mut label_bytes = 0usize; for effect in &update.effects { match effect { UpdateEffect::Diagnostics { buffer_id, revision, items, } => { self.editor .validate_diagnostics(*buffer_id, *revision, items)?; } UpdateEffect::SemanticTokens { buffer_id, revision, tokens, } => { if owner != "core" { return Err("Semantic tokens are native LSP updates only".into()); } self.editor.validate_semantic_tokens( *buffer_id, *revision, tokens.as_deref(), )?; } UpdateEffect::Navigate { buffer_id, revision, path, .. } | UpdateEffect::NavigateUtf16 { buffer_id, revision, path, .. } => { if owner != "core" && matches!(effect, UpdateEffect::NavigateUtf16 { .. }) { return Err("UTF-16 navigation is a native LSP update only".into()); } let source = InvocationSource { buffer_id: *buffer_id, pane_id: self.editor.pane_id(), revision: *revision, }; if self.request_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::Hover { buffer_id, revision, text, } => { let source = InvocationSource { buffer_id: *buffer_id, pane_id: self.editor.pane_id(), revision: *revision, }; if self.hover_is_current(owner, update.request_id, source) { crate::editor::validate_hover_text(text)?; } } UpdateEffect::Completion { .. } => {} UpdateEffect::Message { .. } => {} UpdateEffect::Label { name, text } => { crate::editor::validate_label(name, text)?; label_bytes = label_bytes .checked_add(name.len()) .and_then(|bytes| bytes.checked_add(text.len())) .ok_or("Labels exceed 4 MiB")?; if label_bytes > 4 * 1024 * 1024 { return Err("Labels exceed 4 MiB".into()); } } } } Ok(()) })(); if let Err(error) = validation { if owner == "core" { self.editor.clear_semantic_tokens(); } return Err(error); } 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::SemanticTokens { buffer_id, revision, tokens, } => { changed |= self .editor .replace_validated_semantic_tokens(buffer_id, revision, tokens); } UpdateEffect::NavigateUtf16 { buffer_id, revision, path, range, target_range, } => { let source = InvocationSource { buffer_id, pane_id: self.editor.pane_id(), revision, }; if self.request_is_current(owner, update.request_id, source) { if let Err(error) = self.editor.navigate_lsp_to(path, range, target_range) { self.editor .set_message(format!("LSP: definition failed: {error}")); } changed = true; } } UpdateEffect::Navigate { buffer_id, revision, path, row, column, } => { let source = InvocationSource { buffer_id, pane_id: self.editor.pane_id(), revision, }; if self.request_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::Hover { buffer_id, revision, text, } => { let source = InvocationSource { buffer_id, pane_id: self.editor.pane_id(), revision, }; if self.hover_is_current(owner, update.request_id, source) { self.editor.show_hover(text)?; self.hover_owner = Some(owner.into()); if owner == "core" { self.pending_hover = false; } changed = true; } } UpdateEffect::Completion { buffer_id, revision, items, } => { let source = InvocationSource { buffer_id, pane_id: self.editor.pane_id(), revision, }; if self.completion_is_current(update.request_id, source) { self.pending_completion = false; if let Err(error) = self.editor.show_completion(items) { self.editor .set_message(format!("LSP completion failed: {error}")); } changed = true; } } UpdateEffect::Message { text } => { changed |= self.editor.message() != text; self.editor.set_message(text); } UpdateEffect::Label { name, text } => { changed |= self.editor.set_label(name, text)?; } } } self.editor.refresh_live_diff(); Ok(changed) } fn apply_response( &mut self, owner: &str, id: u64, mut effects: Vec, ) -> Result { 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()); }; // Document effects retain source freshness; labels belong to the whole session. let current = self.source_is_current(source) && self.editor.buffer_id() == source.buffer_id && self.editor.pane_id() == source.pane_id; if !current { effects.retain(|effect| matches!(effect, Effect::Label { .. })); } if effects .iter() .all(|effect| matches!(effect, Effect::Message(_) | Effect::Label { .. })) { crate::editor::validate_effects(&effects)?; let mut changed = false; for effect in effects { match effect { Effect::Message(text) => { changed |= self.editor.message() != text; self.editor.set_message(text); } Effect::Label { name, text } => { changed |= self.editor.set_label(name, text)?; } _ => unreachable!("presentation-only effects checked above"), } } return Ok(changed); } if self.editor.mutation_blocked() { return Ok(false); } let changed = !effects.is_empty(); self.editor.apply(effects)?; Ok(changed) } fn poll_gutter_at(&mut self, now: Instant) -> bool { let line_gutter = self.editor.gutter_style() != crate::gutter::GutterStyle::None; let picker_changed = self.picker_source.as_deref() != Some(self.editor.project_root()); if self.editor.should_quit() { self.gutter_worker = None; self.gutter_sources.clear(); self.gutter_pending = false; self.picker_source = None; return false; } let sources_changed = !self .editor .documents() .filter(|document| line_gutter || document.dirty) .filter_map(|document| { document.path.map(|path| { ( document.id, document.revision, document.save_revision, path, document.dirty, ) }) }) .eq(self.gutter_sources.iter().map(|source| { ( source.buffer, source.revision, source.save_revision, source.path.as_path(), source.dirty, ) })); let refresh_due = !self.gutter_pending && self .last_gutter_refresh .is_none_or(|last| now.saturating_duration_since(last) >= Duration::from_secs(1)); if sources_changed || picker_changed || refresh_due { if picker_changed { self.picker_source = Some(self.editor.project_root().to_owned()); } if sources_changed { self.gutter_sources = self .editor .documents() .filter(|document| line_gutter || document.dirty) .filter_map(|document| { let path = document.path?; let contents = self .gutter_sources .binary_search_by_key(&document.id, |source| source.buffer) .ok() .map(|index| &self.gutter_sources[index]) .filter(|source| { source.revision == document.revision && source.path == path }) .map_or_else( || Arc::from(document.lines.join("\n")), |source| Arc::clone(&source.contents), ); Some(GutterSource { buffer: document.id, revision: document.revision, save_revision: document.save_revision, path: path.to_path_buf(), dirty: document.dirty, contents, }) }) .collect(); } self.gutter_request_id = self.gutter_request_id.wrapping_add(1); self.last_gutter_refresh = Some(now); if self.gutter_worker.is_none() { match crate::gutter::Worker::start() { Ok(worker) => self.gutter_worker = Some(worker), Err(error) => { self.editor.set_message(format!("Gutter: {error}")); return true; } } } let files = self .gutter_sources .iter() .filter(|_| line_gutter) .map(|source| { ( source.buffer, source.path.clone(), Arc::clone(&source.contents), ) }) .collect(); let picker = self.picker_source.as_ref().map(|root| { let buffers = self .gutter_sources .iter() .filter(|source| source.dirty) .map(|source| (source.path.clone(), Arc::clone(&source.contents))) .collect(); (root.clone(), buffers) }); match self .gutter_worker .as_mut() .expect("gutter worker started") .request(self.gutter_request_id, files, picker) { Ok(()) => self.gutter_pending = true, Err(error) => { self.gutter_worker = None; self.gutter_pending = false; self.editor.set_message(format!("Gutter: {error}")); return true; } } } let mut changed = false; while let Some((id, result, picker)) = self .gutter_worker .as_mut() .and_then(crate::gutter::Worker::poll) { changed |= self.apply_gutter_result(id, result); if let Some(picker) = picker { changed |= self.apply_picker_result(id, picker); } } changed } fn apply_gutter_result(&mut self, id: u64, result: crate::gutter::ResultBatch) -> bool { if id != self.gutter_request_id { return false; } self.gutter_pending = false; let mut changed = false; for (buffer, result) in result { let Some(source) = self .gutter_sources .iter() .find(|source| source.buffer == buffer) else { continue; }; let changes = match result { Ok(changes) => changes, Err(error) => { self.editor.set_message(format!("Gutter: {error}")); changed = true; Vec::new() } }; changed |= self.editor .set_gutter_changes(buffer, source.revision, &source.path, changes); } changed } fn apply_picker_result(&mut self, id: u64, reply: crate::gutter::PickerChanges) -> bool { let Some(root) = &self.picker_source else { return false; }; if id != self.gutter_request_id || *root != reply.root || *root != self.editor.project_root() { return false; } let mut changed = false; let changes = match reply.result { Ok(changes) => changes, Err(error) => { self.editor.set_message(format!("File status: {error}")); changed = true; Vec::new() } }; changed | self .editor .set_picker_repository_changes(&reply.root, changes) } pub(crate) fn refresh_gutter(&mut self) -> bool { self.last_gutter_refresh = None; self.poll_gutter_at(Instant::now()) } fn poll_blame(&mut self) -> bool { let mut changed = false; if !self .blame_source .as_ref() .is_some_and(|source| self.editor.blame_file_source_is_current(source)) { self.blame_source = self.editor.blame_source(); self.blame_request_id = self.blame_request_id.wrapping_add(1); self.blame_lines = None; changed |= self.editor.clear_line_blame(); if let Some(source) = &self.blame_source { if self.blame_worker.is_none() { self.blame_worker = BlameWorker::start().ok(); } if let Some(worker) = &mut self.blame_worker && worker .request( self.blame_request_id, source.path.to_path_buf(), self.editor.lines().join("\n"), ) .is_err() { self.blame_worker = None; } } else { self.blame_worker = None; } } while let Some((id, result)) = self.blame_worker.as_mut().and_then(BlameWorker::poll) { changed |= self.apply_blame_result(id, result); } changed | self.refresh_line_blame() } fn apply_blame_result( &mut self, id: u64, result: Result>>>, String>, ) -> bool { if id != self.blame_request_id || !self .blame_source .as_ref() .is_some_and(|source| self.editor.blame_file_source_is_current(source)) { return false; } self.blame_lines = result.ok().flatten(); self.refresh_line_blame() } fn refresh_line_blame(&mut self) -> bool { if let Some(source) = &self.blame_source && self.editor.blame_file_source_is_current(source) && let Some(blame) = self .blame_lines .as_ref() .and_then(|lines| lines.get(self.editor.cursor().row)) .and_then(Option::as_ref) { self.editor.set_cached_line_blame(source, blame.clone()) } else { self.editor.clear_line_blame() } } pub(crate) fn refresh_blame(&mut self) -> bool { self.blame_source = None; self.poll_blame() } fn poll_battery(&mut self) -> bool { let enabled = self.editor.label("battery").is_empty() && [ self.editor.top_bar_config(), self.editor.bottom_bar_config(), ] .into_iter() .any(|bar| { bar.left .iter() .chain(&bar.center) .chain(&bar.right) .any(|name| name == "battery") }); let mut changed = false; if enabled != self.battery_enabled { self.battery_enabled = enabled; // Unit sessions do not read external hardware; provider and renderer checks inject data. #[cfg(not(test))] { self.battery = if enabled { crate::battery::Battery::start().ok() } else { None }; } if !enabled { changed |= self.editor.set_battery_charge(None); } } if let Some(charge) = self.battery.as_mut().and_then(|battery| battery.poll()) { changed |= self.editor.set_battery_charge(charge); } changed } fn refresh_files_at(&mut self, now: Instant) -> bool { if now.saturating_duration_since(self.last_file_refresh) < Duration::from_millis(250) { return false; } self.last_file_refresh = now; self.editor.refresh_files() } fn refresh_checkout_diffs_at(&mut self, now: Instant) { if now.saturating_duration_since(self.last_diff_refresh) < Duration::from_secs(1) { return; } self.last_diff_refresh = now; self.editor .queue_checkout_refreshes(self.diff_follow_working_copy); } /// Drains ready plugin work without waiting for command or server responses. pub fn poll(&mut self) -> bool { self.poll_at(Instant::now()) } fn poll_at(&mut self, now: Instant) -> bool { let mut changed = self.refresh_files_at(now); changed |= self.poll_gutter_at(now); changed |= self.synchronize_documents(); changed |= self.poll_battery(); self.refresh_checkout_diffs_at(now); changed |= self.poll_diff(); changed |= self.poll_resolver(); for update in self.lsp.poll() { match self.apply_update("core", update) { Ok(applied) => changed |= applied, Err(error) => { self.editor.clear_plugin_diagnostics("core"); self.editor.set_message(format!("LSP: {error}")); changed = true; } } } 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)] #[path = "../tests/unit/session.rs"] mod tests;