Skip to content

fix(file): dash-safe args, real error status, symlink delete, streaming read #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sebyx07 merged 2 commits into from
Jul 2, 2026
+366 −155
Merged

fix(file): dash-safe args, real error status, symlink delete, streaming read #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3f60526
fix(file): dash-safe args, real error status, symlink delete, streami…
sebyx07 Jul 2, 2026
6c6edf5
fix(file): bound per-line read, thiserror ShError, split tests
sebyx07 Jul 2, 2026
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
290 changes: 135 additions & 155 deletions src/tools/files.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,22 @@ use tokio::{fs, io::AsyncWriteExt};
/// tells the agent to narrow the path/pattern.
const MAX_SHELL_OUTPUT_BYTES: usize = 64 * 1024;

/// Largest single line accumulated while streaming a `read`. `read_until` would
/// buffer a whole line before any trimming, so a no-newline multi-GB file could OOM
/// the service. We keep at most this many bytes per line and drop the rest to the
/// next newline — the agent still sees the line's head, memory stays bounded.
const MAX_READ_LINE_BYTES: usize = 64 * 1024;

/// Read a file, paginated by line AND bounded by bytes (via the shared job-log
/// paginator) so neither a huge file nor a single pathological line can flood the
/// agent context.
///
/// Streams line by line, holding at most the requested window (plus the line in
/// flight) — never the whole file. Slurping first and paginating after bounded
/// the reply but not server memory: paging 200 lines of a multi-GB log would
/// have materialized all of it and could OOM the service.
pub async fn read(path: &str, cursor: usize, limit: usize) -> Result<String, String> {
use tokio::io::{AsyncBufReadExt, BufReader};
// A directory can't be read as text: surface a useful redirect to `list`
// instead of a raw "Is a directory" errno.
match fs::metadata(path).await {
Expand All @@ -24,25 +36,77 @@ pub async fn read(path: &str, cursor: usize, limit: usize) -> Result<String, Str
Ok(_) => {}
Err(e) => return Err(e.to_string()),
}
// Binary-safe: replace non-UTF-8 bytes with U+FFFD rather than hard-erroring.
let bytes = fs::read(path).await.map_err(|e| e.to_string())?;
let owned = String::from_utf8_lossy(&bytes).into_owned();
let lines: Vec<&str> = owned.lines().collect();
let limit = limit.max(1);
let file = fs::File::open(path).await.map_err(|e| e.to_string())?;
let mut reader = BufReader::new(file);
let mut total = 0usize;
let mut window: Vec<String> = Vec::new();
// The line in flight, capped at MAX_READ_LINE_BYTES: fill_buf/consume lets us
// stop accumulating a pathological no-newline line instead of read_until slurping
// the whole file into one buffer.
let mut line: Vec<u8> = Vec::new();
loop {
let chunk = reader.fill_buf().await.map_err(|e| e.to_string())?;
if chunk.is_empty() {
// EOF: emit a final line that had no terminating newline.
if !line.is_empty() {
if total >= cursor && window.len() < limit {
window.push(String::from_utf8_lossy(&line).into_owned());
}
total += 1;
}
break;
}
match chunk.iter().position(|&b| b == b'\n') {
Some(nl) => {
append_capped(&mut line, &chunk[..nl]);
reader.consume(nl + 1);
if line.last() == Some(&b'\r') {
line.pop();
}
if total >= cursor && window.len() < limit {
// Binary-safe: replace non-UTF-8 bytes with U+FFFD, like `lines()`.
window.push(String::from_utf8_lossy(&line).into_owned());
}
total += 1;
line.clear();
}
None => {
let n = chunk.len();
append_capped(&mut line, chunk);
reader.consume(n);
}
}
}
// Forward pagination from the top (cursor 0 = first line): a file is read
// start-to-end. (`job poll` instead reads newest-first — a live log's latest
// output matters most.) Same byte/line ceilings via `paginate`.
let page = crate::jobs::paginate(&lines, cursor, limit);
// output matters most.) Same byte/line ceilings via `paginate`, applied to
// the pre-cut window (so the byte cap can shorten the page further).
let refs: Vec<&str> = window.iter().map(String::as_str).collect();
let page = crate::jobs::paginate(&refs, 0, limit);
let body = page.lines.join("\n");
if page.has_more {
let next = cursor.min(total) + page.lines.len();
if next < total {
Ok(format!(
"{body}\n[lines {cursor}..{} of {}; next_cursor={}]",
page.next_cursor, page.total_lines, page.next_cursor
"{body}\n[lines {cursor}..{next} of {total}; next_cursor={next}]"
))
} else {
Ok(body)
}
}

/// Append `bytes` to the in-flight line without letting it grow past
/// `MAX_READ_LINE_BYTES`. Overflow bytes are dropped (the line is already longer
/// than any page will show), so one pathological no-newline line can't grow the
/// buffer without bound.
fn append_capped(line: &mut Vec<u8>, bytes: &[u8]) {
let room = MAX_READ_LINE_BYTES.saturating_sub(line.len());
if room == 0 {
return;
}
line.extend_from_slice(&bytes[..room.min(bytes.len())]);
}

pub async fn write(path: &str, content: &str) -> Result<String, String> {
ensure_parent(path).await?;
fs::write(path, content).await.map_err(|e| e.to_string())?;
Expand All @@ -66,7 +130,13 @@ pub async fn append(path: &str, content: &str) -> Result<String, String> {
}

pub async fn delete(path: &str) -> Result<String, String> {
let meta = fs::metadata(path).await.map_err(|e| e.to_string())?;
// symlink_metadata, not metadata: a symlink must be unlinked, never followed.
// Following classified a dir-symlink as a directory (remove_dir_all refuses
// the top-level link) and made a dangling symlink undeletable (ENOENT on the
// stat before any removal was attempted).
let meta = fs::symlink_metadata(path)
.await
.map_err(|e| e.to_string())?;
let r = if meta.is_dir() {
fs::remove_dir_all(path).await
} else {
Expand Down Expand Up @@ -99,28 +169,76 @@ async fn ensure_parent(path: &str) -> Result<(), String> {

pub async fn list(path: &str, recursive: bool) -> Result<String, String> {
if recursive {
sh("find", &[path]).await
// `find` has no `--`; anchor a leading-dash relative path with `./` so
// it can't be parsed as an expression.
let path = if path.starts_with('-') {
format!("./{path}")
} else {
path.to_string()
};
sh("find", &[&path]).await.map_err(|e| e.to_string())
} else {
sh("ls", &["-la", path]).await
sh("ls", &["-la", "--", path])
.await
.map_err(|e| e.to_string())
}
}

pub async fn grep(pattern: &str, path: &str, recursive: bool) -> Result<String, String> {
let flag = if recursive { "-rn" } else { "-n" };
sh("grep", &[flag, pattern, path]).await
// `--` so a pattern like `->` or `-r` is a pattern, not an option: without
// it, grepping Rust code for `->` errored, and `-r` silently recursed with
// the *path* as the pattern — wrong results, not even an error.
match sh("grep", &[flag, "--", pattern, path]).await {
Ok(s) => Ok(s),
// Exit 1 is grep's "no line matched" — a legitimate empty result. The
// marker keeps it distinguishable from matching an empty line.
Err(ShError::Status { code: 1, out }) if out.is_empty() => {
Ok("[grep: no matches]".to_string())
}
Err(e) => Err(e.to_string()),
}
}

async fn sh(prog: &str, args: &[&str]) -> Result<String, String> {
/// A shelled-out command that didn't produce a clean result. Exit status and
/// combined output stay separate so callers can special-case a status (grep's
/// exit-1-means-no-matches) without parsing message text.
#[derive(Debug, thiserror::Error)]
enum ShError {
#[error("{0}")]
Spawn(String),
#[error("exit status {code}: {out}")]
Status { code: i32, out: String },
// A signal death carries no exit code: its own variant keeps it distinct from a
// real status so the grep-exit-1 path can never accidentally match it.
#[error("killed by signal: {out}")]
Signal { out: String },
}

/// Run `prog` and return its combined stdout+stderr — as `Ok` only on exit 0.
/// A failed command must surface as an error, not as a success whose body
/// happens to contain `ls: cannot access ...`.
async fn sh(prog: &str, args: &[&str]) -> Result<String, ShError> {
let out = tokio::process::Command::new(prog)
.args(args)
.output()
.await
.map_err(|e| e.to_string())?;
.map_err(|e| ShError::Spawn(e.to_string()))?;
let mut s = String::from_utf8_lossy(&out.stdout).into_owned();
if !out.stderr.is_empty() {
s.push_str(&String::from_utf8_lossy(&out.stderr));
}
Ok(cap_bytes(s, MAX_SHELL_OUTPUT_BYTES))
let s = cap_bytes(s, MAX_SHELL_OUTPUT_BYTES);
if out.status.success() {
Ok(s)
} else {
// No exit code means a signal killed the process — model it as its own
// variant rather than folding it into a sentinel status.
Err(match out.status.code() {
Some(code) => ShError::Status { code, out: s },
None => ShError::Signal { out: s },
})
}
}

/// Bound a non-paginated listing to `max` bytes, cut on a UTF-8 boundary, with a
Expand All @@ -142,142 +260,4 @@ fn cap_bytes(mut s: String, max: usize) -> String {
}

#[cfg(test)]
mod tests {
use super::*;

#[tokio::test]
async fn write_read_paginate_append_move_delete() {
let dir = tempfile::tempdir().unwrap();
let a = dir.path().join("a.txt");
let a = a.to_str().unwrap();

write(a, "l1\nl2\nl3").await.unwrap();
let page = read(a, 0, 2).await.unwrap();
assert!(page.contains("l1") && page.contains("l2") && !page.contains("l3"));
assert!(page.contains("next_cursor=2"));

append(a, "\nl4").await.unwrap();
assert!(read(a, 0, 100).await.unwrap().contains("l4"));

let b = dir.path().join("b.txt");
let b = b.to_str().unwrap();
rename(a, b).await.unwrap();
assert!(read(a, 0, 10).await.is_err());

delete(b).await.unwrap();
assert!(read(b, 0, 10).await.is_err());
}

#[tokio::test]
async fn write_creates_missing_parent_dirs() {
let dir = tempfile::tempdir().unwrap();
// Two levels that don't exist yet — write must `mkdir -p` them.
let nested = dir.path().join("a/b/c.txt");
let nested = nested.to_str().unwrap();
write(nested, "hi").await.unwrap();
assert_eq!(read(nested, 0, 10).await.unwrap(), "hi");

// append to a fresh path under a new dir works too.
let ap = dir.path().join("x/y/z.log");
let ap = ap.to_str().unwrap();
append(ap, "one\n").await.unwrap();
assert!(read(ap, 0, 10).await.unwrap().contains("one"));
}

#[tokio::test]
async fn read_on_directory_redirects_to_list() {
let dir = tempfile::tempdir().unwrap();
let err = read(dir.path().to_str().unwrap(), 0, 10).await.unwrap_err();
assert!(
err.contains("is a directory") && err.contains("list"),
"dir read should redirect to list: {err}"
);
}

#[tokio::test]
async fn grep_finds_match() {
let dir = tempfile::tempdir().unwrap();
let c = dir.path().join("c.txt");
let c = c.to_str().unwrap();
write(c, "alpha\nbeta\ngamma").await.unwrap();
assert!(grep("beta", c, false).await.unwrap().contains("beta"));
}

#[tokio::test]
async fn list_non_recursive_shows_top_level_only() {
let dir = tempfile::tempdir().unwrap();
let sub = dir.path().join("sub");
tokio::fs::create_dir(&sub).await.unwrap();
let f_top = dir.path().join("top.txt");
let f_nested = sub.join("nested.txt");
write(f_top.to_str().unwrap(), "top").await.unwrap();
write(f_nested.to_str().unwrap(), "nested").await.unwrap();

let out = list(dir.path().to_str().unwrap(), false).await.unwrap();
assert!(out.contains("top.txt"), "should list top-level file: {out}");
assert!(out.contains("sub"), "should list sub dir: {out}");
assert!(!out.contains("nested.txt"), "should not recurse: {out}");
}

#[tokio::test]
async fn list_recursive_finds_nested_files() {
let dir = tempfile::tempdir().unwrap();
let sub = dir.path().join("sub");
tokio::fs::create_dir(&sub).await.unwrap();
let f_nested = sub.join("deep.txt");
write(f_nested.to_str().unwrap(), "deep").await.unwrap();

let out = list(dir.path().to_str().unwrap(), true).await.unwrap();
assert!(
out.contains("deep.txt"),
"recursive find should reach nested file: {out}"
);
}

#[tokio::test]
async fn delete_directory_removes_entire_tree() {
let dir = tempfile::tempdir().unwrap();
let sub = dir.path().join("to_delete");
tokio::fs::create_dir(&sub).await.unwrap();
let f = sub.join("file.txt");
write(f.to_str().unwrap(), "content").await.unwrap();

delete(sub.to_str().unwrap()).await.unwrap();
assert!(
!sub.exists(),
"directory and its contents should be removed"
);
}

#[tokio::test]
async fn grep_recursive_finds_match_in_subdirs() {
let dir = tempfile::tempdir().unwrap();
let sub = dir.path().join("sub");
tokio::fs::create_dir(&sub).await.unwrap();
let f = sub.join("d.txt");
write(f.to_str().unwrap(), "alpha\nbeta\ngamma")
.await
.unwrap();

let out = grep("beta", dir.path().to_str().unwrap(), true)
.await
.unwrap();
assert!(
out.contains("beta"),
"recursive grep should find pattern in subdir: {out}"
);
}

#[tokio::test]
async fn binary_file_reads_as_lossy_utf8_not_error() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("bin.dat");
// Write bytes that are not valid UTF-8.
tokio::fs::write(&p, b"hello\xff\xfeworld\n").await.unwrap();
let result = read(p.to_str().unwrap(), 0, 100).await;
assert!(result.is_ok(), "binary read should not hard-error");
let content = result.unwrap();
assert!(content.contains("hello"), "ASCII prefix should survive");
assert!(content.contains("world"), "ASCII suffix should survive");
}
}
mod tests;
Loading
Loading