diff --git a/CHANGELOG.md b/CHANGELOG.md index ffa38d6..318f4a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added -- Sibling change notices, core (#380): `sibling_watch` tracks which files each child agent reads and writes in one shared append-only log (owner-only, bounded) and tells a child, once per stale path, when a sibling changed a file it had read or written; the `Tracked` tool decorator records `read_file` / `write_file` / `edit_file` / `delete_file` on success. Interrupts gain a `source`: a notice is delivered at the next safe point as an attributed entry (`source: notice`) and never withdraws output or supersedes a final answer. Wiring into `tole run` children follows separately. +- Sibling change notices (#380): `sibling_watch` tracks which files each child agent reads and writes in one shared append-only log (owner-only, bounded) and tells a child, once per stale path, when a sibling changed a file it had read or written; the `Tracked` tool decorator records `read_file` / `write_file` / `edit_file` / `delete_file` on success. Interrupts gain a `source`: a notice is delivered at the next safe point as an attributed entry (`source: notice`) and never withdraws output or supersedes a final answer. Children spawned by `agent_start` get `TOLE_AGENT_ID` and a shared `TOLE_AGENT_TOUCH_LOG`; in a child, `tole run` wraps its file tools and runs the watcher, so a sibling's change reaches the model at its next safe point. - `tole doctor` coverage ledger (#378): every run is recorded per provider + model (level reached, first blocker, fix command, timestamp) in an owner-only `doctor-ledger.json` (atomic write, a corrupt file is an error, never replaced); `--summary` prints `READY n-of-m` + the first blocker and its fix, `--no-ledger` skips recording. - Soft interrupts (#377): a bounded queue on the session's `CancelToken` that the turn loop drains at the safe point between a settled tool and the next provider call, committing each message as an attributed user-role entry (`interrupt: true`). An `urgent` interrupt (or any interrupt, for a final answer) withdraws an output that has not taken effect, so no tool call is left without its result, and records the drop durably. Host entry points: `POST /sessions/{id}/interrupt` (`tole serve`) and the `tole_session_interrupt` MCP tool (`409`/error when no turn is running). - `tole doctor` (#378): tiered provider diagnostics — `offline` (wiring), `catalog` (live `/models`, model must be served) and `full` (chat, streaming and a real tool-call loop) — each blocker carries its fix command, and keys / URL credentials are never printed. Exits non-zero when the requested tier is not reached; `--tier full` is the release-validation gate. diff --git a/README.md b/README.md index 47c5ab7..5069463 100644 --- a/README.md +++ b/README.md @@ -252,7 +252,7 @@ hard error. | `verify_package` | RO | crates.io / npm registry check before any install — hallucinated names get NOT FOUND + candidates, edit-distance-1 candidates get a typo-squat warning | | `load_skill` | RO | loads a discovered SKILL.md on demand (`--skill` pins one upfront; `--no-skills` disables) | | `gitea` | Write | Gitea counterpart of `gh` over the instance REST API — registers when `origin` is a Gitea remote AND `TOLE_GITEA_TOKEN`/`GITEA_TOKEN` is set; same six ops | -| `agent_start`, `agent_poll` | Write / Write | depth-1 child agents: spawn durable child sessions (structurally no grandchildren), results via per-child uteke mailboxes (ephemeral by default); `agent_poll` is Write because a settled poll consumes (cleans) the mailbox, and `--trust internal` allowlists it by name; `--agents-worktree` gives each child its own git worktree | +| `agent_start`, `agent_poll` | Write / Write | depth-1 child agents: spawn durable child sessions (structurally no grandchildren), results via per-child uteke mailboxes (ephemeral by default); `agent_poll` is Write because a settled poll consumes (cleans) the mailbox, and `--trust internal` allowlists it by name; `--agents-worktree` gives each child its own git worktree; siblings are told (#380) when another child changes a file they read or wrote, at their next safe point | | `systemone_decide` | RO | typed decisions (choice/score/noul + confidence) from a System One backend — active when `SYSTEMONE_API_KEY` is set; `SYSTEMONE_BASE_URL` picks the backend (hosted Jev default, self-hosted compatible) | | `job_start`, `job_poll` | Write / RO | detached long-running jobs with log tailing | | `cora_search` | RO | hybrid codebase search via `cora brain`; native fallback — skipped when the cora MCP surface is attached | diff --git a/crates/tole-cli/src/lib.rs b/crates/tole-cli/src/lib.rs index afbe20e..8e14581 100644 --- a/crates/tole-cli/src/lib.rs +++ b/crates/tole-cli/src/lib.rs @@ -10,6 +10,7 @@ pub mod config_trust; pub mod doctor; pub mod session_host; pub mod session_tools; +pub mod sibling; pub mod tools; pub mod trust; pub mod usage_report; diff --git a/crates/tole-cli/src/main.rs b/crates/tole-cli/src/main.rs index ccbf2f3..e9d6142 100644 --- a/crates/tole-cli/src/main.rs +++ b/crates/tole-cli/src/main.rs @@ -1471,6 +1471,8 @@ fn build_registry( #[cfg_attr(not(feature = "shell-tools"), allow(unused_variables))] let cwd = std::env::current_dir().context("resolving cwd")?; let file_root = resolve_workspace_root(workspace)?; + // Child agents (#380): present only when spawned by `agent_start`. + let tracking = tole_cli::sibling::tracking_from_env(&tole_cli::config_apply::real_env); // Uteke first-class (B4): recall (read) + document (write), behind // startup probing — a missing uteke binary degrades to a warning, // not phantom tools. @@ -1543,8 +1545,12 @@ fn build_registry( ))) .map_err(|e| anyhow::anyhow!("registering agent_poll: {e}"))?; } - reg.register(Box::new(ReadFileTool::new(file_root.clone()))) - .map_err(|e| anyhow::anyhow!("registering read_file: {e}"))?; + reg.register(tole_cli::sibling::track( + Box::new(ReadFileTool::new(file_root.clone())), + tracking.as_ref(), + &file_root, + )) + .map_err(|e| anyhow::anyhow!("registering read_file: {e}"))?; // Web tools (issue #215): probe-first — fetch always; search only // with TOLE_WEB_SEARCH_URL. reg.register(Box::new(tole_core::web::WebFetchTool)) @@ -1554,10 +1560,18 @@ fn build_registry( .map_err(|e| anyhow::anyhow!("registering web_search: {e}"))?; } // Write tools: gated per call. The jail root is the workspace. - reg.register(Box::new(WriteFileTool::new(file_root.clone()))) - .map_err(|e| anyhow::anyhow!("registering write_file: {e}"))?; - reg.register(Box::new(EditFileTool::new(file_root.clone()))) - .map_err(|e| anyhow::anyhow!("registering edit_file: {e}"))?; + reg.register(tole_cli::sibling::track( + Box::new(WriteFileTool::new(file_root.clone())), + tracking.as_ref(), + &file_root, + )) + .map_err(|e| anyhow::anyhow!("registering write_file: {e}"))?; + reg.register(tole_cli::sibling::track( + Box::new(EditFileTool::new(file_root.clone())), + tracking.as_ref(), + &file_root, + )) + .map_err(|e| anyhow::anyhow!("registering edit_file: {e}"))?; #[cfg(feature = "shell-tools")] { register_gh(&mut reg, &cwd)?; @@ -1569,8 +1583,12 @@ fn build_registry( .map_err(|e| anyhow::anyhow!("registering git: {e}"))?; // Destructive tools: interactive-approver-only registration; every // call prompts — allowlists and --yes never apply (PRD risk table). - reg.register(Box::new(DeleteFileTool::new(file_root))) - .map_err(|e| anyhow::anyhow!("registering delete_file: {e}"))?; + reg.register(tole_cli::sibling::track( + Box::new(DeleteFileTool::new(file_root.clone())), + tracking.as_ref(), + &file_root, + )) + .map_err(|e| anyhow::anyhow!("registering delete_file: {e}"))?; // MCP servers (#74): registered last so a slow server never blocks // native tool availability; per-call approval applies as usual. #[cfg(feature = "mcp")] @@ -1624,6 +1642,8 @@ fn build_server_registry( ToolRegistry::with_approver(AllowlistApprover::allow_only(allow_patterns.to_vec())); let cwd = std::env::current_dir().context("resolving cwd")?; let file_root = resolve_workspace_root(workspace)?; + // Child agents (#380): present only when spawned by `agent_start`. + let tracking = tole_cli::sibling::tracking_from_env(&tole_cli::config_apply::real_env); let count = |reg: &ToolRegistry| reg.specs().len(); #[cfg(feature = "shell-tools")] @@ -1675,8 +1695,12 @@ fn build_server_registry( reg.register(Box::new(t)) .map_err(|e| anyhow::anyhow!("registering systemone_decide: {e}"))?; } - reg.register(Box::new(ReadFileTool::new(file_root.clone()))) - .map_err(|e| anyhow::anyhow!("registering read_file: {e}"))?; + reg.register(tole_cli::sibling::track( + Box::new(ReadFileTool::new(file_root.clone())), + tracking.as_ref(), + &file_root, + )) + .map_err(|e| anyhow::anyhow!("registering read_file: {e}"))?; // Web tools (issue #215): probe-first — fetch always; search only // with TOLE_WEB_SEARCH_URL. reg.register(Box::new(tole_core::web::WebFetchTool)) @@ -1685,10 +1709,18 @@ fn build_server_registry( reg.register(Box::new(tole_core::web::WebSearchTool::from_env().unwrap())) .map_err(|e| anyhow::anyhow!("registering web_search: {e}"))?; } - reg.register(Box::new(WriteFileTool::new(file_root.clone()))) - .map_err(|e| anyhow::anyhow!("registering write_file: {e}"))?; - reg.register(Box::new(EditFileTool::new(file_root.clone()))) - .map_err(|e| anyhow::anyhow!("registering edit_file: {e}"))?; + reg.register(tole_cli::sibling::track( + Box::new(WriteFileTool::new(file_root.clone())), + tracking.as_ref(), + &file_root, + )) + .map_err(|e| anyhow::anyhow!("registering write_file: {e}"))?; + reg.register(tole_cli::sibling::track( + Box::new(EditFileTool::new(file_root.clone())), + tracking.as_ref(), + &file_root, + )) + .map_err(|e| anyhow::anyhow!("registering edit_file: {e}"))?; #[cfg(feature = "shell-tools")] { register_gh(&mut reg, &cwd)?; @@ -1930,6 +1962,11 @@ fn run_command( // token at the deadline; the loop's checkpoints unwind it into a // durable Cancelled — resumable, never dead. let cancel = tole_core::cancel::CancelToken::default(); + // Child agents (#380): fold the shared touch log and queue a notice at + // the next safe point when a sibling changes a file this child read or + // wrote. A no-op outside a child. + let _sibling_watcher = tole_cli::sibling::tracking_from_env(&tole_cli::config_apply::real_env) + .map(|t| tole_cli::sibling::start_watcher(&t, &cancel, tole_cli::sibling::POLL_INTERVAL)); let timer = timeout_secs.map(|secs| { let token = cancel.clone(); std::thread::spawn(move || { diff --git a/crates/tole-cli/src/sibling.rs b/crates/tole-cli/src/sibling.rs new file mode 100644 index 0000000..65b54c1 --- /dev/null +++ b/crates/tole-cli/src/sibling.rs @@ -0,0 +1,185 @@ +//! Host wiring for sibling change notices (#380). +//! +//! A child agent (spawned by `agent_start`) finds its identity and the +//! shared touch log in its environment. When present, its file tools are +//! wrapped so every successful access is recorded, and a watcher thread +//! folds the shared log and queues a notice into the run's soft-interrupt +//! queue when a sibling changes a file this child read or wrote. Outside a +//! child (no env) everything here is a no-op. + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::Duration; +use tole_core::cancel::CancelToken; +use tole_core::sibling_watch::{ + SiblingWatcher, TouchRecorder, Tracked, ENV_AGENT_ID, ENV_TOUCH_LOG, +}; +use tole_core::tool::Tool; + +/// How often the watcher folds the shared log. Notices are only ever +/// delivered at the turn loop's next safe point, so this bounds latency, +/// not correctness. +pub const POLL_INTERVAL: Duration = Duration::from_millis(300); + +/// A child's identity and the shared log it reports to. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Tracking { + pub agent: String, + pub log: PathBuf, +} + +/// Read the child identity from the environment. `None` when this is not +/// a tracked child, or when the id is not a well-formed agent id (never +/// trusted blindly: it ends up in the shared log). +pub fn tracking_from_env(env: &dyn Fn(&str) -> Option) -> Option { + let agent = env(ENV_AGENT_ID)?.trim().to_string(); + let log = env(ENV_TOUCH_LOG)?.trim().to_string(); + if !tole_core::sibling_watch::valid_agent_id(&agent) || log.is_empty() { + return None; + } + Some(Tracking { + agent, + log: PathBuf::from(log), + }) +} + +/// Wrap a file tool so its successful accesses are recorded; identity when +/// not tracking. +pub fn track(tool: Box, tracking: Option<&Tracking>, root: &Path) -> Box { + match tracking { + Some(t) => Box::new(Tracked::new( + tool, + TouchRecorder::new(t.agent.clone(), t.log.clone(), root), + )), + None => tool, + } +} + +/// Background watcher; stops when dropped. +pub struct Watcher { + stop: Arc, +} + +impl Drop for Watcher { + fn drop(&mut self) { + self.stop.store(true, Ordering::SeqCst); + } +} + +/// Start folding the shared log, queueing a notice into `cancel`'s +/// soft-interrupt queue for every sibling change. A full queue drops the +/// notice (it is informational); a read error is reported once. +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()); + std::thread::spawn(move || { + let mut reported = false; + while !flag.load(Ordering::SeqCst) { + match watcher.poll() { + Ok(notices) => { + for n in notices { + let _ = token.interrupts().push_notice(&n.text()); + } + } + Err(e) if !reported => { + reported = true; + eprintln!("tole: sibling tracking: cannot read the touch log: {e}"); + } + Err(_) => {} + } + std::thread::sleep(every); + } + }); + Watcher { stop } +} + +#[cfg(test)] +mod tests { + use super::*; + use tole_core::sibling_watch::Op; + + fn env_of<'a>(pairs: &'a [(&'a str, &'a str)]) -> impl Fn(&str) -> Option + 'a { + move |k| { + pairs + .iter() + .find(|(n, _)| *n == k) + .map(|(_, v)| v.to_string()) + } + } + + #[test] + fn tracking_needs_a_well_formed_id_and_a_log() { + let ok = env_of(&[(ENV_AGENT_ID, "a-1f-2-0"), (ENV_TOUCH_LOG, "/x/touch.log")]); + assert_eq!( + tracking_from_env(&ok), + Some(Tracking { + agent: "a-1f-2-0".into(), + log: "/x/touch.log".into() + }) + ); + assert_eq!(tracking_from_env(&env_of(&[])), None); + assert_eq!( + tracking_from_env(&env_of(&[(ENV_AGENT_ID, "a-1")])), + None, + "no log" + ); + for bad in ["", "A-1", "a b", "a\n[x]", "../x"] { + let pairs = [(ENV_AGENT_ID, bad), (ENV_TOUCH_LOG, "/x")]; + let e = env_of(&pairs); + assert_eq!(tracking_from_env(&e), None, "{bad:?}"); + } + } + + #[test] + fn the_watcher_turns_a_sibling_write_into_a_queued_notice() { + let dir = std::env::temp_dir().join(format!("tole-sibwire-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + let log = dir.join("tole-agents").join("touch.log"); + let mine = Tracking { + agent: "a-1".into(), + log: log.clone(), + }; + let cancel = CancelToken::default(); + let _w = start_watcher(&mine, &cancel, Duration::from_millis(20)); + + let me = TouchRecorder::new("a-1", &log, "/ws"); + let sib = TouchRecorder::new("a-2", &log, "/ws"); + me.record(Op::Read, "src/lib.rs").unwrap(); + sib.record(Op::Write, "src/lib.rs").unwrap(); + + let deadline = std::time::Instant::now() + Duration::from_secs(5); + let mut got = Vec::new(); + while got.is_empty() && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + got = cancel.interrupts().drain(); + } + assert_eq!(got.len(), 1, "{got:?}"); + assert_eq!(got[0].source, tole_core::interrupt::Source::Notice); + assert!(got[0].text.contains("child agent a-2"), "{}", got[0].text); + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn track_is_the_identity_outside_a_child() { + struct T; + impl Tool for T { + fn name(&self) -> &str { + "read_file" + } + fn risk(&self) -> tole_core::tool::Risk { + tole_core::tool::Risk::ReadOnly + } + fn execute(&self, i: serde_json::Value) -> Result { + Ok(i) + } + } + let t = track(Box::new(T), None, Path::new("/ws")); + assert_eq!(t.name(), "read_file"); + assert_eq!( + t.execute(serde_json::json!({"path": "x"})).unwrap()["path"], + "x" + ); + } +} diff --git a/crates/tole-cli/tests/sibling_notice_e2e.rs b/crates/tole-cli/tests/sibling_notice_e2e.rs new file mode 100644 index 0000000..5b06a2d --- /dev/null +++ b/crates/tole-cli/tests/sibling_notice_e2e.rs @@ -0,0 +1,240 @@ +//! #380 sibling change notices — end-to-end with the REAL tole binary and +//! TWO real child processes sharing one touch log. +//! +//! Child A reads `shared.txt`, then takes a slow model step (the mock holds +//! the answer); meanwhile child B overwrites `shared.txt`. A's watcher +//! folds the shared log, queues a notice, and the turn loop delivers it at +//! A's next safe point — so the model's third request carries a +//! "sibling notice" entry. The mock answers `A_NOTICED` only if it does. + +#![cfg(feature = "shell-tools")] + +use serde_json::{json, Value}; +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::process::{Command, Stdio}; +use std::time::Duration; + +fn completion_to_sse(reply: &Value) -> String { + let mut out = String::new(); + let msg = &reply["choices"][0]["message"]; + let model = reply["model"].as_str().unwrap_or("mock"); + let mut chunk_of = |delta: Value| { + let c = json!({ + "id": "chatcmpl-sse", "object": "chat.completion.chunk", "created": 1, + "model": model, + "choices": [{"index": 0, "delta": delta, "finish_reason": null}], + "usage": null + }); + out.push_str(&format!("data: {c}\n\n")); + }; + if let Some(c) = msg["content"].as_str() { + if !c.is_empty() { + let (a, b) = c.split_at(c.len().div_ceil(2)); + chunk_of(json!({"content": a})); + chunk_of(json!({"content": b})); + } + } + if let Some(tcs) = msg["tool_calls"].as_array() { + for (i, tc) in tcs.iter().enumerate() { + let name = tc["function"]["name"].as_str().unwrap_or(""); + let args = tc["function"]["arguments"].as_str().unwrap_or("{}"); + let (a1, a2) = args.split_at(args.len().div_ceil(2)); + chunk_of(json!({"tool_calls": [{"index": i, "id": tc["id"], + "function": {"name": name, "arguments": a1}}]})); + chunk_of(json!({"tool_calls": [{"index": i, + "function": {"arguments": a2}}]})); + } + } + if !msg["tool_calls"].is_null() { + out.push_str(&format!( + "data: {}\n\n", + json!({"choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}], "usage": null}) + )); + } + out.push_str(&format!( + "data: {}\n\n", + json!({"choices": [], "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}}) + )); + out.push_str("data: [DONE]\n\n"); + out +} + +fn completion(msg: Value, finish: &str) -> Value { + json!({ + "id": "chatcmpl-mock", "object": "chat.completion", "created": 1, "model": "mock", + "choices": [{"index": 0, "message": msg, "finish_reason": finish}], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} + }) +} + +fn tool_call(i: u64, name: &str, args: Value) -> Value { + completion( + json!({"role": "assistant", "content": null, "tool_calls": [{ + "id": format!("call_{i}"), "type": "function", + "function": {"name": name, "arguments": args.to_string()} + }]}), + "tool_calls", + ) +} + +fn final_text(t: &str) -> Value { + completion(json!({"role": "assistant", "content": t}), "stop") +} + +/// Content-driven: CHILD_A reads twice (the second model step is slow) and +/// reports whether a sibling notice reached it; CHILD_B overwrites the file. +fn decide(body: &Value) -> (Value, Duration) { + let msgs = body["messages"].as_array().cloned().unwrap_or_default(); + let first_user = msgs + .iter() + .find(|m| m["role"] == "user") + .and_then(|m| m["content"].as_str()) + .unwrap_or("") + .to_string(); + let tool_results = msgs.iter().filter(|m| m["role"] == "tool").count(); + let notice_seen = msgs.iter().any(|m| { + m["content"] + .as_str() + .is_some_and(|c| c.contains("sibling notice") && c.contains("child agent a-b")) + }); + if first_user.contains("CHILD_A") { + return match tool_results { + 0 => ( + tool_call(1, "read_file", json!({"path": "shared.txt"})), + Duration::ZERO, + ), + // The slow step: B changes the file while this answer is held. + 1 => ( + tool_call(2, "read_file", json!({"path": "shared.txt", "limit": 5})), + Duration::from_secs(4), + ), + _ if notice_seen => (final_text("A_NOTICED"), Duration::ZERO), + _ => (final_text("A_BLIND"), Duration::ZERO), + }; + } + if tool_results == 0 { + return ( + tool_call( + 1, + "write_file", + json!({"path": "shared.txt", "content": "changed by b"}), + ), + Duration::ZERO, + ); + } + (final_text("B_DONE"), Duration::ZERO) +} + +fn spawn_mock() -> String { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + for stream in listener.incoming().flatten() { + std::thread::spawn(move || { + let mut s = stream; + let mut buf = Vec::new(); + let mut chunk = [0u8; 16384]; + let mut body = Value::Null; + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while std::time::Instant::now() < deadline { + let n = s.read(&mut chunk).unwrap_or(0); + if n > 0 { + buf.extend_from_slice(&chunk[..n]); + } + if let Some(h) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + if let Ok(v) = serde_json::from_slice::(&buf[h + 4..]) { + body = v; + break; + } + } + if n == 0 { + break; + } + } + let streamed = body.get("stream").and_then(Value::as_bool) == Some(true); + let (reply, delay) = decide(&body); + std::thread::sleep(delay); + let (ctype, payload) = if streamed { + ("text/event-stream", completion_to_sse(&reply)) + } else { + ("application/json", reply.to_string()) + }; + let resp = format!( + "HTTP/1.1 200 OK\r\nContent-Type: {ctype}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{payload}", + payload.len() + ); + let _ = s.write_all(resp.as_bytes()); + }); + } + }); + format!("http://{addr}/v1") +} + +fn tole( + base: &str, + ws: &std::path::Path, + log: &std::path::Path, + id: &str, + prompt: &str, +) -> Command { + let mut cmd = Command::new(env!("CARGO_BIN_EXE_tole")); + cmd.args(["run", "--allow", "write_file", "--"]) + .arg(prompt) + .current_dir(ws) + .env("TOLE_BASE_URL", base) + .env("TOLE_API_KEY", "test") + .env("TOLE_MODEL", "mock") + .env("TOLE_AGENT_ID", id) + .env("TOLE_AGENT_TOUCH_LOG", log) + .env_remove("TOLE_MEMORY") + .env_remove("TOLE_TRUST") + .env_remove("TOLE_SYSTEM_PROMPT") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + cmd +} + +#[test] +fn a_sibling_edit_reaches_the_other_child_at_its_next_safe_point() { + let base = spawn_mock(); + let root = std::env::temp_dir().join(format!("tole-sibling-e2e-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + let ws = root.join("ws"); + std::fs::create_dir_all(&ws).unwrap(); + std::fs::write(ws.join("shared.txt"), "original contents\n").unwrap(); + let log = root.join("tole-agents").join("touch.log"); + + // Child A starts and reads the file, then waits on its slow model step. + let a = tole(&base, &ws, &log, "a-a", "CHILD_A: read shared.txt twice") + .spawn() + .unwrap(); + std::thread::sleep(Duration::from_millis(1500)); + + // Child B overwrites it while A is waiting. + let b = tole(&base, &ws, &log, "a-b", "CHILD_B: overwrite shared.txt") + .output() + .unwrap(); + let b_out = String::from_utf8_lossy(&b.stdout); + assert!( + b.status.success(), + "B failed: {b_out}\n{}", + String::from_utf8_lossy(&b.stderr) + ); + assert!(b_out.contains("B_DONE"), "{b_out}"); + assert_eq!( + std::fs::read_to_string(ws.join("shared.txt")).unwrap(), + "changed by b" + ); + + let a = a.wait_with_output().unwrap(); + let a_out = String::from_utf8_lossy(&a.stdout); + let a_err = String::from_utf8_lossy(&a.stderr); + assert!(a.status.success(), "A failed: {a_out}\n{a_err}"); + assert!( + a_out.contains("A_NOTICED"), + "A never got the notice: {a_out}\n{a_err}" + ); + let _ = std::fs::remove_dir_all(&root); +} diff --git a/crates/tole-core/src/agents.rs b/crates/tole-core/src/agents.rs index 08cca6e..f833aad 100644 --- a/crates/tole-core/src/agents.rs +++ b/crates/tole-core/src/agents.rs @@ -244,7 +244,8 @@ fn new_agent_id() -> String { format!("a-{ms:x}-{:x}-{n:x}", std::process::id()) } -pub(crate) fn valid_agent_id(id: &str) -> bool { +/// The agent-id charset: see [`crate::sibling_watch::valid_agent_id`]. +pub fn valid_agent_id(id: &str) -> bool { crate::sibling_watch::valid_agent_id(id) } @@ -662,6 +663,13 @@ impl Tool for AgentStartTool { // needs the provider key. Isolation = depth cap + jail + the // approved child allowlist (see the module doc). cmd.env("TOLE_AGENT_DEPTH", "1") + // Sibling change notices (#380): the child's identity and the + // shared touch log every sibling of this parent appends to. + .env(crate::sibling_watch::ENV_AGENT_ID, &id) + .env( + crate::sibling_watch::ENV_TOUCH_LOG, + agents_root(&self.root).join("touch.log"), + ) .env("TOLE_MEMORY_NAMESPACE", &mailbox_ns) .env("TOLE_MEMORY", "uteke") .arg("run") @@ -1195,6 +1203,50 @@ mod tests { let _ = std::fs::remove_dir_all(&d); } + /// #380: the child learns its own id and the shared touch log of this + /// parent through the environment. + #[test] + fn child_gets_its_id_and_the_shared_touch_log_in_env() { + let d = tmp("sibenv"); + #[cfg(unix)] + { + let p = d.join("dump-env"); + let dump = d.join("env.txt"); + let script = format!( + "#!/bin/sh\nprintf '%s\\n%s\\n' \"$TOLE_AGENT_ID\" \"$TOLE_AGENT_TOUCH_LOG\" > \"{}\"\nexit 0\n", + dump.display() + ); + std::fs::write(&p, script).unwrap(); + use std::os::unix::fs::PermissionsExt; + let _ = std::fs::set_permissions(&p, std::fs::Permissions::from_mode(0o755)); + let out = AgentStartTool::new(&p, d.clone()) + .execute(json!({"prompt": "x"})) + .unwrap(); + let mut dumped = None; + for _ in 0..50 { + std::thread::sleep(std::time::Duration::from_millis(100)); + if let Ok(t) = std::fs::read_to_string(&dump) { + if t.lines().count() >= 2 { + dumped = Some(t); + break; + } + } + } + let dumped = dumped.expect("child never dumped its env"); + let mut lines = dumped.lines(); + let id = out["agent"].as_str().unwrap(); + assert_eq!(lines.next(), Some(id)); + assert!(valid_agent_id(id)); + let log = lines.next().unwrap(); + assert_eq!( + std::path::Path::new(log), + agents_root(&d).join("touch.log"), + "all siblings of one parent share one log" + ); + } + let _ = std::fs::remove_dir_all(&d); + } + #[test] fn depth_one_refuses_to_spawn() { let d = tmp("depth"); diff --git a/crates/tole-core/src/sibling_watch.rs b/crates/tole-core/src/sibling_watch.rs index e39a7c9..1f1c042 100644 --- a/crates/tole-core/src/sibling_watch.rs +++ b/crates/tole-core/src/sibling_watch.rs @@ -23,6 +23,11 @@ use std::collections::HashSet; use std::io::{Read, Seek, SeekFrom, Write}; use std::path::{Component, Path, PathBuf}; +/// Env var carrying a child's agent id (set by `agent_start`). +pub const ENV_AGENT_ID: &str = "TOLE_AGENT_ID"; +/// Env var carrying the shared touch-log path (set by `agent_start`). +pub const ENV_TOUCH_LOG: &str = "TOLE_AGENT_TOUCH_LOG"; + /// The agent-id charset (`a---` as generated by /// `agent_start`): lowercase alphanumerics and `-`, at most 64 bytes. Ids /// read back from env or the shared log are checked with this before they