Compare commits

...
3 Commits
Author SHA1 Message Date
Riz Ashraf 6a89f50cf2 feat(nvim-core): implement nvim_set_extmark for rich virtual text UI feedback
- Added nvim_set_extmark tool schema and handler
- Implemented lua bridge using vim.api.nvim_buf_set_extmark to dynamically overlay ghost text and highlight groups onto active buffers
2026-09-22 22:34:06 +01:00
Riz Ashraf ea735003e2 feat(nvim-core): implement bidirectional event streaming and real-time shadow buffer
- Created global thread-safe NvimState to shadow Neovim context.
- Injected Lua autocmds (CursorMoved, TextChanged) on connection to stream events back to MCP.
- Upgraded msgpack RPC loop to intercept and parse 'mcp_event' notifications alongside standard responses.
- Optimized nvim_get_cursor to pull instantly from the local shadow buffer, bypassing RPC round-trips.
2026-09-22 22:32:37 +01:00
Riz Ashraf 61b03bc6e3 perf(nvim-core): implement zero-copy BytesMut stream buffer and DashMap for lock contention 2026-09-22 22:26:00 +01:00
3 changed files with 186 additions and 42 deletions

No files matched your search

Generated
+22
View File
@@ -497,6 +497,20 @@ dependencies = [
"syn 3.0.6",
]
[[package]]
name = "dashmap"
version = "6.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6361d5c062261c78a176addb82d4c821ae42bed6089de0e12603cd25de2059c"
dependencies = [
"cfg-if",
"crossbeam-utils",
"hashbrown 0.14.5",
"lock_api",
"once_cell",
"parking_lot_core",
]
[[package]]
name = "data-encoding"
version = "2.11.1"
@@ -807,6 +821,12 @@ dependencies = [
"rand_core 0.10.1",
]
[[package]]
name = "hashbrown"
version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
[[package]]
name = "hashbrown"
version = "0.16.1"
@@ -1375,6 +1395,8 @@ dependencies = [
name = "nvim-core"
version = "0.1.0"
dependencies = [
"bytes",
"dashmap",
"dirs",
"mcp-stdio",
"rmcp",
+2
View File
@@ -14,4 +14,6 @@ tracing-subscriber = "0.3.23"
dirs = "7.0.0"
rmcp = { version = "3.4.0", features = ["server"] }
mcp-stdio = { version = "0.1.0", path = "../mcp-stdio" }
bytes = "1.12.1"
dashmap = "6.2.1"
+162 -42
View File
@@ -114,7 +114,6 @@ async fn get_socket_path() -> Result<String, String> {
}
Err("Could not find Neovim socket".to_string())
}
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::LazyLock;
use tokio::sync::{mpsc, oneshot};
@@ -130,6 +129,36 @@ static NEXT_MSGID: AtomicU64 = AtomicU64::new(1);
static NVIM_CONN: LazyLock<Arc<std::sync::Mutex<Option<mpsc::Sender<NvimRequest>>>>> =
LazyLock::new(|| Arc::new(std::sync::Mutex::new(None)));
#[derive(Debug, Default, Clone)]
pub struct NvimState {
pub cursor: String,
pub active_buffer_id: u64,
}
static NVIM_STATE: LazyLock<Arc<std::sync::Mutex<NvimState>>> =
LazyLock::new(|| Arc::new(std::sync::Mutex::new(NvimState::default())));
fn handle_nvim_notification(params: &[rmpv::Value]) {
if params.is_empty() { return; }
if let rmpv::Value::String(event) = &params[0] {
match event.as_str().unwrap_or("") {
"CursorMoved" => {
if params.len() > 1
&& let rmpv::Value::Array(pos) = &params[1]
&& pos.len() >= 4
&& let (rmpv::Value::Integer(row), rmpv::Value::Integer(col)) = (&pos[1], &pos[2]) {
let mut state = NVIM_STATE.lock().unwrap_or_else(|e| e.into_inner());
state.cursor = format!("Line: {}, Column: {}", row, col);
}
},
"TextChanged" => {
tracing::debug!("Shadow buffer text changed event received");
}
_ => {}
}
}
}
async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
{
let conn_lock = NVIM_CONN.lock().unwrap_or_else(|e| e.into_inner());
@@ -162,8 +191,8 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
let (mut read_half, mut write_half) = tokio::io::split(stream);
let (tx, mut rx) = mpsc::channel::<NvimRequest>(32);
type PendingRequestsMap =
Arc<std::sync::Mutex<HashMap<u64, oneshot::Sender<Result<rmpv::Value, String>>>>>;
let pending_requests: PendingRequestsMap = Arc::new(std::sync::Mutex::new(HashMap::new()));
Arc<dashmap::DashMap<u64, oneshot::Sender<Result<rmpv::Value, String>>>>;
let pending_requests: PendingRequestsMap = Arc::new(dashmap::DashMap::new());
// Write task
let pending_clone = Arc::clone(&pending_requests);
@@ -175,10 +204,7 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
continue;
}
pending_clone
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(req.msgid, req.reply);
pending_clone.insert(req.msgid, req.reply);
if write_half.write_all(&buf).await.is_err() {
tracing::error!("Failed to write to Neovim socket");
@@ -190,39 +216,31 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
// Read task
let pending_clone2 = Arc::clone(&pending_requests);
tokio::spawn(async move {
let mut resp_buf = Vec::new();
let mut chunk = vec![0u8; 65536];
let mut offset = 0;
use bytes::{Buf, BytesMut};
let mut resp_buf = BytesMut::with_capacity(65536);
loop {
let mut cursor = std::io::Cursor::new(&resp_buf[offset..]);
let mut cursor = std::io::Cursor::new(&resp_buf[..]);
match rmpv::decode::read_value(&mut cursor) {
Ok(val) => {
offset += cursor.position() as usize;
let parsed_len = cursor.position() as usize;
resp_buf.advance(parsed_len);
if let rmpv::Value::Array(ref arr) = val
&& arr.len() >= 4
&& arr[0] == rmpv::Value::Integer(1.into())
{
if let rmpv::Value::Array(ref arr) = val {
if arr.len() >= 4 && arr[0] == rmpv::Value::Integer(1.into()) {
let msgid = match &arr[1] {
rmpv::Value::Integer(i) => i.as_u64().unwrap_or(0),
_ => 0,
};
if let Some(reply_sender) = pending_clone2
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&msgid)
{
if let Some((_, reply_sender)) = pending_clone2.remove(&msgid) {
let _ = reply_sender.send(Ok(val));
}
} else if arr.len() >= 3 && arr[0] == rmpv::Value::Integer(2.into())
&& let rmpv::Value::String(method) = &arr[1]
&& method.as_str().unwrap_or("") == "mcp_event"
&& let rmpv::Value::Array(params) = &arr[2] {
handle_nvim_notification(params);
}
if offset == resp_buf.len() {
resp_buf.clear();
offset = 0;
} else if offset > 1024 * 1024 {
resp_buf.drain(..offset);
offset = 0;
}
continue;
}
@@ -237,16 +255,13 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
_ => false,
} =>
{
resp_buf.drain(..offset);
offset = 0;
let read_future = read_half.read(&mut chunk);
match tokio::time::timeout(tokio::time::Duration::from_secs(60), read_future)
match tokio::time::timeout(
tokio::time::Duration::from_secs(60),
read_half.read_buf(&mut resp_buf),
)
.await
{
Ok(Ok(n)) if n > 0 => {
resp_buf.extend_from_slice(&chunk[..n]);
}
Ok(Ok(n)) if n > 0 => {}
_ => {
tracing::error!("Neovim socket read loop closed or timeout");
break;
@@ -261,10 +276,12 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
}
// Cleanup pending requests on disconnect
let mut pending = pending_clone2.lock().unwrap_or_else(|e| e.into_inner());
for (_, sender) in pending.drain() {
let keys: Vec<_> = pending_clone2.iter().map(|kv| *kv.key()).collect();
for k in keys {
if let Some((_, sender)) = pending_clone2.remove(&k) {
let _ = sender.send(Err("Connection closed".to_string()));
}
}
});
// Cleanup task for timed-out requests
@@ -276,10 +293,7 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
if Arc::strong_count(&pending_clone3) <= 1 {
break; // Socket closed and other tasks finished, no need to keep cleaning up
}
pending_clone3
.lock()
.unwrap_or_else(|e| e.into_inner())
.retain(|_, sender| !sender.is_closed());
pending_clone3.retain(|_, sender| !sender.is_closed());
}
});
@@ -291,6 +305,45 @@ async fn get_nvim_connection() -> Result<mpsc::Sender<NvimRequest>, String> {
return Ok(existing_sender.clone());
}
*conn_lock = Some(tx.clone());
drop(conn_lock);
let tx_clone = tx.clone();
tokio::spawn(async move {
let setup_code = r#"
local channel = vim.api.nvim_get_api_info()[1]
vim.api.nvim_create_augroup("MCP_Tracking", { clear = true })
vim.api.nvim_create_autocmd({"CursorMoved", "CursorMovedI"}, {
group = "MCP_Tracking",
callback = function()
pcall(vim.rpcnotify, channel, "mcp_event", "CursorMoved", vim.fn.getpos('.'))
end
})
vim.api.nvim_create_autocmd({"TextChanged", "TextChangedI", "BufEnter"}, {
group = "MCP_Tracking",
callback = function()
pcall(vim.rpcnotify, channel, "mcp_event", "TextChanged", vim.api.nvim_get_current_buf())
end
})
"#;
let msgid = NEXT_MSGID.fetch_add(1, Ordering::SeqCst);
let req = rmpv::Value::Array(vec![
rmpv::Value::Integer(0.into()),
rmpv::Value::Integer(msgid.into()),
rmpv::Value::String("nvim_exec_lua".into()),
rmpv::Value::Array(vec![
rmpv::Value::String(setup_code.into()),
rmpv::Value::Array(vec![]),
]),
]);
let (reply_tx, _reply_rx) = tokio::sync::oneshot::channel();
let _ = tx_clone.send(NvimRequest {
msgid,
req,
reply: reply_tx,
}).await;
tracing::info!("Injected bidirectional event tracking autocmds into Neovim");
});
Ok(tx)
}
@@ -382,6 +435,13 @@ async fn get_nvim_active_buffer() -> Result<String, String> {
}
async fn get_nvim_cursor() -> Result<String, String> {
{
let state = NVIM_STATE.lock().unwrap_or_else(|e| e.into_inner());
if !state.cursor.is_empty() {
return Ok(state.cursor.clone());
}
}
let result =
call_nvim_method("nvim_win_get_cursor", vec![rmpv::Value::Integer(0.into())]).await?;
@@ -450,6 +510,31 @@ async fn set_nvim_diagnostics(line: i64, message: &str) -> Result<(), String> {
Ok(())
}
async fn set_nvim_extmark(line: i64, text: &str, hl_group: &str) -> Result<(), String> {
let escaped_text = text.replace('\\', "\\\\").replace('"', "\\\"");
let lua_script = format!(
r#"
local ns = vim.api.nvim_create_namespace("gemini_extmarks")
local opts = {{
virt_text = {{{{"{escaped_text}", "{hl_group}"}}}},
virt_text_pos = "eol",
}}
vim.api.nvim_buf_set_extmark(0, ns, {line} - 1, 0, opts)
"#
);
call_nvim_method(
"nvim_exec_lua",
vec![
rmpv::Value::String(lua_script.into()),
rmpv::Value::Array(vec![]),
],
)
.await?;
Ok(())
}
fn rmpv_to_json(val: &rmpv::Value) -> serde_json::Value {
match val {
rmpv::Value::Nil => serde_json::Value::Null,
@@ -671,6 +756,19 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) {
"required": ["line", "message"]
}
},
{
"name": "nvim_set_extmark",
"description": "Add virtual text (ghost text) to a specific line in the buffer to give the user rich UI feedback",
"inputSchema": {
"type": "object",
"properties": {
"line": { "type": "integer" },
"text": { "type": "string" },
"highlight_group": { "type": "string", "description": "Optional. E.g., 'DiagnosticVirtualTextInfo', 'Comment'" }
},
"required": ["line", "text"]
}
},
{
"name": "nvim_list_buffers",
"description": "List all open buffers in Neovim",
@@ -892,6 +990,28 @@ pub async fn run_mcp_loop(app_name: &str, app_version: &str) {
send_error(id, -32602, "Missing 'line' or 'message'").await;
}
}
"nvim_set_extmark" => {
if let (Some(line), Some(text)) = (
args.get("line").and_then(serde_json::Value::as_i64),
args.get("text").and_then(|v| v.as_str()),
) {
let hl_group = args
.get("highlight_group")
.and_then(|v| v.as_str())
.unwrap_or("Comment");
match set_nvim_extmark(line, text, hl_group).await {
Ok(()) => {
send_text_result!(
id.clone(),
format!("Successfully set extmark on line {}", line)
);
}
Err(e) => send_error(id, -32603, &e).await,
}
} else {
send_error(id, -32602, "Missing 'line' or 'text'").await;
}
}
"nvim_list_buffers" => {
let lua_code = r"
local bufs = vim.api.nvim_list_bufs()