fix(cli): doctor ledger lock, catalog timeout and host check; sibling watcher starts at the log end

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

crates/tole-cli/src/doctor.rs

@@ -22,14 +22,20 @@ 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};
 
 /// 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
+    /// (`<ledger>.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<Ledger, String> {
+        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<std::fs::File, String> {
+    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);
+    }
 }

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(

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).

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) => {

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<Vec<String>, 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<Vec<String>, String> {
+    let agent = OpenAiProvider::build_agent(timeout);
     let url = format!("{}/models", base_url.trim_end_matches('/'));
     let mut res = agent
         .get(&url)

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<String>, log: impl Into<PathBuf>) -> std::io::Result<Self> {
+        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);
+    }
 }