diff --git a/CHANGELOG.md b/CHANGELOG.md index bb13e3a..dd2f468 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Per-user-turn token stats (#379): `tole status` lists input / output / cache-read / cache-write / reasoning tokens per turn and the status JSON (`GET /sessions/{id}/status`, `tole_session_status`) gains an additive `usage_report.turns` array. Derived from the durable usage ledger, so old sessions read the same way and nothing is stored twice; a value no step reported is `n/a` / `null`, never 0, and no elapsed time is invented. ### Fixed +- `tole doctor`: concurrent runs no longer overwrite each other's ledger entry (load/record/save now happen under an exclusive lock on `doctor-ledger.json.lock`); the catalog tier's `GET /models` gets its own 20 s timeout instead of the ACP picker's 3 s convenience timeout, so a slow but healthy gateway is not reported as a blocker; and the offline tier now requires a host in the base URL (`https://?x` or `https://u:p@/v1` no longer pass). +- Sibling change notices: a child's watcher starts at the end of the shared touch log instead of folding every earlier child's history on start-up. +- The optional `web_search` tool is registered with `if let` instead of an `is_some()` check followed by `unwrap()`. - `tole serve --transport mcp` is hardened to parity with the REST transport: the bearer token is compared in constant time and its scheme is case-insensitive; failed authentication is rate-limited per source IP (10 per minute, then `429`, same policy as REST); request bodies over 1 MiB are refused with `413` (declared or streamed, never buffered whole); a transient `accept()` error is logged and no longer ends the server and every session; and the no-op `timeout()` around the auth check is gone. The token comparison, header matching and limiter now live in one shared `auth` module used by both transports. ### Changed diff --git a/crates/tole-cli/src/doctor.rs b/crates/tole-cli/src/doctor.rs index fd41f33..bef9165 100644 --- a/crates/tole-cli/src/doctor.rs +++ b/crates/tole-cli/src/doctor.rs @@ -22,7 +22,7 @@ use serde_json::{json, Value}; use std::path::{Path, PathBuf}; use std::time::Duration; use tole_core::approval::{Approver, ToolRequest, Verdict}; -use tole_core::openai::{fetch_model_ids, OpenAiConfig, OpenAiProvider, ENV_PREFIXES}; +use tole_core::openai::{fetch_model_ids_with_timeout, OpenAiConfig, OpenAiProvider, ENV_PREFIXES}; use tole_core::storage::JsonlStorage; use tole_core::tool::{Risk, Tool, ToolRegistry}; use tole_core::turn::{run_turn, TurnOutcome}; @@ -30,6 +30,12 @@ use tole_core::turn::{run_turn, TurnOutcome}; /// Ledger file name under the tole data dir. pub const LEDGER_FILE: &str = "doctor-ledger.json"; +/// Timeout of the catalog tier's `GET /models`. Generous on purpose: here +/// the probe IS the check, so a slow but healthy gateway must not be +/// reported as a blocker (the ACP model picker keeps its 3 s convenience +/// timeout). +const CATALOG_TIMEOUT: Duration = Duration::from_secs(20); + /// Per-call timeout of the `full` tier. const FULL_CALL_TIMEOUT: Duration = Duration::from_secs(60); @@ -141,10 +147,17 @@ fn base_url_ok(url: &str) -> bool { let rest = url .strip_prefix("https://") .or_else(|| url.strip_prefix("http://")); - match rest { - Some(r) => !r.is_empty() && !r.starts_with('/') && !url.chars().any(char::is_whitespace), - None => false, + let Some(rest) = rest else { + return false; + }; + if url.chars().any(char::is_whitespace) { + return false; } + // The authority ends at the first '/', '?' or '#'; after any userinfo + // there must be a host (`https://?x` or `https://u:p@/v1` has none). + let authority = rest.split(['/', '?', '#']).next().unwrap_or(""); + let host = authority.rsplit('@').next().unwrap_or(""); + !host.is_empty() && !host.starts_with(':') } struct DenyAll; @@ -431,7 +444,7 @@ fn diagnose_raw( } // ---- catalog ---- - match fetch_model_ids(&cfg.base_url, &cfg.api_key) { + match fetch_model_ids_with_timeout(&cfg.base_url, &cfg.api_key, CATALOG_TIMEOUT) { Err(e) => { out.fail( Tier::Catalog, @@ -533,6 +546,18 @@ impl Ledger { std::fs::rename(&tmp, path).map_err(|e| format!("cannot replace {path:?}: {e}")) } + /// Load, change and save the ledger under an exclusive advisory lock + /// (`.lock`), so concurrent `tole doctor` runs (several + /// providers checked in parallel) cannot overwrite each other's entry + /// or race on the temp file. Returns the saved ledger. + pub fn update(path: &Path, change: impl FnOnce(&mut Ledger)) -> Result { + let _lock = lock_file(&path.with_extension("json.lock"))?; + let mut l = Ledger::load(path)?; + change(&mut l); + l.save(path)?; + Ok(l) + } + /// Record the latest run for `provider`+`model` (replaces the previous /// entry for the pair; the latest evidence wins). pub fn record(&mut self, provider: &str, model: &str, outcome: &Outcome, at_ms: u64) { @@ -579,6 +604,37 @@ impl Ledger { } } +/// Open (creating, owner-only) and exclusively lock `path`; the lock is +/// released when the returned file is dropped. Advisory and cooperative, +/// like the agents-root lock. Non-unix hosts get no lock (single-writer). +fn lock_file(path: &Path) -> Result { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).map_err(|e| format!("cannot create {parent:?}: {e}"))?; + } + let mut opts = std::fs::OpenOptions::new(); + opts.create(true).truncate(false).write(true).read(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + opts.mode(0o600); + } + let f = opts + .open(path) + .map_err(|e| format!("cannot open the ledger lock {path:?}: {e}"))?; + #[cfg(unix)] + { + use std::os::fd::AsRawFd; + // SAFETY: a valid, owned descriptor for the life of `f`. + if unsafe { libc::flock(f.as_raw_fd(), libc::LOCK_EX) } != 0 { + return Err(format!( + "cannot lock the ledger: {}", + std::io::Error::last_os_error() + )); + } + } + Ok(f) +} + /// Write `bytes` readable by the owner only (0600 on Unix). A stale file at /// `path` is removed first and the new one is created exclusively /// (`O_EXCL`), so a pre-planted file or symlink is never written through @@ -628,10 +684,9 @@ pub fn run_cli( // Only a configured provider has a ledger identity. if let (Some(cfg), false) = (cfg, no_ledger) { let saved = ledger_path().and_then(|p| { - let mut l = Ledger::load(&p)?; - l.record(&redact_url(&cfg.base_url), &cfg.model, &outcome, now_ms()); - l.save(&p)?; - Ok(l) + Ledger::update(&p, |l| { + l.record(&redact_url(&cfg.base_url), &cfg.model, &outcome, now_ms()) + }) }); match saved { Ok(l) => println!("{}", l.summary()), @@ -994,4 +1049,54 @@ mod tests { assert_eq!(Tier::parse("nope"), None); assert!(Tier::Offline < Tier::Catalog && Tier::Catalog < Tier::Full); } + + #[test] + fn base_url_needs_a_host() { + for ok in [ + "https://api.example.com/v1", + "http://127.0.0.1:8080", + "https://u:p@host/v1", + "https://host?x=1", + ] { + assert!(base_url_ok(ok), "{ok}"); + } + for bad in [ + "https://?x", + "https://#f", + "https:///v1", + "https://u:p@/v1", + "https://:8080/v1", + "ftp://host", + "https://ho st/v1", + "https://", + ] { + assert!(!base_url_ok(bad), "{bad}"); + } + } + + #[test] + fn concurrent_ledger_updates_keep_every_entry() { + let dir = scratch_dir("ledger-concurrent"); + let path = dir.join(LEDGER_FILE); + let threads: Vec<_> = (0..8) + .map(|i| { + let path = path.clone(); + std::thread::spawn(move || { + let o = Outcome { + reached: Some(Tier::Catalog), + blocker: None, + lines: vec![], + }; + Ledger::update(&path, |l| l.record(&format!("https://p{i}/v1"), "m", &o, 1)) + .unwrap(); + }) + }) + .collect(); + for t in threads { + t.join().unwrap(); + } + let l = Ledger::load(&path).unwrap(); + assert_eq!(l.entries.len(), 8, "no update was lost: {:?}", l.entries); + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/crates/tole-cli/src/main.rs b/crates/tole-cli/src/main.rs index 2b08461..209d22f 100644 --- a/crates/tole-cli/src/main.rs +++ b/crates/tole-cli/src/main.rs @@ -1560,8 +1560,8 @@ fn build_registry( // with TOLE_WEB_SEARCH_URL. reg.register(Box::new(tole_core::web::WebFetchTool)) .map_err(|e| anyhow::anyhow!("registering web_fetch: {e}"))?; - if tole_core::web::WebSearchTool::from_env().is_some() { - reg.register(Box::new(tole_core::web::WebSearchTool::from_env().unwrap())) + if let Some(search) = tole_core::web::WebSearchTool::from_env() { + reg.register(Box::new(search)) .map_err(|e| anyhow::anyhow!("registering web_search: {e}"))?; } // Write tools: gated per call. The jail root is the workspace. @@ -1715,8 +1715,8 @@ fn build_server_registry( // with TOLE_WEB_SEARCH_URL. reg.register(Box::new(tole_core::web::WebFetchTool)) .map_err(|e| anyhow::anyhow!("registering web_fetch: {e}"))?; - if tole_core::web::WebSearchTool::from_env().is_some() { - reg.register(Box::new(tole_core::web::WebSearchTool::from_env().unwrap())) + if let Some(search) = tole_core::web::WebSearchTool::from_env() { + reg.register(Box::new(search)) .map_err(|e| anyhow::anyhow!("registering web_search: {e}"))?; } reg.register(tole_cli::sibling::track( diff --git a/crates/tole-cli/src/session_host.rs b/crates/tole-cli/src/session_host.rs index ceafe16..2befdc2 100644 --- a/crates/tole-cli/src/session_host.rs +++ b/crates/tole-cli/src/session_host.rs @@ -355,8 +355,8 @@ pub fn open_session( // when a backend is configured (TOLE_WEB_SEARCH_URL) — probe-first: // no backend, no tool, never a phantom. reg.register(Box::new(tole_core::web::WebFetchTool))?; - if tole_core::web::WebSearchTool::from_env().is_some() { - reg.register(Box::new(tole_core::web::WebSearchTool::from_env().unwrap()))?; + if let Some(search) = tole_core::web::WebSearchTool::from_env() { + reg.register(Box::new(search))?; } // Plan publishing (issue #196 phase 4): model-facing tool, present // only when the transport has a client that renders plans (ACP). diff --git a/crates/tole-cli/src/sibling.rs b/crates/tole-cli/src/sibling.rs index 65b54c1..982ac87 100644 --- a/crates/tole-cli/src/sibling.rs +++ b/crates/tole-cli/src/sibling.rs @@ -73,9 +73,19 @@ impl Drop for Watcher { pub fn start_watcher(tracking: &Tracking, cancel: &CancelToken, every: Duration) -> Watcher { let stop = Arc::new(AtomicBool::new(false)); let (flag, token) = (Arc::clone(&stop), cancel.clone()); - let mut watcher = SiblingWatcher::new(tracking.agent.clone(), tracking.log.clone()); + // Start at the end of the shared log: its history predates this child. + let (mut watcher, mut reported) = + match SiblingWatcher::from_end(tracking.agent.clone(), tracking.log.clone()) { + Ok(w) => (w, false), + Err(e) => { + eprintln!("tole: sibling tracking: cannot read the touch log: {e}"); + ( + SiblingWatcher::new(tracking.agent.clone(), tracking.log.clone()), + true, + ) + } + }; std::thread::spawn(move || { - let mut reported = false; while !flag.load(Ordering::SeqCst) { match watcher.poll() { Ok(notices) => { diff --git a/crates/tole-core/src/openai.rs b/crates/tole-core/src/openai.rs index 7c28c5c..d8dd9e0 100644 --- a/crates/tole-core/src/openai.rs +++ b/crates/tole-core/src/openai.rs @@ -747,7 +747,18 @@ fn scrub(msg: &str, secret: &str) -> String { /// model picker), never a correctness dependency — callers degrade to /// "no picker" on any error, per the probe-first contract. pub fn fetch_model_ids(base_url: &str, api_key: &str) -> Result, String> { - let agent = OpenAiProvider::build_agent(Duration::from_secs(3)); + fetch_model_ids_with_timeout(base_url, api_key, Duration::from_secs(3)) +} + +/// [`fetch_model_ids`] with a caller-chosen timeout, for callers where the +/// probe IS the check (the provider doctor) rather than a UI convenience: a +/// slow but healthy gateway must not be reported as broken. +pub fn fetch_model_ids_with_timeout( + base_url: &str, + api_key: &str, + timeout: Duration, +) -> Result, String> { + let agent = OpenAiProvider::build_agent(timeout); let url = format!("{}/models", base_url.trim_end_matches('/')); let mut res = agent .get(&url) diff --git a/crates/tole-core/src/sibling_watch.rs b/crates/tole-core/src/sibling_watch.rs index 1f1c042..66c407d 100644 --- a/crates/tole-core/src/sibling_watch.rs +++ b/crates/tole-core/src/sibling_watch.rs @@ -238,6 +238,34 @@ impl SiblingWatcher { } } + /// A watcher that starts at the current end of the log (at a line + /// boundary) instead of its beginning. The log is shared by every child + /// a parent ever spawned and is never truncated, so folding its whole + /// history on start-up costs O(lifetime of the agents root) for nothing: + /// changes made before this agent looked at a file are not news anyway. + pub fn from_end(agent: impl Into, log: impl Into) -> std::io::Result { + let mut w = Self::new(agent, log); + let mut f = match std::fs::File::open(&w.log) { + Ok(f) => f, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(w), + Err(e) => return Err(e), + }; + let len = f.metadata()?.len(); + // Only the tail is read: a line is at most MAX_LINE bytes, so the + // last complete line ends inside the final 2 * MAX_LINE bytes. + let start = len.saturating_sub(2 * MAX_LINE as u64); + f.seek(SeekFrom::Start(start))?; + let mut tail = Vec::new(); + f.read_to_end(&mut tail)?; + w.offset = match tail.iter().rposition(|&b| b == b'\n') { + Some(i) => start + i as u64 + 1, + // No newline in the tail: a half-written first line; start + // before it so it is read once complete. + None => start, + }; + Ok(w) + } + /// Read what was appended since the last poll and return the notices. /// Only complete lines are consumed (a half-written tail waits for the /// next poll); unparseable lines are skipped. A missing log is empty. @@ -590,4 +618,54 @@ mod tests { assert!(t.contains("not an instruction")); let _ = std::fs::remove_dir_all(d); } + + #[test] + fn from_end_skips_history_but_sees_what_comes_after() { + let (a, b, _wa, _wb, d) = pair("fromend"); + let log = d.join("tole-agents").join("touch.log"); + // History: I read x, a sibling changed it (before this watcher). + a.record(Op::Read, "x").unwrap(); + b.record(Op::Write, "x").unwrap(); + let mut w = SiblingWatcher::from_end("a", &log).unwrap(); + assert!(w.poll().unwrap().is_empty(), "history is not replayed"); + // New activity after start-up is seen as usual. + a.record(Op::Read, "y").unwrap(); + b.record(Op::Write, "y").unwrap(); + assert_eq!(w.poll().unwrap().len(), 1); + // A missing log is fine. + let none = SiblingWatcher::from_end("a", d.join("nope.log")).unwrap(); + assert_eq!(none.offset, 0); + let _ = std::fs::remove_dir_all(d); + } + + #[test] + fn from_end_waits_for_a_half_written_last_line() { + let (a, _b, _wa, _wb, d) = pair("fromend-partial"); + let log = d.join("tole-agents").join("touch.log"); + a.record(Op::Read, "z").unwrap(); + let line = format!( + "{}\n", + json!({"a": "b", "o": "w", "p": normalize(&d.join("ws"), "z")}) + ); + let (head, tail) = line.split_at(line.len() / 2); + std::fs::OpenOptions::new() + .append(true) + .open(&log) + .unwrap() + .write_all(head.as_bytes()) + .unwrap(); + let mut w = SiblingWatcher::from_end("a", &log).unwrap(); + std::fs::OpenOptions::new() + .append(true) + .open(&log) + .unwrap() + .write_all(tail.as_bytes()) + .unwrap(); + // The completed line is read (not skipped), even though it began + // before the watcher started; it is not news for `a` here because + // a's own read was history. + w.poll().unwrap(); + assert_eq!(w.offset, std::fs::metadata(&log).unwrap().len()); + let _ = std::fs::remove_dir_all(d); + } }