Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/160.fixed.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fixed file tool correctness: streaming `read_lines` for offset/limit reads, UTF-8-safe truncation, grep quote stripping, and long-line handling in read output.
1 change: 1 addition & 0 deletions changelog.d/160.performance.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Improved tool registry lookups with a HashMap-backed snapshot instead of linear scans.
40 changes: 24 additions & 16 deletions n00n-agent/src/tools/grep.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,10 @@

use std::io;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::sync::Arc;

use crate::{GrepFileEntry, GrepLine, GrepMatchGroup};
use flume;
use grep_regex::RegexMatcher;
use grep_searcher::Searcher;
use grep_searcher::SearcherBuilder;
Expand Down Expand Up @@ -55,6 +56,11 @@ impl GrepParams {
/// Returns an error if the search path or grep execution fails.
pub fn grep_search(params: &GrepParams) -> Result<(PathBuf, Vec<GrepFileEntry>), String> {
let search_path = resolve_search_path(params.path.as_deref())?;

if !Path::new(&search_path).exists() {
return Err(format!("grep: path not found: {search_path}"));
}

let is_multiline = needs_multiline(&params.pattern);
debug!(
pattern = %params.pattern,
Expand Down Expand Up @@ -97,21 +103,25 @@ pub fn grep_search(params: &GrepParams) -> Result<(PathBuf, Vec<GrepFileEntry>),
};
let has_context = params.context_before > 0 || params.context_after > 0;
let max_line_bytes = params.max_line_bytes;
let results: Arc<Mutex<Vec<GrepFileEntry>>> = Arc::new(Mutex::new(Vec::new()));
let (tx, rx) = flume::unbounded::<GrepFileEntry>();
let base: Arc<Path> = Arc::from(base);

walker.build_parallel().run({
let results = Arc::clone(&results);
let tx = tx.clone();
let matcher = Arc::new(matcher);
let base = Arc::clone(&base);
move || {
let mut searcher = builder.build();
let matcher = Arc::clone(&matcher);
let results = Arc::clone(&results);
let tx = tx.clone();
let base = Arc::clone(&base);
Box::new(move |entry| {
let Ok(entry) = entry else {
return WalkState::Continue;
let entry = match entry {
Ok(e) => e,
Err(e) => {
warn!(error = %e, "grep walk entry error");
return WalkState::Continue;
}
};
if !entry.file_type().is_some_and(|ft| ft.is_file()) {
return WalkState::Continue;
Expand All @@ -124,29 +134,27 @@ pub fn grep_search(params: &GrepParams) -> Result<(PathBuf, Vec<GrepFileEntry>),
max_line_bytes,
has_context,
};
let _ = searcher.search_path(&*matcher, &path, &mut sink);
if let Err(e) = searcher.search_path(&*matcher, &path, &mut sink) {
warn!(path = %path.display(), error = %e, "grep search_path failed");
}

if !groups.is_empty() {
let rel = path
.strip_prefix(&*base)
.unwrap_or_else(|_| &path)
.to_string_lossy()
.into_owned();
let mut guard = results
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
guard.push(GrepFileEntry { path: rel, groups });
if tx.send(GrepFileEntry { path: rel, groups }).is_err() {
return WalkState::Quit;
}
}
WalkState::Continue
})
}
});

let mut entries = std::mem::take(
&mut *results
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
);
drop(tx);
let mut entries: Vec<GrepFileEntry> = rx.iter().collect();

if entries.is_empty() {
return Ok((base.to_path_buf(), entries));
Expand Down
2 changes: 1 addition & 1 deletion n00n-agent/src/tools/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ pub use file_tracker::FileReadTracker;
pub use registry::{
ActiveTools, BoxFuture, ExecFuture, HeaderFuture, HeaderResult, ParseError, PermissionScopes,
RegisteredTool, RegistryError, Tool, ToolAudience, ToolExecResult, ToolInvocation,
ToolRegistry, ToolSearchResult, ToolSource,
ToolRegistry, ToolSearchResult, ToolSource, ToolsSnapshot,
};

use std::collections::HashMap;
Expand Down
145 changes: 113 additions & 32 deletions n00n-agent/src/tools/registry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
//! path, no parallel lists that can drift.

use std::borrow::Cow;
use std::collections::HashMap;
use std::future::Future;
use std::path::Path;
use std::pin::Pin;
Expand Down Expand Up @@ -274,9 +275,67 @@ impl RegisteredTool {
}
}

#[derive(Clone)]
pub struct ToolsSnapshot {
tools: Vec<RegisteredTool>,
by_name: HashMap<String, usize>,
}

impl ToolsSnapshot {
#[must_use]
pub fn empty() -> Self {
Self {
tools: Vec::new(),
by_name: HashMap::new(),
}
}

#[must_use]
pub fn from_tools(tools: Vec<RegisteredTool>) -> Self {
let mut by_name = HashMap::new();
for (i, tool) in tools.iter().enumerate() {
by_name.insert(tool.name().to_owned(), i);
}
Self { tools, by_name }
}

#[must_use]
pub fn get(&self, name: &str) -> Option<RegisteredTool> {
self.by_name.get(name).map(|&idx| self.tools[idx].clone())
}

#[must_use]
pub fn has(&self, name: &str) -> bool {
self.by_name.contains_key(name)
}

pub fn iter(&self) -> std::slice::Iter<'_, RegisteredTool> {
self.tools.iter()
}

#[must_use]
pub fn len(&self) -> usize {
self.tools.len()
}

#[must_use]
pub fn is_empty(&self) -> bool {
self.tools.is_empty()
}
}

impl<'a> IntoIterator for &'a ToolsSnapshot {
type Item = &'a RegisteredTool;
type IntoIter = std::slice::Iter<'a, RegisteredTool>;

fn into_iter(self) -> Self::IntoIter {
self.tools.iter()
}
}

/// Lock-free reads via `ArcSwap`, writes swap in a new snapshot atomically.
pub struct ToolRegistry {
tools: ArcSwap<Vec<RegisteredTool>>,
tools: ArcSwap<ToolsSnapshot>,
}

impl Default for ToolRegistry {
Expand All @@ -295,7 +354,7 @@ impl ToolRegistry {
#[must_use]
pub fn new() -> Self {
Self {
tools: ArcSwap::from_pointee(Vec::new()),
tools: ArcSwap::from_pointee(ToolsSnapshot::empty()),
}
}

Expand All @@ -314,11 +373,11 @@ impl ToolRegistry {
}

pub fn get(&self, name: &str) -> Option<RegisteredTool> {
self.tools.load().iter().find(|t| t.name() == name).cloned()
self.tools.load().get(name)
}

pub fn has(&self, name: &str) -> bool {
self.tools.load().iter().any(|t| t.name() == name)
self.tools.load().has(name)
}

/// Register a tool with the registry.
Expand All @@ -332,19 +391,18 @@ impl ToolRegistry {
let mut conflict = None;
self.tools.rcu(|current| {
conflict = None;
if let Some(existing) = current.iter().find(|t| t.name() == name) {
if let Some(existing) = current.get(&name) {
conflict = Some(existing.source.as_log_field().into_owned());
return Vec::clone(current);
return Arc::clone(current);
}
let mut next = Vec::with_capacity(current.len() + 1);
next.extend(current.iter().cloned());
next.push(RegisteredTool {
let mut next_tools = current.tools.clone();
next_tools.push(RegisteredTool {
tool: Arc::clone(tool),
source: source.clone(),
defer_loading,
namespace: namespace.clone(),
});
next
Arc::new(ToolsSnapshot::from_tools(next_tools))
});
if let Some(existing) = conflict {
return Err(RegistryError::NameConflict { name, existing });
Expand All @@ -365,24 +423,34 @@ impl ToolRegistry {
let mut conflict = None;
self.tools.rcu(|current| {
conflict = None;
let mut next = Vec::clone(current);
let mut next_tools = current.tools.clone();
let mut new_sources: HashMap<String, ToolSource> =
HashMap::with_capacity(entries.len());
for (tool, source) in &entries {
let name = tool.name();
if let Some(existing) = next.iter().find(|t| t.name() == name) {
let name = tool.name().to_owned();
if let Some(existing_source) = new_sources.get(&name) {
conflict = Some(RegistryError::NameConflict {
name,
existing: existing_source.as_log_field().into_owned(),
});
return Arc::clone(current);
}
if let Some(existing) = current.get(&name) {
conflict = Some(RegistryError::NameConflict {
name: name.to_owned(),
name,
existing: existing.source.as_log_field().into_owned(),
});
return Vec::clone(current);
return Arc::clone(current);
}
next.push(RegisteredTool {
new_sources.insert(name, source.clone());
next_tools.push(RegisteredTool {
tool: Arc::clone(tool),
source: source.clone(),
defer_loading: tool.defer_loading(),
namespace: tool.namespace().map(Arc::from).clone(),
});
}
next
Arc::new(ToolsSnapshot::from_tools(next_tools))
});
if let Some(e) = conflict {
return Err(e);
Expand All @@ -392,13 +460,14 @@ impl ToolRegistry {

pub fn clear_mcp_server(&self, server: &str) {
self.tools.rcu(|current| {
current
let filtered: Vec<_> = current
.iter()
.filter(
|t| !matches!(&t.source, ToolSource::Mcp { server: s } if s.as_ref() == server),
)
.cloned()
.collect::<Vec<_>>()
.collect();
Arc::new(ToolsSnapshot::from_tools(filtered))
});
}

Expand All @@ -414,30 +483,40 @@ impl ToolRegistry {
let mut conflict = None;
self.tools.rcu(|current| {
conflict = None;
let mut next: Vec<RegisteredTool> = current
let mut next_tools: Vec<RegisteredTool> = current
.iter()
.filter(
|t| !matches!(&t.source, ToolSource::Lua { plugin: p } if p.as_ref() == plugin),
)
.cloned()
.collect();
let mut new_sources: HashMap<String, ToolSource> =
HashMap::with_capacity(new_entries.len());
for (tool, source) in new_entries {
let name = tool.name();
if let Some(existing) = next.iter().find(|t| t.name() == name) {
let name = tool.name().to_owned();
if let Some(existing_source) = new_sources.get(&name) {
conflict = Some(RegistryError::NameConflict {
name,
existing: existing_source.as_log_field().into_owned(),
});
return Arc::clone(current);
}
if let Some(existing) = next_tools.iter().find(|t| t.name() == name) {
conflict = Some(RegistryError::NameConflict {
name: name.to_owned(),
name,
existing: existing.source.as_log_field().into_owned(),
});
return Vec::clone(current);
return Arc::clone(current);
}
next.push(RegisteredTool {
new_sources.insert(name, source.clone());
next_tools.push(RegisteredTool {
tool: Arc::clone(tool),
source: source.clone(),
defer_loading: tool.defer_loading(),
namespace: tool.namespace().map(Arc::from).clone(),
});
}
next
Arc::new(ToolsSnapshot::from_tools(next_tools))
});
if let Some(e) = conflict {
return Err(e);
Expand All @@ -447,23 +526,25 @@ impl ToolRegistry {

pub fn clear_lua(&self) {
self.tools.rcu(|current| {
current
let filtered: Vec<_> = current
.iter()
.filter(|t| !matches!(t.source, ToolSource::Lua { .. }))
.cloned()
.collect::<Vec<_>>()
.collect();
Arc::new(ToolsSnapshot::from_tools(filtered))
});
}

pub fn clear_plugin(&self, plugin: &str) {
self.tools.rcu(|current| {
current
let filtered: Vec<_> = current
.iter()
.filter(
|t| !matches!(&t.source, ToolSource::Lua { plugin: p } if p.as_ref() == plugin),
)
.cloned()
.collect::<Vec<_>>()
.collect();
Arc::new(ToolsSnapshot::from_tools(filtered))
});
}

Expand Down Expand Up @@ -622,12 +703,12 @@ impl ToolRegistry {
/// `Arc::ptr_eq` against a previously returned `Arc` detects the change.
/// Holding the old `Arc` keeps its allocation alive, so its address can't
/// be recycled by a future registration (ABA-safe).
pub fn snapshot_arc(&self) -> Arc<Vec<RegisteredTool>> {
pub fn snapshot_arc(&self) -> Arc<ToolsSnapshot> {
self.tools.load_full()
}
}

pub struct RegistrySnapshot(Arc<Vec<RegisteredTool>>);
pub struct RegistrySnapshot(Arc<ToolsSnapshot>);

impl<'a> IntoIterator for &'a RegistrySnapshot {
type Item = &'a RegisteredTool;
Expand Down
Loading
Loading