diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index c7d2038..0094413 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -31,10 +31,10 @@ crates above it in this table: |---|---|---| | `bullpen-llm` | Provider-neutral conversation types, `Provider` trait, wire-format adapters (Anthropic messages, OpenAI chat-completions, Codex Responses/SSE), shared retry policy | Tools, transcripts, UI | | `bullpen-auth` | Credential store (`~/.bullpen/auth.json` or `$BULLPEN_HOME/auth.json`, 0600, atomic), PKCE, OpenRouter OAuth, Codex device-code flow + refresh, read-only borrow of `~/.codex/auth.json` | Tools, the loop, UI | -| `bullpen-tools` | `Tool` trait, `Registry`, built-ins (bash, read/write/edit, grep, glob), parallel-safety flags | Providers, the loop | +| `bullpen-tools` | `Tool` trait, `Registry`, built-ins (bash, hashline read/write/edit, grep, glob, ask), parallel-safety flags | Providers, the loop | | `bullpen-store` | SQLite persistence: sessions, transcripts, usage; schema migrations via `user_version` | Providers, tools, the loop | | `bullpen-agent` | The loop: transcript, provider calls, tool continuation, events, max-turns fuse, the `Journal` durability protocol (trait only) | Config files, sessions, vendors, UI, storage | -| `bullpen-harness` | Durable execution: `StoreJournal` (implements `Journal` over the store), crash recovery orchestration; future home of the pen | Vendors, UI | +| `bullpen-harness` | Durable execution: `StoreJournal` (implements `Journal` over the store), crash recovery orchestration; the pen (durable subagents), `job`, `todo`, worktree placement | Vendors, UI | | `bullpen` (cli) | Composition root: wiring, system prompt, headless `run`, `sessions` | — (the only crate that knows everything) | The rule that makes the table above enforceable, kept absolute: **the core @@ -232,7 +232,7 @@ Milestones, in order. Each lands as its own crate or a bounded extension: request order. `Tool::parallel_safe` judges the concrete input — the runtime owns the decision, never the model. Read/grep/glob are parallel-safe; bash and writes are serial; the `agent` tool is - parallel-safe only in `inspect` mode. Durability is unchanged by + parallel-safe for `inspect`, worktree-isolated, and background children. Durability is unchanged by scheduling: the whole batch's intents are journaled before any execution, and a crash mid-batch synthesizes the whole batch on recovery. Still open for M2.x: live child event streaming to the parent's UI, token budgets diff --git a/Cargo.lock b/Cargo.lock index ccc704f..f9b3d1c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -138,6 +138,7 @@ name = "bullpen" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "bullpen-agent", "bullpen-auth", "bullpen-harness", @@ -193,6 +194,7 @@ dependencies = [ name = "bullpen-harness" version = "0.1.0" dependencies = [ + "anyhow", "async-trait", "bullpen-agent", "bullpen-llm", @@ -233,6 +235,7 @@ version = "0.1.0" dependencies = [ "bullpen-llm", "fs2", + "libc", "rusqlite", "serde", "serde_json", @@ -248,11 +251,14 @@ dependencies = [ "async-trait", "bullpen-llm", "bullpen-sandbox", + "futures", "globset", "ignore", "regex", + "reqwest", "serde", "serde_json", + "sha2", "tempfile", "thiserror", "tokio", diff --git a/README.md b/README.md index 661e766..9de7d95 100644 --- a/README.md +++ b/README.md @@ -106,10 +106,13 @@ so cleaning up is yours to decide. ## The pen The model can delegate bounded work to child agents through the `agent` -tool: `inspect` for read-only reconnaissance, `work` for the full toolset. -Children are ordinary sessions — durable, budgeted, listed by -`bullpen sessions`, resumable — with deterministic identities, so a replayed -delegation reattaches to its child instead of running it twice. +tool: `inspect` for read-only reconnaissance, `work` for the full toolset — +optionally in the child's own git worktree (`worktree: true`), optionally +detached (`background: true`) with the `job` tool to list, wait on, and +cancel what's in flight. Children are ordinary sessions — durable, +budgeted, listed by `bullpen sessions`, resumable — with deterministic +identities, so a replayed delegation reattaches to its child instead of +running it twice. ## Providers @@ -125,7 +128,9 @@ Adapters are organized by wire format rather than vendor, which is why compatible hosts are configuration instead of code. Built-in tools: `bash`, `read_file`, `write_file`, `edit_file`, `grep`, -`glob` — plus `agent` when the pen is enabled. +`glob`, `todo` (a durable session plan), `ask` (follow-up questions on +interactive runs) — plus `agent` and `job` when the pen is enabled. +[docs/TOOLS.md](docs/TOOLS.md) maps the rest of the planned surface. ## Where state lives @@ -151,7 +156,7 @@ directly in it, with no `.bullpen` segment appended. | | | |---|---| -| ✅ Shipped | Durable execution + crash recovery · the pen (durable subagents) · write-confinement sandbox with Seatbelt on macOS · agent view (dispatch, peek, live state) · 5 providers | +| ✅ Shipped | Durable execution + crash recovery · the pen (durable subagents, worktree isolation, background dispatch + `job`) · hashline edits with anchor recovery · durable session plans (`todo`) · follow-up questions (`ask`) · write-confinement sandbox with Seatbelt on macOS · agent view (dispatch, peek, live state) · 5 providers | | 🚧 Next | Interactive attach to a live session · needs-input state · notifications · compaction | | 📋 Planned | Landlock confinement on Linux · a durable workflow engine (steps in SQLite, resumable from any step) | diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index 8b02cec..0434344 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -17,6 +17,7 @@ bullpen-llm.workspace = true bullpen-store.workspace = true bullpen-tools.workspace = true anyhow.workspace = true +async-trait.workspace = true clap.workspace = true reqwest.workspace = true serde_json.workspace = true diff --git a/crates/cli/src/bg.rs b/crates/cli/src/bg.rs index cd3d23e..424fb3a 100644 --- a/crates/cli/src/bg.rs +++ b/crates/cli/src/bg.rs @@ -17,17 +17,7 @@ fn logs_dir() -> PathBuf { bullpen_store::home_dir().join("logs") } -/// Whether `pid` is a live process. Uses `kill(pid, 0)`: success or an -/// `EPERM` both mean the process exists. -pub fn pid_alive(pid: i64) -> bool { - if pid <= 0 { - return false; - } - // SAFETY: kill with signal 0 performs only the existence/permission - // check and never delivers a signal. - let rc = unsafe { libc::kill(pid as libc::pid_t, 0) }; - rc == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) -} +pub use bullpen_store::status::pid_alive; /// Spawn a detached `bullpen run --resume ` that outlives /// this process and the controlling terminal. Returns the child's pid. diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 82cc150..ebf15ac 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -7,7 +7,8 @@ mod agents; mod bg; mod json; -mod worktree; + +use bullpen_harness::worktree; use std::sync::Arc; @@ -587,11 +588,24 @@ async fn run( pen_config = pen_config.with_sandbox(sb.clone()); } let mut registry = Registry::standard(); - registry.register(std::sync::Arc::new(bullpen_harness::PenTool::new( - provider.clone(), + let pen = bullpen_harness::PenTool::new(provider.clone(), &session.id, pen_config); + // The job tool shares the pen's cancel registry, so background children + // dispatched by this process can be cancelled from the same session. + registry.register(std::sync::Arc::new(pen.job_tool())); + registry.register(std::sync::Arc::new(pen)); + // The plan: a durable todo list scoped to this session, in the store. + registry.register(std::sync::Arc::new(bullpen_harness::TodoTool::new( + Store::default_path(), &session.id, - pen_config, ))); + // Follow-up questions reach the terminal only when someone is on it; + // otherwise the detached variant answers with the reason instead. + let interactive = !json && std::io::IsTerminal::is_terminal(&std::io::stdin()); + registry.register(std::sync::Arc::new(if interactive { + bullpen_tools::Ask::interactive(Arc::new(TtyAsker)) + } else { + bullpen_tools::Ask::detached() + })); let mut agent = Agent::new(provider, registry, tool_ctx, config) .with_transcript(transcript, usage) .with_events(tx) @@ -744,6 +758,37 @@ fn sessions(json: bool) -> anyhow::Result<()> { Ok(()) } +/// The CLI's [`bullpen_tools::Asker`]: the question goes to stderr — never +/// into the answer stream on stdout — and the reply is one line from the +/// terminal. Reading blocks a spawned blocking thread, not the runtime. +struct TtyAsker; + +#[async_trait::async_trait] +impl bullpen_tools::Asker for TtyAsker { + async fn ask(&self, prompt: &str) -> Result { + let prompt = prompt.to_string(); + tokio::task::spawn_blocking(move || { + use std::io::{BufRead, Write}; + let mut err = std::io::stderr(); + let _ = writeln!(err, "\n[bullpen asks]\n{prompt}"); + let _ = write!(err, "> "); + let _ = err.flush(); + let mut line = String::new(); + match std::io::stdin().lock().read_line(&mut line) { + Ok(0) => Err(bullpen_tools::ToolError::Failed( + "input closed before an answer".into(), + )), + Ok(_) => Ok(line), + Err(e) => Err(bullpen_tools::ToolError::Failed(format!( + "could not read the answer: {e}" + ))), + } + }) + .await + .map_err(|e| bullpen_tools::ToolError::Failed(format!("ask task failed: {e}")))? + } +} + fn system_prompt(cwd: &std::path::Path) -> String { format!( "You are bullpen, a coding agent operating in a repository.\n\ diff --git a/crates/harness/Cargo.toml b/crates/harness/Cargo.toml index 5b5bf09..979ad6a 100644 --- a/crates/harness/Cargo.toml +++ b/crates/harness/Cargo.toml @@ -10,6 +10,7 @@ bullpen-llm.workspace = true bullpen-store.workspace = true bullpen-tools.workspace = true bullpen-sandbox.workspace = true +anyhow.workspace = true async-trait.workspace = true serde_json.workspace = true tokio.workspace = true diff --git a/crates/harness/src/job.rs b/crates/harness/src/job.rs new file mode 100644 index 0000000..8b8e9a8 --- /dev/null +++ b/crates/harness/src/job.rs @@ -0,0 +1,384 @@ +//! The job tool: the coordination plane, exposed to the model. +//! +//! `bullpen agents` reads sessions from the store and never talks to the +//! processes running them; `job` gives the model the same read-and-signal +//! view over its own pen children. `list` derives each child's state from +//! its stored status plus process liveness — the store is the source of +//! truth, so a `job` call after a crash sees reality, not a stale +//! in-process handle. `wait` polls that truth to a terminal state and +//! returns the child's recorded answer. `cancel` signals a background +//! child dispatched by this process; the child finishes as failed and +//! stays resumable, because cancellation is an outcome, not an erasure. + +use std::path::PathBuf; +use std::time::Duration; + +use bullpen_llm::{Role, ToolSpec}; +use bullpen_store::status::{AgentStatus, for_session, pid_alive}; +use bullpen_store::{Session, Store, StoreError}; +use bullpen_tools::{Tool, ToolCtx, ToolError}; +use serde_json::{Value, json}; + +use crate::pen::Cancels; + +const DEFAULT_WAIT_SECS: u64 = 900; +const MAX_WAIT_SECS: u64 = 3600; +const POLL_INTERVAL: Duration = Duration::from_millis(250); + +/// Built by [`crate::PenTool::job_tool`], sharing the pen's cancellation +/// registry. +pub struct JobTool { + store_path: PathBuf, + session_id: String, + cancels: Cancels, +} + +impl JobTool { + pub(crate) fn new( + store_path: PathBuf, + session_id: impl Into, + cancels: Cancels, + ) -> Self { + Self { + store_path, + session_id: session_id.into(), + cancels, + } + } + + fn children(&self, store: &Store) -> Result, ToolError> { + store + .list_children(&self.session_id) + .map_err(|e| ToolError::Failed(format!("store: {e}"))) + } + + /// Resolve a child by id prefix, mirroring session/todo resolution. + fn resolve(&self, store: &Store, prefix: &str) -> Result { + let matches: Vec = self + .children(store)? + .into_iter() + .filter(|s| s.id.starts_with(prefix)) + .collect(); + match matches.len() { + 0 => Err(ToolError::InvalidInput(format!( + "no child of this session matches `{prefix}`" + ))), + 1 => Ok(matches.into_iter().next().unwrap()), + _ => Err(ToolError::InvalidInput(format!( + "child id `{prefix}` is ambiguous — use a longer prefix from `list`" + ))), + } + } +} + +fn derived(session: &Session) -> AgentStatus { + for_session(session, session.pid.is_some_and(pid_alive)) +} + +fn render(children: &[Session]) -> String { + if children.is_empty() { + return "No children dispatched in this session.".into(); + } + let mut out = format!("Children ({}):\n", children.len()); + for child in children { + let title = if child.title.is_empty() { + "(no task recorded)" + } else { + &child.title + }; + out.push_str(&format!( + " {} {:9} {title}{}\n", + &child.id[..8], + derived(child).label(), + if child.worktree_path.is_some() { + " · worktree" + } else { + "" + } + )); + } + out +} + +/// The child's final report: the last assistant message on its path. +fn answer_of(store: &Store, child_id: &str) -> Result { + Ok(store + .path_messages(child_id)? + .iter() + .rev() + .find(|m| m.role == Role::Assistant) + .map(|m| m.text()) + .unwrap_or_default()) +} + +#[async_trait::async_trait] +impl Tool for JobTool { + fn name(&self) -> &'static str { + "job" + } + + fn spec(&self) -> ToolSpec { + ToolSpec { + name: "job".into(), + description: format!( + "Coordinate this session's child agents. `list` shows every \ + child and its state (Working, Completed, Failed, Idle). \ + `wait` blocks until one child finishes and returns its final \ + report (default timeout {DEFAULT_WAIT_SECS}s, maximum \ + {MAX_WAIT_SECS}s). `cancel` stops a background child \ + dispatched in this process; its session is kept and \ + resumable. Children are addressed by the id prefix `list` \ + shows. Pair with `agent` + `background: true`: dispatch \ + several, then wait on each." + ), + input_schema: json!({ + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["list", "wait", "cancel"], + "description": "What to do" + }, + "id": { + "type": "string", + "description": "For `wait`/`cancel`: the child's id (any unique prefix)" + }, + "timeout_seconds": { + "type": "integer", + "description": "For `wait`: how long to block before giving up" + } + }, + "required": ["action"] + }), + } + } + + fn parallel_safe(&self, input: &Value) -> bool { + // `list` and `wait` only read the store, so several waits can block + // side by side — that is how a fan-out joins. `cancel` signals a + // running child and stays serial. + matches!( + input.get("action").and_then(Value::as_str), + Some("list") | Some("wait") + ) + } + + async fn run(&self, _ctx: &ToolCtx, _call_id: &str, input: Value) -> Result { + let action = input.get("action").and_then(Value::as_str).ok_or_else(|| { + ToolError::InvalidInput("missing required string field `action`".into()) + })?; + let store = + Store::open(&self.store_path).map_err(|e| ToolError::Failed(format!("store: {e}")))?; + let prefix = || { + input + .get("id") + .and_then(Value::as_str) + .ok_or_else(|| ToolError::InvalidInput(format!("`{action}` needs an `id` prefix"))) + }; + + match action { + "list" => Ok(render(&self.children(&store)?)), + "wait" => { + let child = self.resolve(&store, prefix()?)?; + let timeout = input + .get("timeout_seconds") + .and_then(Value::as_u64) + .unwrap_or(DEFAULT_WAIT_SECS) + .min(MAX_WAIT_SECS); + let short = &child.id[..8]; + let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout); + loop { + let current = store + .get_session(&child.id) + .map_err(|e| ToolError::Failed(format!("store: {e}")))?; + match derived(¤t) { + AgentStatus::Completed => { + let answer = answer_of(&store, &child.id) + .map_err(|e| ToolError::Failed(format!("store: {e}")))?; + return Ok(format!("{answer}\n\n[child {short} · completed]")); + } + AgentStatus::Failed => { + return Err(ToolError::Failed(format!( + "child {short} failed or was interrupted; its \ + session is saved — `agent` with the same task \ + continues it" + ))); + } + // Idle covers the dispatch race: created but not yet + // started. Keep polling; the deadline bounds it. + AgentStatus::Working | AgentStatus::Idle => {} + } + if tokio::time::Instant::now() >= deadline { + return Err(ToolError::Timeout(timeout)); + } + tokio::time::sleep(POLL_INTERVAL).await; + } + } + "cancel" => { + let child = self.resolve(&store, prefix()?)?; + let short = &child.id[..8]; + match self.cancels.lock().unwrap().remove(&child.id) { + Some(cancel) => { + // A dropped receiver means the child just finished on + // its own — that is a success for `cancel` too. + let _ = cancel.send(()); + Ok(format!( + "cancel signalled for child {short}; its session is \ + saved and resumable" + )) + } + None => match derived(&child) { + AgentStatus::Working => Err(ToolError::Failed(format!( + "child {short} is running in another process (pid \ + {}); this session did not dispatch it, so cancel \ + it there", + child.pid.unwrap_or(0) + ))), + _ => Err(ToolError::InvalidInput(format!( + "child {short} is not running ({})", + derived(&child).label() + ))), + }, + } + } + other => Err(ToolError::InvalidInput(format!( + "unknown action `{other}` (expected list, wait, or cancel)" + ))), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::pen::PenTool; + use crate::testutil::{FakeProvider, text_response}; + use crate::{PenConfig, prepare_session}; + + fn setup(dir: &tempfile::TempDir) -> (PenTool, JobTool, String) { + let store_path = dir.path().join("t.db"); + let parent = Store::open(&store_path) + .unwrap() + .create_session("/tmp", "fake", "m") + .unwrap() + .id; + let pen = PenTool::new( + FakeProvider::new(vec![text_response("child answer")]), + &parent, + PenConfig::new( + store_path, + std::env::temp_dir(), + "fake", + "test-model", + "base system", + ), + ); + let job = pen.job_tool(); + (pen, job, parent) + } + + fn ctx() -> ToolCtx { + ToolCtx::new(std::env::temp_dir()) + } + + #[tokio::test] + async fn empty_list_and_bad_inputs() { + let dir = tempfile::tempdir().unwrap(); + let (_pen, job, _) = setup(&dir); + + let out = job + .run(&ctx(), "j", json!({"action": "list"})) + .await + .unwrap(); + assert_eq!(out, "No children dispatched in this session."); + + for input in [ + json!({}), + json!({"action": "wait"}), + json!({"action": "cancel", "id": "zzz"}), + json!({"action": "nope"}), + ] { + let err = job.run(&ctx(), "j", input).await.unwrap_err(); + assert!(matches!(err, ToolError::InvalidInput(_)), "{err}"); + } + } + + #[tokio::test] + async fn list_shows_the_completed_child_and_wait_returns_its_answer() { + let dir = tempfile::tempdir().unwrap(); + let (pen, job, parent) = setup(&dir); + pen.run(&ctx(), "call_1", json!({"prompt": "go"})) + .await + .unwrap(); + + let listed = job + .run(&ctx(), "j1", json!({"action": "list"})) + .await + .unwrap(); + let short = &crate::pen::child_session_id(&parent, "call_1")[..8]; + assert!(listed.contains(short), "{listed}"); + assert!(listed.contains("Completed"), "{listed}"); + + // Waiting on an already-finished child returns immediately. + let waited = job + .run(&ctx(), "j2", json!({"action": "wait", "id": short})) + .await + .unwrap(); + assert!(waited.contains("child answer"), "{waited}"); + + // A finished child cannot be cancelled. + let err = job + .run(&ctx(), "j3", json!({"action": "cancel", "id": short})) + .await + .unwrap_err(); + assert!(err.to_string().contains("not running"), "{err}"); + } + + #[tokio::test] + async fn a_crashed_child_reads_as_failed_not_working() { + let dir = tempfile::tempdir().unwrap(); + let (_pen, job, parent) = setup(&dir); + let child_id = crate::pen::child_session_id(&parent, "call_1"); + + // Fabricate a child whose process died mid-run: status running, a + // pid that cannot be alive. + let mut store = Store::open(&dir.path().join("t.db")).unwrap(); + store + .create_child_session(&child_id, &parent, "/tmp", "fake", "m") + .unwrap(); + store.start_operation(&child_id, &json!({})).unwrap(); + store.start_worker(&child_id, i32::MAX as i64 - 1).unwrap(); + + let listed = job + .run(&ctx(), "j1", json!({"action": "list"})) + .await + .unwrap(); + assert!(listed.contains("Failed"), "{listed}"); + + let err = job + .run( + &ctx(), + "j2", + json!({"action": "wait", "id": &child_id[..8], "timeout_seconds": 5}), + ) + .await + .unwrap_err(); + assert!( + err.to_string().contains("failed or was interrupted"), + "{err}" + ); + + // And the promised continuation path actually recovers it. + let (_, recovery) = prepare_session(&mut store, &child_id).unwrap(); + assert!(recovery.is_some()); + } + + #[test] + fn reads_are_parallel_safe_cancel_is_not() { + let dir = tempfile::tempdir().unwrap(); + let (_pen, job, _) = setup(&dir); + assert!(job.parallel_safe(&json!({"action": "list"}))); + assert!(job.parallel_safe(&json!({"action": "wait", "id": "a"}))); + assert!(!job.parallel_safe(&json!({"action": "cancel", "id": "a"}))); + } +} diff --git a/crates/harness/src/lib.rs b/crates/harness/src/lib.rs index bba9c27..846856e 100644 --- a/crates/harness/src/lib.rs +++ b/crates/harness/src/lib.rs @@ -6,13 +6,20 @@ //! idempotent. [`prepare_session`] is the front door: recover if a previous //! process died mid-run, then hand back the rebuilt transcript. //! -//! This crate is the future home of the pen (durable subagents). +//! This crate also hosts the session-scoped tools that need the store: +//! the pen (durable subagents), `job` (the coordination plane), `todo` +//! (the durable session plan), and worktree placement. +pub mod job; pub mod pen; #[cfg(test)] pub(crate) mod testutil; +pub mod todo; +pub mod worktree; +pub use job::JobTool; pub use pen::{PenConfig, PenTool}; +pub use todo::TodoTool; use bullpen_agent::{Journal, JournalError, RunOutcome, ToolIntent}; use bullpen_llm::{Message, Usage}; diff --git a/crates/harness/src/pen.rs b/crates/harness/src/pen.rs index 062b4a2..41fff81 100644 --- a/crates/harness/src/pen.rs +++ b/crates/harness/src/pen.rs @@ -15,18 +15,32 @@ //! //! Budgets are durable too: the child count is a database count, not an //! in-process counter, so a crash-restart loop cannot reset it. - -use std::path::PathBuf; -use std::sync::Arc; +//! +//! Work children can be **isolated** (`worktree: true`): each runs in its +//! own git worktree on a `bullpen/` branch, exactly like +//! `bullpen run --bg --worktree`, which is what makes isolated work +//! children safe to run in parallel. And any child can be **dispatched to +//! the background** (`background: true`): the spawn returns immediately +//! and the child runs in this process, coordinated — like everything else +//! — through the store, where the `job` tool finds it. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; use std::time::Duration; use bullpen_agent::{Agent, AgentConfig}; use bullpen_llm::{Provider, Role, ToolSpec}; -use bullpen_store::{SessionWorker, Store}; +use bullpen_store::{Session, SessionWorker, Store}; use bullpen_tools::{Glob, Grep, ReadFile, Registry, Tool, ToolCtx, ToolError}; use serde_json::{Value, json}; -use crate::{StoreJournal, prepare_session}; +use crate::{StoreJournal, prepare_session, worktree}; + +/// Cancellation handles for background children running in this process, +/// shared with the `job` tool. A child's entry exists exactly while its +/// task runs here; children run by other processes are never in it. +pub(crate) type Cancels = Arc>>>; #[derive(Clone)] pub struct PenConfig { @@ -77,6 +91,7 @@ pub struct PenTool { provider: Arc, parent_session: String, config: PenConfig, + cancels: Cancels, } impl PenTool { @@ -89,6 +104,68 @@ impl PenTool { provider, parent_session: parent_session.into(), config, + cancels: Cancels::default(), + } + } + + /// The `job` tool over this pen's children, sharing its cancellation + /// registry so background children dispatched here can be cancelled. + pub fn job_tool(&self) -> crate::job::JobTool { + crate::job::JobTool::new( + self.config.store_path.clone(), + &self.parent_session, + self.cancels.clone(), + ) + } + + /// Where a child runs. First worktree dispatch records then creates the + /// tree; after that the recorded state decides, exactly as it does for + /// a CLI resume — a replayed spawn reattaches to the same worktree. + fn place_child( + &self, + store: &Store, + child: &Session, + use_worktree: bool, + ) -> Result { + if use_worktree && child.worktree_path.is_none() { + let root = worktree::repo_root(&self.config.workspace) + .map_err(|e| ToolError::Failed(e.to_string()))?; + let path = worktree::worktree_path_for_store(&self.config.store_path, &child.id); + let branch = worktree::branch_for(&child.id); + // Recorded before it is created (see the CLI dispatch path): a + // failed creation leaves a row naming a missing directory, which + // resume refuses, never a row naming none. + store + .set_worktree(&child.id, &path.display().to_string(), &branch) + .map_err(|e| ToolError::Failed(format!("store: {e}")))?; + worktree::create(&root, &path, &branch) + .map_err(|e| ToolError::Failed(e.to_string()))?; + return Ok(path); + } + match worktree::locate( + child.worktree_path.as_deref(), + child.worktree_branch.as_deref(), + Path::new(&child.cwd), + ) { + worktree::Location::Shared => Ok(self.config.workspace.clone()), + worktree::Location::Use(path) => Ok(path), + worktree::Location::Recreate { path, branch } => { + let root = worktree::repo_root(Path::new(&child.cwd)) + .map_err(|e| ToolError::Failed(e.to_string()))?; + worktree::recreate(&root, &path, &branch) + .map_err(|e| ToolError::Failed(e.to_string()))?; + Ok(path) + } + worktree::Location::Fail { path, branch } => Err(ToolError::Failed(format!( + "child worktree {} and branch {branch} are both gone; restore \ + them or use a new task", + path.display() + ))), + worktree::Location::Occupied { path, .. } => Err(ToolError::Failed(format!( + "something that is not the child's worktree occupies {}; move \ + it aside or use a new task", + path.display() + ))), } } } @@ -132,8 +209,13 @@ impl Tool for PenTool { description: "Delegate a bounded task to a child agent with its own \ context window. Mode `inspect` (default) gives it \ read-only tools (read_file, grep, glob); mode `work` \ - adds bash and file editing. The child runs to \ - completion and returns its final report. Use for \ + adds bash and file editing. By default the child runs \ + to completion and returns its final report. \ + `worktree: true` (work mode only) runs the child in \ + its own git worktree on its own branch, so isolated \ + work children can run in parallel. `background: true` \ + dispatches the child and returns immediately — use \ + the job tool to list, wait on, or cancel it. Use for \ research across many files or self-contained subtasks." .into(), input_schema: json!({ @@ -147,6 +229,14 @@ impl Tool for PenTool { "type": "string", "enum": ["inspect", "work"], "description": "Capability set for the child (default: inspect)" + }, + "worktree": { + "type": "boolean", + "description": "Run a work child in its own git worktree (default: false)" + }, + "background": { + "type": "boolean", + "description": "Dispatch and return immediately instead of waiting (default: false)" } }, "required": ["prompt"] @@ -157,13 +247,16 @@ impl Tool for PenTool { fn parallel_safe(&self, input: &Value) -> bool { // Inspect children are read-only in the workspace and each writes // only to its own child session (WAL handles the store contention), - // so they can run alongside each other. Work children mutate the - // workspace and stay serial. - input + // so they can run alongside each other. So can isolated work + // children — each mutates only its own worktree — and background + // dispatches, which only spawn and return. A work child in the + // shared checkout stays serial. + let mode = input .get("mode") .and_then(Value::as_str) - .unwrap_or("inspect") - == "inspect" + .unwrap_or("inspect"); + let flag = |key| input.get(key).and_then(Value::as_bool).unwrap_or(false); + mode == "inspect" || flag("worktree") || flag("background") } async fn run(&self, _ctx: &ToolCtx, call_id: &str, input: Value) -> Result { @@ -174,11 +267,29 @@ impl Tool for PenTool { .get("mode") .and_then(Value::as_str) .unwrap_or("inspect"); + let flag = |key| input.get(key).and_then(Value::as_bool).unwrap_or(false); + let (use_worktree, background) = (flag("worktree"), flag("background")); + if use_worktree && mode != "work" { + return Err(ToolError::InvalidInput( + "worktree isolation is for `work` children; inspect children \ + read the shared checkout" + .into(), + )); + } let registry = registry_for_mode(mode)?; let child_id = child_session_id(&self.parent_session, call_id); let short = &child_id[..8]; - let mut store = Store::open(&self.config.store_path) + + // A child this process already has in flight: report, don't respawn. + if self.cancels.lock().unwrap().contains_key(&child_id) { + return Ok(format!( + "child {short} is already running in the background; use the \ + job tool to wait on or cancel it" + )); + } + + let store = Store::open(&self.config.store_path) .map_err(|e| ToolError::Failed(format!("store: {e}")))?; // Budget applies to *new* children only; reattaching to an existing @@ -229,93 +340,207 @@ impl Tool for PenTool { )); } - // Fresh or interrupted child: acquire the same exclusive ownership - // used by top-level CLI runs before recovery or provider activity. - // This prevents a manual `bullpen run -r ` from racing the pen. - let mut session_worker = SessionWorker::acquire(&self.config.store_path, &child_id) - .map_err(|e| ToolError::Failed(format!("child {short}: {e}")))?; - session_worker - .start(&mut store) - .map_err(|e| ToolError::Failed(format!("child {short}: {e}")))?; - let (transcript, recovery) = prepare_session(&mut store, &child_id) - .map_err(|e| ToolError::Failed(format!("store: {e}")))?; - let task = if transcript.is_empty() { - prompt.to_string() - } else { - "The previous attempt above was interrupted. Continue and complete the \ - original task, then give your final report." - .to_string() - }; + let child_cwd = self.place_child(&store, &child, use_worktree)?; drop(store); - let system = format!( - "{}\n\nYou are a bullpen relief agent handling one delegated task. \ - Work it to completion and end with a final report — your last \ - message goes back to the coordinating agent, which cannot see \ - your intermediate steps.{}", - self.config.system, - if mode == "inspect" { - " You have read-only tools." - } else { - "" - } - ); - - let journal = StoreJournal::new( - Store::open(&self.config.store_path) - .map_err(|e| ToolError::Failed(format!("store: {e}")))?, - &child_id, - ); - let mut child_ctx = ToolCtx::new(self.config.workspace.clone()); - if let Some(sandbox) = &self.config.sandbox { - child_ctx = child_ctx.with_sandbox(sandbox.clone()); - } - let mut agent = Agent::new( - self.provider.clone(), + let spec = ChildSpec { + provider: self.provider.clone(), + config: self.config.clone(), + child_id: child_id.clone(), + child_cwd, registry, - child_ctx, - AgentConfig { - model: self.config.model.clone(), - system, - max_turns: self.config.child_max_turns, - ..Default::default() - }, - ) - .with_transcript(transcript, child.usage) - .with_journal(Box::new(journal)); - - let result = tokio::time::timeout(self.config.child_timeout, agent.send(&task)).await; - let usage = agent.usage(); - let outcome = match result { - Ok(Ok(answer)) => Ok(format!( - "{answer}\n\n[child {short} · mode {mode} · {} in / {} out tokens{}]", - usage.input_tokens, - usage.output_tokens, - if recovery.is_some() { - " · recovered" + mode: mode.to_string(), + prompt: prompt.to_string(), + usage: child.usage, + }; + + if background { + let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel(); + self.cancels + .lock() + .unwrap() + .insert(child_id.clone(), cancel_tx); + let cancels = self.cancels.clone(); + let isolated = spec.child_cwd != self.config.workspace; + let short = short.to_string(); + tokio::spawn(async move { + // The outcome text is discarded — the durable child session + // is the record, and the job tool reads it from there. + let _ = run_child(spec, Some(cancel_rx)).await; + cancels.lock().unwrap().remove(&child_id); + }); + return Ok(format!( + "dispatched child {short} in the background (mode {mode}{}); \ + use the job tool to list, wait on, or cancel it", + if isolated { + " · isolated worktree" } else { "" - }, - )), - Ok(Err(e)) => Err(ToolError::Failed(format!( - "child {short} failed: {e} (session is saved and resumable)" - ))), - Err(_) => Err(ToolError::Failed(format!( - "child {short} timed out after {}s; its session is saved and \ - recoverable — calling agent again with the same task will continue it", - self.config.child_timeout.as_secs() - ))), - }; - let status = if outcome.is_ok() { - "completed" + } + )); + } + + run_child(spec, None).await + } +} + +/// Everything one child run needs, owned, so a background dispatch can move +/// it into its task. +struct ChildSpec { + provider: Arc, + config: PenConfig, + child_id: String, + child_cwd: PathBuf, + registry: Registry, + mode: String, + prompt: String, + usage: bullpen_llm::Usage, +} + +/// Run one child to its outcome: acquire exclusive ownership, recover and +/// continue anything a previous process left open, run the agent, record +/// the terminal state. `cancel` (background children only) resolves the run +/// early: the child finishes as failed and stays resumable. +async fn run_child( + spec: ChildSpec, + cancel: Option>, +) -> Result { + let ChildSpec { + provider, + config, + child_id, + child_cwd, + registry, + mode, + prompt, + usage, + } = spec; + let short = child_id[..8].to_string(); + + // Fresh or interrupted child: acquire the same exclusive ownership used + // by top-level CLI runs before recovery or provider activity. This + // prevents a manual `bullpen run -r ` from racing the pen. + let mut store = + Store::open(&config.store_path).map_err(|e| ToolError::Failed(format!("store: {e}")))?; + let mut session_worker = SessionWorker::acquire(&config.store_path, &child_id) + .map_err(|e| ToolError::Failed(format!("child {short}: {e}")))?; + session_worker + .start(&mut store) + .map_err(|e| ToolError::Failed(format!("child {short}: {e}")))?; + let (transcript, recovery) = prepare_session(&mut store, &child_id) + .map_err(|e| ToolError::Failed(format!("store: {e}")))?; + let task = if transcript.is_empty() { + prompt + } else { + "The previous attempt above was interrupted. Continue and complete the \ + original task, then give your final report." + .to_string() + }; + drop(store); + + let system = format!( + "{}\n\nYou are a bullpen relief agent handling one delegated task. \ + Work it to completion and end with a final report — your last \ + message goes back to the coordinating agent, which cannot see \ + your intermediate steps.{}", + config.system, + if mode == "inspect" { + " You have read-only tools." } else { - "failed" - }; - session_worker - .finish(status) - .map_err(|e| ToolError::Failed(format!("child {short}: {e}")))?; - outcome + "" + } + ); + + // An isolated child gets a sandbox rebased onto its worktree (same + // network policy), widened to the git dirs a linked worktree commits + // through; a shared-checkout child inherits the parent's as-is. + let sandbox = config.sandbox.as_ref().map(|sb| { + if child_cwd == config.workspace { + sb.clone() + } else { + let base = if sb.capabilities().allow_network { + bullpen_sandbox::Sandbox::workspace(&child_cwd) + } else { + bullpen_sandbox::Sandbox::strict(&child_cwd) + }; + Arc::new(base.allowing_writes(worktree::git_write_roots(&child_cwd))) + } + }); + + let journal = StoreJournal::new( + Store::open(&config.store_path).map_err(|e| ToolError::Failed(format!("store: {e}")))?, + &child_id, + ); + let mut child_ctx = ToolCtx::new(child_cwd); + if let Some(sandbox) = sandbox { + child_ctx = child_ctx.with_sandbox(sandbox); + } + let mut agent = Agent::new( + provider, + registry, + child_ctx, + AgentConfig { + model: config.model.clone(), + system, + max_turns: config.child_max_turns, + ..Default::default() + }, + ) + .with_transcript(transcript, usage) + .with_journal(Box::new(journal)); + + enum Ran { + Done(Result), + TimedOut, + Cancelled, } + let work = tokio::time::timeout(config.child_timeout, agent.send(&task)); + let ran = match cancel { + Some(cancel) => tokio::select! { + result = work => match result { + Ok(r) => Ran::Done(r), + Err(_) => Ran::TimedOut, + }, + _ = cancel => Ran::Cancelled, + }, + None => match work.await { + Ok(r) => Ran::Done(r), + Err(_) => Ran::TimedOut, + }, + }; + let usage = agent.usage(); + let outcome = match ran { + Ran::Done(Ok(answer)) => Ok(format!( + "{answer}\n\n[child {short} · mode {mode} · {} in / {} out tokens{}]", + usage.input_tokens, + usage.output_tokens, + if recovery.is_some() { + " · recovered" + } else { + "" + }, + )), + Ran::Done(Err(e)) => Err(ToolError::Failed(format!( + "child {short} failed: {e} (session is saved and resumable)" + ))), + Ran::TimedOut => Err(ToolError::Failed(format!( + "child {short} timed out after {}s; its session is saved and \ + recoverable — calling agent again with the same task will continue it", + config.child_timeout.as_secs() + ))), + Ran::Cancelled => Err(ToolError::Failed(format!( + "child {short} was cancelled; its session is saved and resumable" + ))), + }; + let status = if outcome.is_ok() { + "completed" + } else { + "failed" + }; + session_worker + .finish(status) + .map_err(|e| ToolError::Failed(format!("child {short}: {e}")))?; + outcome } #[cfg(test)] @@ -558,6 +783,207 @@ mod tests { )),); } + #[test] + fn isolated_and_background_children_are_parallel_safe() { + let dir = tempfile::tempdir().unwrap(); + let pen = PenTool::new(FakeProvider::new(vec![]), "p", config(&dir)); + assert!(pen.parallel_safe(&json!({"mode": "work", "worktree": true, "prompt": "x"}))); + assert!(pen.parallel_safe(&json!({"mode": "work", "background": true, "prompt": "x"}))); + assert!(!pen.parallel_safe(&json!({"mode": "work", "prompt": "x"}))); + } + + /// A repository at `root` with one committed file (see worktree tests). + fn init_repo(root: &std::path::Path) { + std::fs::create_dir_all(root).unwrap(); + let git = |args: &[&str]| { + let out = std::process::Command::new("git") + .arg("-C") + .arg(root) + .args(["-c", "user.email=t@example.com", "-c", "user.name=t"]) + .args(args) + .output() + .unwrap(); + assert!(out.status.success(), "git {args:?}: {out:?}"); + }; + git(&["init"]); + std::fs::write(root.join("f.txt"), "x").unwrap(); + git(&["add", "f.txt"]); + git(&["commit", "-m", "init"]); + } + + #[tokio::test] + async fn worktree_child_works_in_its_own_tree() { + let dir = tempfile::tempdir().unwrap(); + let repo = dir.path().join("repo"); + init_repo(&repo); + let mut cfg = config(&dir); + cfg.workspace = repo.clone(); + let parent = parent(&dir); + let pen = PenTool::new( + FakeProvider::new(vec![ + response( + vec![ContentBlock::ToolUse { + id: "tu_1".into(), + name: "bash".into(), + input: json!({"command": "echo done > marker.txt"}), + }], + StopReason::ToolUse, + ), + text_response("isolated work done"), + ]), + &parent, + cfg, + ); + + let out = pen + .run( + &tool_ctx(), + "call_1", + json!({"prompt": "work", "mode": "work", "worktree": true}), + ) + .await + .unwrap(); + assert!(out.contains("isolated work done"), "{out}"); + + let store = Store::open(&dir.path().join("t.db")).unwrap(); + let child = store + .get_session(&child_session_id(&parent, "call_1")) + .unwrap(); + assert_eq!( + child.worktree_branch.as_deref(), + Some(format!("bullpen/{}", child.id).as_str()) + ); + let wt = PathBuf::from(child.worktree_path.unwrap()); + // The marker landed in the worktree beside the store, not in the + // shared checkout. + assert!(wt.join("marker.txt").is_file()); + assert!(!repo.join("marker.txt").exists()); + assert!(wt.starts_with(dir.path())); + } + + #[tokio::test] + async fn worktree_needs_work_mode_and_a_repository() { + let dir = tempfile::tempdir().unwrap(); + let parent = parent(&dir); + let pen = PenTool::new(FakeProvider::new(vec![]), &parent, config(&dir)); + + let err = pen + .run(&tool_ctx(), "c1", json!({"prompt": "x", "worktree": true})) + .await + .unwrap_err(); + assert!(matches!(err, ToolError::InvalidInput(_)), "{err}"); + + // Work mode, but the workspace (a temp dir) is not a repository — + // and isolation never degrades to the shared checkout. + let err = pen + .run( + &tool_ctx(), + "c2", + json!({"prompt": "x", "mode": "work", "worktree": true}), + ) + .await + .unwrap_err(); + assert!( + err.to_string().contains("not inside a git repository"), + "{err}" + ); + } + + #[tokio::test] + async fn background_dispatch_returns_immediately_and_job_waits() { + let dir = tempfile::tempdir().unwrap(); + let parent = parent(&dir); + let pen = PenTool::new( + FakeProvider::new(vec![text_response("bg answer")]), + &parent, + config(&dir), + ); + let job = pen.job_tool(); + + let out = pen + .run( + &tool_ctx(), + "call_1", + json!({"prompt": "task", "background": true}), + ) + .await + .unwrap(); + assert!(out.contains("dispatched child"), "{out}"); + + let short = child_session_id(&parent, "call_1")[..8].to_string(); + let waited = job + .run( + &tool_ctx(), + "j1", + json!({"action": "wait", "id": short, "timeout_seconds": 30}), + ) + .await + .unwrap(); + assert!(waited.contains("bg answer"), "{waited}"); + assert!(waited.contains("completed"), "{waited}"); + } + + #[tokio::test] + async fn cancelled_background_child_finishes_failed_and_stays_resumable() { + let dir = tempfile::tempdir().unwrap(); + let parent = parent(&dir); + let pen = PenTool::new( + Arc::new(crate::testutil::HangingProvider), + &parent, + config(&dir), + ); + let job = pen.job_tool(); + + pen.run( + &tool_ctx(), + "call_1", + json!({"prompt": "never ends", "background": true}), + ) + .await + .unwrap(); + let short = child_session_id(&parent, "call_1")[..8].to_string(); + + // The dispatch is visible on the coordination plane... + let listed = job + .run(&tool_ctx(), "j1", json!({"action": "list"})) + .await + .unwrap(); + assert!(listed.contains(&short), "{listed}"); + + // ...and a second spawn of the same call reports, not respawns. + let again = pen + .run( + &tool_ctx(), + "call_1", + json!({"prompt": "never ends", "background": true}), + ) + .await + .unwrap(); + assert!(again.contains("already running"), "{again}"); + + let out = job + .run(&tool_ctx(), "j2", json!({"action": "cancel", "id": short})) + .await + .unwrap(); + assert!(out.contains("cancel signalled"), "{out}"); + + // wait observes the terminal state the cancelled child recorded. + let err = job + .run( + &tool_ctx(), + "j3", + json!({"action": "wait", "id": short, "timeout_seconds": 30}), + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("failed"), "{err}"); + let store = Store::open(&dir.path().join("t.db")).unwrap(); + let child = store + .get_session(&child_session_id(&parent, "call_1")) + .unwrap(); + assert_eq!(child.status, "failed"); + } + #[tokio::test] async fn bad_mode_is_rejected() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/harness/src/testutil.rs b/crates/harness/src/testutil.rs index 0b252dd..04a0a06 100644 --- a/crates/harness/src/testutil.rs +++ b/crates/harness/src/testutil.rs @@ -30,6 +30,20 @@ impl Provider for FakeProvider { } } +/// A provider whose call never resolves — for cancellation and liveness +/// tests. +pub struct HangingProvider; + +#[async_trait::async_trait] +impl Provider for HangingProvider { + fn name(&self) -> &str { + "hang" + } + async fn complete(&self, _: &Request) -> Result { + std::future::pending().await + } +} + pub fn response(blocks: Vec, stop: StopReason) -> Response { Response { content: blocks, diff --git a/crates/harness/src/todo.rs b/crates/harness/src/todo.rs new file mode 100644 index 0000000..94032ca --- /dev/null +++ b/crates/harness/src/todo.rs @@ -0,0 +1,325 @@ +//! The todo tool: a durable session plan. +//! +//! The plan lives in the store, not in the model's context window — it +//! survives crashes, resumes, and compaction like everything else in the +//! session. Two choices carry the design: +//! +//! - **Deterministic item ids** (`uuidv5(session, call_id, index)`) make a +//! replayed `add` converge on the same rows instead of duplicating them, +//! so the whole tool is replay-safe. +//! - **The store owns the one-active-item invariant**: marking an item +//! `in_progress` returns any other active item to `pending`. The model +//! cannot talk itself into three parallel "current" tasks. +//! +//! Every action returns the rendered plan, so the model always acts on the +//! current state rather than its memory of it. + +use std::path::PathBuf; + +use bullpen_llm::ToolSpec; +use bullpen_store::{Store, StoreError, Todo}; +use bullpen_tools::{Tool, ToolCtx, ToolError}; +use serde_json::{Value, json}; + +pub struct TodoTool { + store_path: PathBuf, + session_id: String, +} + +impl TodoTool { + pub fn new(store_path: PathBuf, session_id: impl Into) -> Self { + Self { + store_path, + session_id: session_id.into(), + } + } +} + +/// Deterministic item identity: same session + same tool call + same index +/// → same todo. What makes a replayed `add` reattach instead of duplicate. +pub fn todo_id(session_id: &str, call_id: &str, index: usize) -> String { + uuid::Uuid::new_v5( + &uuid::Uuid::NAMESPACE_OID, + format!("bullpen-todo:{session_id}:{call_id}:{index}").as_bytes(), + ) + .to_string() +} + +fn terr(e: StoreError) -> ToolError { + match e { + StoreError::NotFound(p) => ToolError::InvalidInput(format!("no todo matches `{p}`")), + StoreError::Ambiguous(p) => ToolError::InvalidInput(format!( + "todo id `{p}` is ambiguous — use a longer prefix from the list" + )), + other => ToolError::Failed(format!("store: {other}")), + } +} + +fn render(todos: &[Todo]) -> String { + if todos.is_empty() { + return "The plan is empty.".into(); + } + let done = todos.iter().filter(|t| t.status == "completed").count(); + let mut out = format!("Plan ({done} of {} done):\n", todos.len()); + for t in todos { + let mark = match t.status.as_str() { + "completed" => "[x]", + "in_progress" => "[>]", + _ => "[ ]", + }; + out.push_str(&format!(" {} {mark} {}\n", &t.id[..8], t.content)); + } + out +} + +#[async_trait::async_trait] +impl Tool for TodoTool { + fn name(&self) -> &'static str { + "todo" + } + + fn spec(&self) -> ToolSpec { + ToolSpec { + name: "todo".into(), + description: "Track the plan for this session as a durable todo list. \ + `add` appends items, `start` marks one in progress (any \ + other active item returns to pending — one thing at a \ + time), `done` completes one, `remove` drops one, `list` \ + shows the plan. Items are addressed by the id prefix \ + shown in the list. Every action returns the current \ + plan. Use it for multi-step work; keep it current as \ + steps finish." + .into(), + input_schema: json!({ + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["add", "start", "done", "remove", "list"], + "description": "What to do to the plan" + }, + "items": { + "type": "array", + "items": {"type": "string"}, + "description": "For `add`: the items to append, in order" + }, + "id": { + "type": "string", + "description": "For `start`/`done`/`remove`: the item's id (any unique prefix)" + } + }, + "required": ["action"] + }), + } + } + + fn parallel_safe(&self, input: &Value) -> bool { + // Reading the plan can ride alongside anything; mutations keep the + // default serial ordering so a batch applies in the order the model + // issued it. + input.get("action").and_then(Value::as_str) == Some("list") + } + + fn replay_safe(&self) -> bool { + // Adds converge via deterministic ids; status changes and removes + // are idempotent by construction. + true + } + + async fn run(&self, _ctx: &ToolCtx, call_id: &str, input: Value) -> Result { + let action = input.get("action").and_then(Value::as_str).ok_or_else(|| { + ToolError::InvalidInput("missing required string field `action`".into()) + })?; + let mut store = Store::open(&self.store_path).map_err(terr)?; + + match action { + "add" => { + let items: Vec<&str> = input + .get("items") + .and_then(Value::as_array) + .map(|a| a.iter().filter_map(Value::as_str).collect()) + .unwrap_or_default(); + if items.is_empty() { + return Err(ToolError::InvalidInput( + "`add` needs a non-empty `items` array of strings".into(), + )); + } + for (index, content) in items.iter().enumerate() { + let id = todo_id(&self.session_id, call_id, index); + store + .add_todo(&self.session_id, &id, content) + .map_err(terr)?; + } + } + "start" | "done" => { + let prefix = input.get("id").and_then(Value::as_str).ok_or_else(|| { + ToolError::InvalidInput(format!("`{action}` needs an `id` prefix")) + })?; + let status = if action == "start" { + "in_progress" + } else { + "completed" + }; + store + .set_todo_status(&self.session_id, prefix, status) + .map_err(terr)?; + } + "remove" => { + let prefix = input.get("id").and_then(Value::as_str).ok_or_else(|| { + ToolError::InvalidInput("`remove` needs an `id` prefix".into()) + })?; + store.remove_todo(&self.session_id, prefix).map_err(terr)?; + } + "list" => {} + other => { + return Err(ToolError::InvalidInput(format!( + "unknown action `{other}` (expected add, start, done, remove, or list)" + ))); + } + } + + Ok(render(&store.list_todos(&self.session_id).map_err(terr)?)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn setup() -> (tempfile::TempDir, TodoTool, String) { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.db"); + let session = Store::open(&path) + .unwrap() + .create_session("/tmp", "fake", "m") + .unwrap(); + let tool = TodoTool::new(path, &session.id); + (dir, tool, session.id) + } + + fn ctx() -> ToolCtx { + ToolCtx::new(std::env::temp_dir()) + } + + #[tokio::test] + async fn add_appends_and_renders_the_plan() { + let (_dir, tool, _) = setup(); + let out = tool + .run( + &ctx(), + "c1", + json!({"action": "add", "items": ["read", "write"]}), + ) + .await + .unwrap(); + assert!(out.contains("Plan (0 of 2 done):"), "{out}"); + assert!(out.contains("[ ] read"), "{out}"); + assert!(out.contains("[ ] write"), "{out}"); + } + + #[tokio::test] + async fn replayed_add_does_not_duplicate() { + let (_dir, tool, _) = setup(); + let input = json!({"action": "add", "items": ["once"]}); + tool.run(&ctx(), "c1", input.clone()).await.unwrap(); + let out = tool.run(&ctx(), "c1", input).await.unwrap(); + assert!(out.contains("of 1 done"), "{out}"); + } + + #[tokio::test] + async fn start_and_done_by_prefix_keep_one_item_active() { + let (_dir, tool, session) = setup(); + tool.run(&ctx(), "c1", json!({"action": "add", "items": ["a", "b"]})) + .await + .unwrap(); + let first = &todo_id(&session, "c1", 0)[..8]; + let second = &todo_id(&session, "c1", 1)[..8]; + + let out = tool + .run(&ctx(), "c2", json!({"action": "start", "id": first})) + .await + .unwrap(); + assert!(out.contains(&format!("{first} [>] a")), "{out}"); + + // Starting the second returns the first to pending. + let out = tool + .run(&ctx(), "c3", json!({"action": "start", "id": second})) + .await + .unwrap(); + assert!(out.contains(&format!("{first} [ ] a")), "{out}"); + assert!(out.contains(&format!("{second} [>] b")), "{out}"); + + let out = tool + .run(&ctx(), "c4", json!({"action": "done", "id": second})) + .await + .unwrap(); + assert!(out.contains("Plan (1 of 2 done):"), "{out}"); + assert!(out.contains(&format!("{second} [x] b")), "{out}"); + } + + #[tokio::test] + async fn remove_and_empty_render() { + let (_dir, tool, session) = setup(); + tool.run(&ctx(), "c1", json!({"action": "add", "items": ["only"]})) + .await + .unwrap(); + let id = &todo_id(&session, "c1", 0)[..8]; + let out = tool + .run(&ctx(), "c2", json!({"action": "remove", "id": id})) + .await + .unwrap(); + assert_eq!(out, "The plan is empty."); + } + + #[tokio::test] + async fn bad_inputs_are_invalid_not_failed() { + let (_dir, tool, _) = setup(); + for input in [ + json!({}), + json!({"action": "add"}), + json!({"action": "add", "items": []}), + json!({"action": "start"}), + json!({"action": "done", "id": "zzz"}), + json!({"action": "yolo"}), + ] { + let err = tool.run(&ctx(), "c", input).await.unwrap_err(); + assert!(matches!(err, ToolError::InvalidInput(_)), "{err}"); + } + } + + #[tokio::test] + async fn ambiguous_prefix_asks_for_more() { + let (_dir, tool, session) = setup(); + // Two items from one call share the session/call prefix only if the + // uuids collide on the first character — force ambiguity with the + // empty prefix instead, which matches everything. + tool.run(&ctx(), "c1", json!({"action": "add", "items": ["a", "b"]})) + .await + .unwrap(); + let err = tool + .run(&ctx(), "c2", json!({"action": "done", "id": ""})) + .await + .unwrap_err(); + assert!(err.to_string().contains("ambiguous"), "{err}"); + drop(session); + } + + #[test] + fn only_list_is_parallel_safe_and_replay_is_safe() { + let dir = tempfile::tempdir().unwrap(); + let tool = TodoTool::new(dir.path().join("t.db"), "s"); + assert!(tool.parallel_safe(&json!({"action": "list"}))); + assert!(!tool.parallel_safe(&json!({"action": "add", "items": ["x"]}))); + assert!(!tool.parallel_safe(&json!({"action": "done", "id": "a"}))); + assert!(tool.replay_safe()); + } + + #[test] + fn todo_ids_are_deterministic() { + let a = todo_id("s1", "c1", 0); + assert_eq!(a, todo_id("s1", "c1", 0)); + assert_ne!(a, todo_id("s1", "c1", 1)); + assert_ne!(a, todo_id("s1", "c2", 0)); + assert_ne!(a, todo_id("s2", "c1", 0)); + } +} diff --git a/crates/cli/src/worktree.rs b/crates/harness/src/worktree.rs similarity index 98% rename from crates/cli/src/worktree.rs rename to crates/harness/src/worktree.rs index 18d8d7a..dd2081b 100644 --- a/crates/cli/src/worktree.rs +++ b/crates/harness/src/worktree.rs @@ -29,6 +29,14 @@ pub fn worktree_path(session_id: &str) -> PathBuf { worktree_dir(&bullpen_store::home_dir(), session_id) } +/// The same layout, anchored to a specific store: worktrees live alongside +/// the database. For the default store this is exactly [`worktree_path`]; +/// for a pen pointed at an isolated store (tests, `$BULLPEN_HOME`), the +/// worktrees follow the store instead of the ambient environment. +pub fn worktree_path_for_store(store_path: &Path, session_id: &str) -> PathBuf { + worktree_dir(store_path.parent().unwrap_or(Path::new(".")), session_id) +} + fn worktree_dir(home: &Path, session_id: &str) -> PathBuf { home.join("worktrees").join(session_id) } diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index 896628f..f9c82f7 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -12,6 +12,7 @@ serde_json.workspace = true thiserror.workspace = true uuid.workspace = true fs2.workspace = true +libc.workspace = true [dev-dependencies] tempfile = "3" diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 4fd98b0..5c5e405 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -103,6 +103,16 @@ pub struct OpenRun { pub records: Vec, } +/// One item on a session's durable plan. `status` is `pending`, +/// `in_progress`, or `completed`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Todo { + pub id: String, + pub position: i64, + pub content: String, + pub status: String, +} + /// The directory holding bullpen's own state: `$BULLPEN_HOME` when set to a /// non-empty path, otherwise `~/.bullpen`. pub fn home_dir() -> PathBuf { @@ -147,7 +157,7 @@ impl Store { let observed: i64 = self .conn .query_row("SELECT * FROM pragma_user_version", [], |r| r.get(0))?; - if observed >= 7 { + if observed >= 8 { return Ok(()); } @@ -249,6 +259,21 @@ impl Store { PRAGMA user_version = 7;", )?; } + if version < 8 { + tx.execute_batch( + "CREATE TABLE todos ( + id TEXT PRIMARY KEY, + session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE, + position INTEGER NOT NULL, + content TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + CREATE INDEX todos_by_session ON todos(session_id, position); + PRAGMA user_version = 8;", + )?; + } tx.commit()?; Ok(()) } @@ -299,6 +324,19 @@ impl Store { self.get_session(child_id) } + /// A session's pen children, oldest first. + pub fn list_children(&self, parent_id: &str) -> Result, StoreError> { + let mut stmt = self.conn.prepare( + "SELECT id, title, cwd, provider, model, input_tokens, output_tokens, + created_at, updated_at, parent_session_id, status, pid, + worktree_path, worktree_branch + FROM sessions WHERE parent_session_id = ?1 ORDER BY created_at, id", + )?; + Ok(stmt + .query_map(params![parent_id], row_to_session)? + .collect::>()?) + } + /// How many children a session has spawned — the durable count budget. pub fn count_children(&self, parent_id: &str) -> Result { Ok(self.conn.query_row( @@ -719,6 +757,121 @@ impl Store { ))), } } + + // ── Todos (the session plan) ──────────────────────────────────────────── + + /// Append a todo with a caller-provisioned (deterministic) id at the end + /// of the session's plan. Idempotent: an id that already exists is left + /// untouched — this is what makes a replayed `add` safe. + pub fn add_todo( + &mut self, + session_id: &str, + id: &str, + content: &str, + ) -> Result<(), StoreError> { + let tx = self + .conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; + let exists: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM todos WHERE id = ?1)", + params![id], + |r| r.get(0), + )?; + if !exists { + // Aggregate subquery: one row even for an empty plan. + tx.execute( + "INSERT INTO todos (id, session_id, position, content) + SELECT ?1, ?2, COALESCE(MAX(position), 0) + 1, ?3 + FROM todos WHERE session_id = ?2", + params![id, session_id, content], + )?; + } + tx.commit()?; + Ok(()) + } + + /// Move a todo (resolved by id prefix) to `status`. Marking one + /// `in_progress` returns every other in-progress todo to `pending`: + /// the store owns the one-active-item invariant, not the model. + pub fn set_todo_status( + &mut self, + session_id: &str, + prefix: &str, + status: &str, + ) -> Result { + let tx = self + .conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; + let todo = resolve_todo(&tx, session_id, prefix)?; + if status == "in_progress" { + tx.execute( + "UPDATE todos SET status = 'pending', updated_at = datetime('now') + WHERE session_id = ?1 AND status = 'in_progress' AND id != ?2", + params![session_id, todo.id], + )?; + } + tx.execute( + "UPDATE todos SET status = ?2, updated_at = datetime('now') WHERE id = ?1", + params![todo.id, status], + )?; + tx.commit()?; + Ok(Todo { + status: status.to_string(), + ..todo + }) + } + + /// Delete a todo, resolved by id prefix. + pub fn remove_todo(&mut self, session_id: &str, prefix: &str) -> Result { + let tx = self + .conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; + let todo = resolve_todo(&tx, session_id, prefix)?; + tx.execute("DELETE FROM todos WHERE id = ?1", params![todo.id])?; + tx.commit()?; + Ok(todo) + } + + /// The session's plan in order. + pub fn list_todos(&self, session_id: &str) -> Result, StoreError> { + let mut stmt = self.conn.prepare( + "SELECT id, position, content, status FROM todos + WHERE session_id = ?1 ORDER BY position", + )?; + Ok(stmt + .query_map(params![session_id], row_to_todo)? + .collect::>()?) + } +} + +/// Resolve a todo by id prefix within one session, mirroring session prefix +/// resolution: zero matches is `NotFound`, two or more is `Ambiguous`. +fn resolve_todo( + tx: &rusqlite::Transaction<'_>, + session_id: &str, + prefix: &str, +) -> Result { + let mut stmt = tx.prepare( + "SELECT id, position, content, status FROM todos + WHERE session_id = ?1 AND id LIKE ?2 || '%' LIMIT 2", + )?; + let mut matches: Vec = stmt + .query_map(params![session_id, prefix], row_to_todo)? + .collect::>()?; + match matches.len() { + 0 => Err(StoreError::NotFound(prefix.to_string())), + 1 => Ok(matches.remove(0)), + _ => Err(StoreError::Ambiguous(prefix.to_string())), + } +} + +fn row_to_todo(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(Todo { + id: row.get(0)?, + position: row.get(1)?, + content: row.get(2)?, + status: row.get(3)?, + }) } /// Allocate the next per-session sequence number inside the caller's @@ -1112,7 +1265,7 @@ mod tests { let version: i64 = conn .query_row("SELECT * FROM pragma_user_version", [], |row| row.get(0)) .unwrap(); - assert_eq!(version, 7); + assert_eq!(version, 8); } #[test] @@ -1141,6 +1294,111 @@ mod tests { assert_eq!(failed.pid, None); } + #[test] + fn todos_append_in_order_and_adds_are_idempotent() { + let (_dir, mut store) = store(); + let s = store.create_session("/tmp", "anthropic", "m").unwrap(); + + store.add_todo(&s.id, "t1", "first").unwrap(); + store.add_todo(&s.id, "t2", "second").unwrap(); + // Replayed add: same id, no duplicate, positions untouched. + store.add_todo(&s.id, "t1", "first").unwrap(); + + let todos = store.list_todos(&s.id).unwrap(); + assert_eq!( + todos + .iter() + .map(|t| (t.position, t.content.as_str(), t.status.as_str())) + .collect::>(), + vec![(1, "first", "pending"), (2, "second", "pending")] + ); + } + + #[test] + fn only_one_todo_is_ever_in_progress() { + let (_dir, mut store) = store(); + let s = store.create_session("/tmp", "anthropic", "m").unwrap(); + store.add_todo(&s.id, "t1", "first").unwrap(); + store.add_todo(&s.id, "t2", "second").unwrap(); + + store.set_todo_status(&s.id, "t1", "in_progress").unwrap(); + store.set_todo_status(&s.id, "t2", "in_progress").unwrap(); + + let statuses: Vec = store + .list_todos(&s.id) + .unwrap() + .into_iter() + .map(|t| t.status) + .collect(); + assert_eq!(statuses, vec!["pending", "in_progress"]); + } + + #[test] + fn todo_prefix_resolution_mirrors_sessions() { + let (_dir, mut store) = store(); + let s = store.create_session("/tmp", "anthropic", "m").unwrap(); + store.add_todo(&s.id, "abc-1", "one").unwrap(); + store.add_todo(&s.id, "abd-2", "two").unwrap(); + + assert!(matches!( + store.set_todo_status(&s.id, "ab", "completed"), + Err(StoreError::Ambiguous(_)) + )); + assert!(matches!( + store.remove_todo(&s.id, "zzz"), + Err(StoreError::NotFound(_)) + )); + + let done = store.set_todo_status(&s.id, "abc", "completed").unwrap(); + assert_eq!(done.status, "completed"); + let removed = store.remove_todo(&s.id, "abd").unwrap(); + assert_eq!(removed.content, "two"); + assert_eq!(store.list_todos(&s.id).unwrap().len(), 1); + + // Another session's todos are invisible to this one's prefixes. + let other = store.create_session("/tmp", "anthropic", "m").unwrap(); + assert!(matches!( + store.set_todo_status(&other.id, "abc", "completed"), + Err(StoreError::NotFound(_)) + )); + } + + #[test] + fn migrates_v7_sessions_adding_todos() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("v7.db"); + { + let conn = Connection::open(&path).unwrap(); + conn.execute_batch( + "CREATE TABLE sessions ( + id TEXT PRIMARY KEY, title TEXT NOT NULL DEFAULT '', + cwd TEXT NOT NULL, model TEXT NOT NULL, + input_tokens INTEGER NOT NULL DEFAULT 0, + output_tokens INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')), + provider TEXT NOT NULL DEFAULT 'anthropic', + next_seq INTEGER NOT NULL DEFAULT 1, + parent_session_id TEXT, + status TEXT NOT NULL DEFAULT 'idle', + pid INTEGER, + worktree_path TEXT, + worktree_branch TEXT, + worker_generation TEXT + ); + INSERT INTO sessions (id, cwd, model) VALUES ('s1', '/tmp', 'm'); + PRAGMA user_version = 7;", + ) + .unwrap(); + } + + let mut store = Store::open(&path).unwrap(); + // Sessions that predate the feature simply have an empty plan. + assert_eq!(store.list_todos("s1").unwrap(), vec![]); + store.add_todo("s1", "t1", "works").unwrap(); + assert_eq!(store.list_todos("s1").unwrap().len(), 1); + } + #[test] fn set_worktree_round_trips_through_list_and_resolve() { let (_dir, store) = store(); diff --git a/crates/store/src/status.rs b/crates/store/src/status.rs index ca70592..ec5ac7b 100644 --- a/crates/store/src/status.rs +++ b/crates/store/src/status.rs @@ -42,6 +42,18 @@ impl AgentStatus { } } +/// Whether `pid` is a live process. Uses `kill(pid, 0)`: success or an +/// `EPERM` both mean the process exists. +pub fn pid_alive(pid: i64) -> bool { + if pid <= 0 { + return false; + } + // SAFETY: kill with signal 0 performs only the existence/permission + // check and never delivers a signal. + let rc = unsafe { libc::kill(pid as libc::pid_t, 0) }; + rc == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) +} + /// Derive the display state. `alive` is whether the session's recorded pid is /// a live process (the caller checks the OS); it only matters while the /// stored status is `running`. diff --git a/crates/tools/Cargo.toml b/crates/tools/Cargo.toml index c52ee42..a40c0cb 100644 --- a/crates/tools/Cargo.toml +++ b/crates/tools/Cargo.toml @@ -8,11 +8,14 @@ license.workspace = true bullpen-llm.workspace = true bullpen-sandbox.workspace = true async-trait.workspace = true +futures.workspace = true globset.workspace = true +reqwest.workspace = true ignore.workspace = true regex.workspace = true serde.workspace = true serde_json.workspace = true +sha2.workspace = true thiserror.workspace = true tokio.workspace = true diff --git a/crates/tools/src/ask.rs b/crates/tools/src/ask.rs new file mode 100644 index 0000000..02f1f19 --- /dev/null +++ b/crates/tools/src/ask.rs @@ -0,0 +1,231 @@ +//! Structured follow-up questions for interactive runs. +//! +//! The tool is transport-agnostic: the application injects an [`Asker`] +//! (the CLI's reads one line from the controlling terminal), and a run +//! with nobody on the other end — background, JSON-streamed, piped — +//! registers the *detached* variant, which fails the call with a clear +//! reason instead of blocking on input nobody will ever type. The model +//! always sees the same tool; only the answerer changes. + +use std::sync::Arc; + +use bullpen_llm::ToolSpec; +use serde_json::{Value, json}; + +use crate::{Tool, ToolCtx, ToolError, required_str}; + +/// Answers questions on behalf of whoever is driving this run. +#[async_trait::async_trait] +pub trait Asker: Send + Sync { + /// Put the rendered `prompt` to the human and return their raw reply. + async fn ask(&self, prompt: &str) -> Result; +} + +pub struct Ask { + asker: Option>, +} + +impl Ask { + /// Someone is on the terminal; questions reach them. + pub fn interactive(asker: Arc) -> Self { + Self { asker: Some(asker) } + } + + /// Nobody is listening (background, `--json`, piped stdin). The tool + /// still registers so the model gets a reason, not an unknown-tool + /// error. + pub fn detached() -> Self { + Self { asker: None } + } +} + +/// The question as the human sees it. Options are numbered so a reply can +/// be just the number. +fn render(question: &str, options: &[&str]) -> String { + let mut out = question.to_string(); + for (index, option) in options.iter().enumerate() { + out.push_str(&format!("\n {}. {option}", index + 1)); + } + if !options.is_empty() { + out.push_str("\nReply with a number or free text."); + } + out +} + +/// A reply of `2` against options means the second option; anything else +/// is taken verbatim. The tool owns this mapping so every transport gets +/// it, and gets it tested. +fn resolve<'a>(answer: &'a str, options: &[&'a str]) -> &'a str { + match answer.parse::() { + Ok(n) if (1..=options.len()).contains(&n) => options[n - 1], + _ => answer, + } +} + +#[async_trait::async_trait] +impl Tool for Ask { + fn name(&self) -> &'static str { + "ask" + } + + fn spec(&self) -> ToolSpec { + ToolSpec { + name: "ask".into(), + description: "Ask the human driving this run one question and \ + get their answer. Optional `options` render as a \ + numbered choice list; the reply is the chosen \ + option's text, or free text. Use it only when \ + genuinely blocked on a decision the task cannot \ + settle — in a detached run there is no one to \ + answer and the call fails with that reason." + .into(), + input_schema: json!({ + "type": "object", + "properties": { + "question": { + "type": "string", + "description": "The question, complete enough to answer without reading the transcript" + }, + "options": { + "type": "array", + "items": {"type": "string"}, + "description": "Choices to offer (optional)" + } + }, + "required": ["question"] + }), + } + } + + async fn run(&self, _ctx: &ToolCtx, _call_id: &str, input: Value) -> Result { + let question = required_str(&input, "question")?; + let options: Vec<&str> = input + .get("options") + .and_then(Value::as_array) + .map(|a| a.iter().filter_map(Value::as_str).collect()) + .unwrap_or_default(); + + let Some(asker) = &self.asker else { + return Err(ToolError::Failed( + "this run is detached — no one is listening to answer. \ + Decide with your best judgment and note the assumption in \ + your report." + .into(), + )); + }; + + let reply = asker.ask(&render(question, &options)).await?; + let answer = reply.trim(); + if answer.is_empty() { + return Err(ToolError::Failed( + "no answer was given (empty reply)".into(), + )); + } + Ok(resolve(answer, &options).to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Mutex; + + /// Replies with a fixed line and records the prompt it was shown. + struct FakeAsker { + reply: &'static str, + seen: Mutex, + } + + impl FakeAsker { + fn new(reply: &'static str) -> Arc { + Arc::new(Self { + reply, + seen: Mutex::new(String::new()), + }) + } + } + + #[async_trait::async_trait] + impl Asker for FakeAsker { + async fn ask(&self, prompt: &str) -> Result { + *self.seen.lock().unwrap() = prompt.to_string(); + Ok(format!("{}\n", self.reply)) + } + } + + fn ctx() -> ToolCtx { + ToolCtx::new(std::env::temp_dir()) + } + + #[tokio::test] + async fn detached_runs_fail_with_a_reason_not_a_hang() { + let err = Ask::detached() + .run(&ctx(), "t", json!({"question": "which?"})) + .await + .unwrap_err(); + assert!(err.to_string().contains("detached"), "{err}"); + } + + #[tokio::test] + async fn a_numbered_reply_selects_the_option() { + let asker = FakeAsker::new("2"); + let out = Ask::interactive(asker.clone()) + .run( + &ctx(), + "t", + json!({"question": "which db?", "options": ["sqlite", "postgres"]}), + ) + .await + .unwrap(); + assert_eq!(out, "postgres"); + let seen = asker.seen.lock().unwrap().clone(); + assert!(seen.contains("1. sqlite"), "{seen}"); + assert!(seen.contains("2. postgres"), "{seen}"); + assert!(seen.contains("number or free text"), "{seen}"); + } + + #[tokio::test] + async fn free_text_and_out_of_range_numbers_pass_through() { + let out = Ask::interactive(FakeAsker::new("use mysql actually")) + .run( + &ctx(), + "t", + json!({"question": "which?", "options": ["a", "b"]}), + ) + .await + .unwrap(); + assert_eq!(out, "use mysql actually"); + + let out = Ask::interactive(FakeAsker::new("7")) + .run( + &ctx(), + "t", + json!({"question": "which?", "options": ["a", "b"]}), + ) + .await + .unwrap(); + assert_eq!(out, "7"); + } + + #[tokio::test] + async fn empty_reply_and_missing_question_are_errors() { + let err = Ask::interactive(FakeAsker::new(" ")) + .run(&ctx(), "t", json!({"question": "q"})) + .await + .unwrap_err(); + assert!(err.to_string().contains("no answer"), "{err}"); + + let err = Ask::interactive(FakeAsker::new("x")) + .run(&ctx(), "t", json!({})) + .await + .unwrap_err(); + assert!(matches!(err, ToolError::InvalidInput(_))); + } + + #[test] + fn asking_is_neither_parallel_nor_replay_safe() { + let ask = Ask::detached(); + assert!(!ask.parallel_safe(&json!({"question": "q"}))); + assert!(!ask.replay_safe()); + } +} diff --git a/crates/tools/src/fs.rs b/crates/tools/src/fs.rs index 37b22a5..06db3e3 100644 --- a/crates/tools/src/fs.rs +++ b/crates/tools/src/fs.rs @@ -1,4 +1,12 @@ -//! File read, write, and single-occurrence edit. +//! File read, write, and edit. +//! +//! Reads are **hashline**: every line carries a `line#hash` anchor (the +//! first four hex chars of the line's SHA-256), and `edit_file` accepts a +//! patch of hunks addressed by those anchors. An anchor is a claim about +//! content, not just a position — so an edit against a file that drifted +//! is *detected*, and when the anchored line still exists uniquely +//! elsewhere it is *recovered* rather than misapplied. The exact-string +//! mode stays for one-off replacements. use bullpen_llm::ToolSpec; use serde_json::{Value, json}; @@ -7,6 +15,27 @@ use crate::{Tool, ToolCtx, ToolError, required_str, resolve_path, truncate_middl const MAX_READ_BYTES: usize = 262_144; // 256 KiB +/// A line's anchor hash: the first two bytes of its SHA-256, in hex. +/// Anchors pair it with a line number, and recovery trusts it only when +/// it matches exactly one line — a four-hex-char collision inside one +/// file therefore degrades to an explicit error, never a wrong edit. +pub(crate) fn line_hash(line: &str) -> String { + use sha2::{Digest, Sha256}; + let digest = Sha256::digest(line.as_bytes()); + format!("{:02x}{:02x}", digest[0], digest[1]) +} + +/// Render `lines[range]` as hashline output, 1-indexed. +fn hashlines(lines: &[&str], first: usize, limit: usize) -> String { + lines + .iter() + .enumerate() + .skip(first - 1) + .take(limit) + .map(|(i, line)| format!("{}#{}\t{line}\n", i + 1, line_hash(line))) + .collect() +} + pub struct ReadFile; #[async_trait::async_trait] @@ -18,15 +47,21 @@ impl Tool for ReadFile { fn spec(&self) -> ToolSpec { ToolSpec { name: "read_file".into(), - description: "Read a file, optionally a line window. Output is 1-indexed \ - `linecontent` and capped at 256 KiB." + description: "Read through one path: a file, a directory, or an \ + http(s) URL. Files render as 1-indexed \ + `line#hashcontent`, capped at 256 KiB — the \ + `line#hash` token is an anchor that `edit_file` \ + patches accept — with an optional line window. \ + Directories render a sorted listing. URLs are \ + fetched with GET (capped, 30s timeout); a sandbox \ + that denies network refuses them." .into(), input_schema: json!({ "type": "object", "properties": { - "path": {"type": "string"}, - "offset": {"type": "integer", "description": "1-indexed first line"}, - "limit": {"type": "integer", "description": "Max lines to return"} + "path": {"type": "string", "description": "File or directory path, or an http(s):// URL"}, + "offset": {"type": "integer", "description": "1-indexed first line (files only)"}, + "limit": {"type": "integer", "description": "Max lines to return (files only)"} }, "required": ["path"] }), @@ -42,7 +77,14 @@ impl Tool for ReadFile { } async fn run(&self, ctx: &ToolCtx, _call_id: &str, input: Value) -> Result { - let path = resolve_path(ctx, required_str(&input, "path")?); + let raw_path = required_str(&input, "path")?; + if raw_path.starts_with("http://") || raw_path.starts_with("https://") { + return read_url(ctx, raw_path).await; + } + let path = resolve_path(ctx, raw_path); + if tokio::fs::metadata(&path).await.is_ok_and(|m| m.is_dir()) { + return list_dir(&path).await; + } let raw = tokio::fs::read(&path) .await .map_err(|e| ToolError::Failed(format!("cannot read {}: {e}", path.display())))?; @@ -58,13 +100,8 @@ impl Tool for ReadFile { .and_then(Value::as_u64) .unwrap_or(u64::MAX) as usize; - let numbered: String = text - .lines() - .enumerate() - .skip(offset - 1) - .take(limit) - .map(|(i, line)| format!("{}\t{line}\n", i + 1)) - .collect(); + let lines: Vec<&str> = text.lines().collect(); + let numbered = hashlines(&lines, offset, limit); if numbered.is_empty() { return Ok(format!( @@ -76,6 +113,96 @@ impl Tool for ReadFile { } } +/// How much of a URL body is kept. Bounded while streaming, not after: a +/// response has no business filling memory just to be truncated. +const MAX_FETCH_BYTES: usize = 2 * 1024 * 1024; + +/// GET a URL. The sandbox's network capability governs this exactly as it +/// governs shell commands: `--sandbox-strict` cuts it. +async fn read_url(ctx: &ToolCtx, url: &str) -> Result { + if let Some(sandbox) = &ctx.sandbox + && !sandbox.capabilities().allow_network + { + return Err(ToolError::Failed( + "sandbox: network access is disabled, so URLs cannot be fetched".into(), + )); + } + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .map_err(|e| ToolError::Failed(format!("http client: {e}")))?; + let response = client + .get(url) + .send() + .await + .map_err(|e| ToolError::Failed(format!("GET {url}: {e}")))?; + let status = response.status(); + + use futures::StreamExt; + let mut stream = response.bytes_stream(); + let mut body = Vec::new(); + let mut clipped = false; + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|e| ToolError::Failed(format!("GET {url}: {e}")))?; + body.extend_from_slice(&chunk); + if body.len() > MAX_FETCH_BYTES { + body.truncate(MAX_FETCH_BYTES); + clipped = true; + break; + } + } + let text = String::from_utf8_lossy(&body); + if !status.is_success() { + return Err(ToolError::Failed(format!( + "GET {url} returned {status}\n{}", + truncate_middle(text.into_owned(), 2_000) + ))); + } + Ok(truncate_middle( + format!( + "GET {url} → {status}{}\n\n{text}", + if clipped { " (body clipped)" } else { "" } + ), + MAX_READ_BYTES, + )) +} + +/// A directory as a sorted listing: directories first, sizes for files. +async fn list_dir(path: &std::path::Path) -> Result { + let mut reader = tokio::fs::read_dir(path) + .await + .map_err(|e| ToolError::Failed(format!("cannot read {}: {e}", path.display())))?; + let mut entries = Vec::new(); + while let Some(entry) = reader + .next_entry() + .await + .map_err(|e| ToolError::Failed(format!("cannot read {}: {e}", path.display())))? + { + let name = entry.file_name().to_string_lossy().into_owned(); + let meta = entry.metadata().await; + let is_dir = meta.as_ref().is_ok_and(|m| m.is_dir()); + let size = meta.map(|m| m.len()).unwrap_or(0); + entries.push((!is_dir, name, size)); + } + if entries.is_empty() { + return Ok(format!("(empty directory {})", path.display())); + } + entries.sort(); + let mut out = format!( + "directory {} ({} entries):\n", + path.display(), + entries.len() + ); + for (is_file, name, size) in entries { + if is_file { + out.push_str(&format!(" {name} {size} bytes\n")); + } else { + out.push_str(&format!(" {name}/\n")); + } + } + Ok(truncate_middle(out, MAX_READ_BYTES)) +} + pub struct WriteFile; #[async_trait::async_trait] @@ -123,6 +250,173 @@ impl Tool for WriteFile { pub struct EditFile; +/// A parsed anchor: 1-indexed line plus its content hash. Line 0 is the +/// virtual top-of-file anchor, valid only for `insert_after`. +fn parse_anchor(raw: &str) -> Result<(usize, String), ToolError> { + if raw == "0" { + return Ok((0, String::new())); + } + let invalid = || { + ToolError::InvalidInput(format!( + "bad anchor `{raw}` — use the `line#hash` token from read_file (or \"0\" \ + with insert_after for the top of the file)" + )) + }; + let (line, hash) = raw.split_once('#').ok_or_else(invalid)?; + let line: usize = line.parse().map_err(|_| invalid())?; + if line == 0 || hash.len() != 4 || !hash.chars().all(|c| c.is_ascii_hexdigit()) { + return Err(invalid()); + } + Ok((line, hash.to_ascii_lowercase())) +} + +/// Resolve an anchor to a 0-based index. The hash decides, the line number +/// only locates: a line that *moved* (hash found on exactly one other line) +/// is followed there; a line that *changed* (hash on zero or several lines) +/// fails with fresh context to re-anchor from, never a misapplied edit. +fn resolve_anchor(lines: &[&str], line: usize, hash: &str) -> Result<(usize, bool), ToolError> { + if line >= 1 && line <= lines.len() && line_hash(lines[line - 1]) == hash { + return Ok((line - 1, false)); + } + let matches: Vec = lines + .iter() + .enumerate() + .filter(|(_, l)| line_hash(l) == hash) + .map(|(i, _)| i) + .collect(); + match matches.as_slice() { + [only] => Ok((*only, true)), + _ => { + let center = line.clamp(1, lines.len().max(1)); + let first = center.saturating_sub(3).max(1); + Err(ToolError::Failed(format!( + "stale anchor {line}#{hash}: {} — re-anchor from the current \ + content:\n{}", + if matches.is_empty() { + "that line's content is no longer in the file" + } else { + "that content now appears on several lines" + }, + hashlines(lines, first, 7), + ))) + } + } +} + +/// One resolved hunk as a splice: replace `start..end` with `content`. +struct Splice { + start: usize, + end: usize, + content: Vec, +} + +fn parse_hunks(lines: &[&str], hunks: &[Value]) -> Result<(Vec, usize), ToolError> { + let mut splices = Vec::new(); + let mut recovered = 0; + for hunk in hunks { + let op = hunk.get("op").and_then(Value::as_str).unwrap_or("replace"); + let anchor = hunk + .get("anchor") + .and_then(Value::as_str) + .ok_or_else(|| ToolError::InvalidInput("each hunk needs an `anchor`".into()))?; + let (line, hash) = parse_anchor(anchor)?; + let content: Option> = hunk + .get("content") + .and_then(Value::as_str) + .map(|s| s.split('\n').map(str::to_string).collect()); + let to = hunk.get("to").and_then(Value::as_str); + + let mut resolve = |line, hash: &str| { + resolve_anchor(lines, line, hash).inspect(|(_, moved)| recovered += *moved as usize) + }; + match op { + "insert_after" => { + if to.is_some() { + return Err(ToolError::InvalidInput( + "insert_after takes no `to` — it inserts at one point".into(), + )); + } + let content = content.ok_or_else(|| { + ToolError::InvalidInput("insert_after needs `content`".into()) + })?; + let at = if line == 0 { + 0 + } else { + resolve(line, &hash)?.0 + 1 + }; + splices.push(Splice { + start: at, + end: at, + content, + }); + } + "replace" | "delete" => { + if line == 0 { + return Err(ToolError::InvalidInput( + "anchor \"0\" is only for insert_after".into(), + )); + } + let content = match (op, content) { + ("replace", Some(content)) => content, + ("replace", None) => { + return Err(ToolError::InvalidInput("replace needs `content`".into())); + } + ("delete", None) => Vec::new(), + ("delete", Some(_)) => { + return Err(ToolError::InvalidInput( + "delete takes no `content` — use replace to substitute".into(), + )); + } + _ => unreachable!(), + }; + let (start, _) = resolve(line, &hash)?; + let end = match to { + None => start + 1, + Some(to) => { + let (line, hash) = parse_anchor(to)?; + if line == 0 { + return Err(ToolError::InvalidInput( + "`to` must be a real line anchor".into(), + )); + } + let (end, _) = resolve(line, &hash)?; + if end < start { + return Err(ToolError::InvalidInput(format!( + "`to` anchor resolves above `anchor` ({} < {})", + end + 1, + start + 1 + ))); + } + end + 1 + } + }; + splices.push(Splice { + start, + end, + content, + }); + } + other => { + return Err(ToolError::InvalidInput(format!( + "unknown op `{other}` (expected replace, insert_after, or delete)" + ))); + } + } + } + + // Hunks must not touch — including two insertions at one point, whose + // order the input cannot express. + splices.sort_by_key(|s| (s.start, s.end)); + for pair in splices.windows(2) { + if pair[1].start < pair[0].end || pair[1].start == pair[0].start { + return Err(ToolError::InvalidInput( + "hunks overlap — merge them or patch in two calls".into(), + )); + } + } + Ok((splices, recovered)) +} + #[async_trait::async_trait] impl Tool for EditFile { fn name(&self) -> &'static str { @@ -132,53 +426,112 @@ impl Tool for EditFile { fn spec(&self) -> ToolSpec { ToolSpec { name: "edit_file".into(), - description: "Replace exactly one occurrence of `old_string` with \ - `new_string`. Fails if the string is absent or ambiguous." + description: "Edit a file, two ways. Exact string: replace exactly one \ + occurrence of `old_string` with `new_string` (fails if \ + absent or ambiguous). Hashline patch: `patch` is an array \ + of hunks addressed by the `line#hash` anchors read_file \ + shows — {anchor, op: replace|insert_after|delete, to?, \ + content?}. `to` extends replace/delete to a span; anchor \ + \"0\" with insert_after prepends at the top; multi-line \ + `content` uses newlines. A moved anchor is followed while \ + its content is unique; a changed one fails with fresh \ + context to re-anchor from. Use exactly one mode per call." .into(), input_schema: json!({ "type": "object", "properties": { "path": {"type": "string"}, "old_string": {"type": "string"}, - "new_string": {"type": "string"} + "new_string": {"type": "string"}, + "patch": { + "type": "array", + "items": { + "type": "object", + "properties": { + "anchor": {"type": "string", "description": "`line#hash` token from read_file"}, + "op": {"type": "string", "enum": ["replace", "insert_after", "delete"]}, + "to": {"type": "string", "description": "End anchor for a replace/delete span (inclusive)"}, + "content": {"type": "string", "description": "Replacement or inserted lines"} + }, + "required": ["anchor", "op"] + } + } }, - "required": ["path", "old_string", "new_string"] + "required": ["path"] }), } } async fn run(&self, ctx: &ToolCtx, _call_id: &str, input: Value) -> Result { let path = resolve_path(ctx, required_str(&input, "path")?); - let old = required_str(&input, "old_string")?; - let new = required_str(&input, "new_string")?; - if old.is_empty() { - return Err(ToolError::InvalidInput( - "old_string must not be empty".into(), - )); - } ctx.check_write(&path)?; - let text = tokio::fs::read_to_string(&path) .await .map_err(|e| ToolError::Failed(format!("cannot read {}: {e}", path.display())))?; - match text.matches(old).count() { - 0 => Err(ToolError::Failed(format!( - "old_string not found in {}", - path.display() - ))), - 1 => { - let updated = text.replacen(old, new, 1); - tokio::fs::write(&path, updated).await.map_err(|e| { - ToolError::Failed(format!("cannot write {}: {e}", path.display())) + let updated = match (input.get("old_string"), input.get("patch")) { + (Some(_), Some(_)) => { + return Err(ToolError::InvalidInput( + "use either old_string/new_string or `patch`, not both".into(), + )); + } + (Some(_), None) => { + let old = required_str(&input, "old_string")?; + let new = required_str(&input, "new_string")?; + if old.is_empty() { + return Err(ToolError::InvalidInput( + "old_string must not be empty".into(), + )); + } + match text.matches(old).count() { + 0 => { + return Err(ToolError::Failed(format!( + "old_string not found in {}", + path.display() + ))); + } + 1 => (text.replacen(old, new, 1), String::new()), + n => { + return Err(ToolError::Failed(format!( + "old_string appears {n} times in {}; provide more \ + context to disambiguate", + path.display() + ))); + } + } + } + (None, Some(patch)) => { + let hunks = patch.as_array().filter(|h| !h.is_empty()).ok_or_else(|| { + ToolError::InvalidInput("`patch` must be a non-empty array of hunks".into()) })?; - Ok(format!("edited {}", path.display())) + let lines: Vec<&str> = text.lines().collect(); + let (splices, recovered) = parse_hunks(&lines, hunks)?; + let mut out: Vec = lines.iter().map(|s| s.to_string()).collect(); + for splice in splices.iter().rev() { + out.splice(splice.start..splice.end, splice.content.iter().cloned()); + } + let mut new_text = out.join("\n"); + if text.ends_with('\n') && !new_text.is_empty() { + new_text.push('\n'); + } + let note = match recovered { + 0 => String::new(), + n => format!(" ({n} moved anchor(s) followed by content hash)"), + }; + (new_text, format!(", {} hunk(s){note}", splices.len())) } - n => Err(ToolError::Failed(format!( - "old_string appears {n} times in {}; provide more context to disambiguate", - path.display() - ))), - } + (None, None) => { + return Err(ToolError::InvalidInput( + "provide old_string/new_string or a `patch`".into(), + )); + } + }; + + let (new_text, detail) = updated; + tokio::fs::write(&path, new_text) + .await + .map_err(|e| ToolError::Failed(format!("cannot write {}: {e}", path.display())))?; + Ok(format!("edited {}{detail}", path.display())) } } @@ -206,7 +559,15 @@ mod tests { .run(&c, "t", json!({"path": "sub/f.txt"})) .await .unwrap(); - assert_eq!(out, "1\tone\n2\ttwo\n3\tthree\n"); + assert_eq!( + out, + format!( + "1#{}\tone\n2#{}\ttwo\n3#{}\tthree\n", + line_hash("one"), + line_hash("two"), + line_hash("three") + ) + ); } #[tokio::test] @@ -221,7 +582,189 @@ mod tests { .run(&c, "t", json!({"path": "f.txt", "offset": 2, "limit": 2})) .await .unwrap(); - assert_eq!(out, "2\tb\n3\tc\n"); + assert_eq!( + out, + format!("2#{}\tb\n3#{}\tc\n", line_hash("b"), line_hash("c")) + ); + } + + /// The anchor for 1-indexed `line` of `content`, as read_file shows it. + fn anchor(content: &str, line: usize) -> String { + let text = content.lines().nth(line - 1).unwrap(); + format!("{line}#{}", line_hash(text)) + } + + async fn file_with(c: &ToolCtx, content: &str) -> String { + WriteFile + .run(c, "t", json!({"path": "f.txt", "content": content})) + .await + .unwrap(); + content.to_string() + } + + async fn read_back(c: &ToolCtx) -> String { + tokio::fs::read_to_string(c.workspace.join("f.txt")) + .await + .unwrap() + } + + #[tokio::test] + async fn patch_applies_replace_insert_and_delete_in_one_call() { + let dir = tempfile::tempdir().unwrap(); + let c = ctx(&dir); + let text = file_with(&c, "one\ntwo\nthree\nfour\n").await; + + let out = EditFile + .run( + &c, + "t", + json!({"path": "f.txt", "patch": [ + {"anchor": anchor(&text, 2), "op": "replace", "content": "TWO\nTWO-B"}, + {"anchor": anchor(&text, 4), "op": "delete"}, + {"anchor": "0", "op": "insert_after", "content": "zero"}, + ]}), + ) + .await + .unwrap(); + assert!(out.contains("3 hunk(s)"), "{out}"); + assert_eq!(read_back(&c).await, "zero\none\nTWO\nTWO-B\nthree\n"); + } + + #[tokio::test] + async fn patch_replaces_a_span_and_preserves_no_trailing_newline() { + let dir = tempfile::tempdir().unwrap(); + let c = ctx(&dir); + let text = file_with(&c, "a\nb\nc\nd").await; + + EditFile + .run( + &c, + "t", + json!({"path": "f.txt", "patch": [ + {"anchor": anchor(&text, 2), "to": anchor(&text, 3), "op": "replace", "content": "BC"}, + ]}), + ) + .await + .unwrap(); + assert_eq!(read_back(&c).await, "a\nBC\nd"); + } + + #[tokio::test] + async fn a_moved_anchor_is_followed_by_its_hash() { + let dir = tempfile::tempdir().unwrap(); + let c = ctx(&dir); + // Anchors taken from this layout... + let stale = "target\nfiller\n"; + // ...but lines were inserted above before the patch lands. + file_with(&c, "new-top\nalso-new\ntarget\nfiller\n").await; + + let out = EditFile + .run( + &c, + "t", + json!({"path": "f.txt", "patch": [ + {"anchor": anchor(stale, 1), "op": "replace", "content": "hit"}, + ]}), + ) + .await + .unwrap(); + assert!(out.contains("1 moved anchor"), "{out}"); + assert_eq!(read_back(&c).await, "new-top\nalso-new\nhit\nfiller\n"); + } + + #[tokio::test] + async fn a_changed_anchor_fails_with_fresh_context_not_a_wrong_edit() { + let dir = tempfile::tempdir().unwrap(); + let c = ctx(&dir); + let stale = "original\nkeep\n"; + file_with(&c, "rewritten\nkeep\n").await; + + let err = EditFile + .run( + &c, + "t", + json!({"path": "f.txt", "patch": [ + {"anchor": anchor(stale, 1), "op": "replace", "content": "x"}, + ]}), + ) + .await + .unwrap_err(); + let msg = err.to_string(); + assert!(msg.contains("stale anchor"), "{msg}"); + // The error carries current hashlines so the model can re-anchor. + assert!( + msg.contains(&format!("1#{}\trewritten", line_hash("rewritten"))), + "{msg}" + ); + assert_eq!(read_back(&c).await, "rewritten\nkeep\n"); + } + + #[tokio::test] + async fn ambiguous_recovery_fails_rather_than_guessing() { + let dir = tempfile::tempdir().unwrap(); + let c = ctx(&dir); + let stale = "dup\nunique\n"; + // The anchored content now appears twice, in neither original spot. + file_with(&c, "moved-a\ndup\ndup\n").await; + + let err = EditFile + .run( + &c, + "t", + json!({"path": "f.txt", "patch": [ + {"anchor": anchor(stale, 1), "op": "delete"}, + ]}), + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("several lines"), "{err}"); + } + + #[tokio::test] + async fn malformed_patches_are_invalid_input() { + let dir = tempfile::tempdir().unwrap(); + let c = ctx(&dir); + let text = file_with(&c, "a\nb\nc\n").await; + let a1 = anchor(&text, 1); + let a2 = anchor(&text, 2); + + for patch in [ + json!([]), + json!([{"op": "replace", "content": "x"}]), + json!([{"anchor": "nope", "op": "replace", "content": "x"}]), + json!([{"anchor": a1, "op": "explode"}]), + json!([{"anchor": a1, "op": "replace"}]), + json!([{"anchor": a1, "op": "delete", "content": "x"}]), + json!([{"anchor": a1, "op": "insert_after"}]), + json!([{"anchor": "0", "op": "delete"}]), + json!([{"anchor": a2, "to": a1, "op": "delete"}]), + // Overlapping hunks, and two insertions at one point. + json!([ + {"anchor": a1, "to": a2, "op": "delete"}, + {"anchor": a2, "op": "replace", "content": "x"}, + ]), + json!([ + {"anchor": a1, "op": "insert_after", "content": "x"}, + {"anchor": a1, "op": "insert_after", "content": "y"}, + ]), + ] { + let err = EditFile + .run(&c, "t", json!({"path": "f.txt", "patch": patch})) + .await + .unwrap_err(); + assert!(matches!(err, ToolError::InvalidInput(_)), "{patch}: {err}"); + } + + // Both modes at once is also a caller error. + let err = EditFile + .run( + &c, + "t", + json!({"path": "f.txt", "old_string": "a", "new_string": "b", "patch": [{"anchor": a1, "op": "delete"}]}), + ) + .await + .unwrap_err(); + assert!(matches!(err, ToolError::InvalidInput(_))); } #[tokio::test] @@ -257,6 +800,84 @@ mod tests { assert!(out.contains("x = 2")); } + #[tokio::test] + async fn a_directory_reads_as_a_sorted_listing() { + let dir = tempfile::tempdir().unwrap(); + let c = ctx(&dir); + tokio::fs::create_dir(dir.path().join("sub")).await.unwrap(); + tokio::fs::write(dir.path().join("b.txt"), "12345") + .await + .unwrap(); + tokio::fs::write(dir.path().join("a.txt"), "1") + .await + .unwrap(); + + let out = ReadFile.run(&c, "t", json!({"path": "."})).await.unwrap(); + let lines: Vec<&str> = out.lines().collect(); + assert!(lines[0].starts_with("directory"), "{out}"); + assert!(lines[0].contains("3 entries"), "{out}"); + // Directories first, then files alphabetically, with sizes. + assert_eq!( + &lines[1..], + [" sub/", " a.txt 1 bytes", " b.txt 5 bytes"] + ); + } + + /// One canned HTTP exchange on a local port; returns the URL to hit. + fn serve_once(response: &'static str) -> String { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + if let Ok((mut sock, _)) = listener.accept() { + use std::io::{Read, Write}; + let mut buf = [0u8; 4096]; + let _ = sock.read(&mut buf); + let _ = sock.write_all(response.as_bytes()); + } + }); + format!("http://{addr}/") + } + + #[tokio::test] + async fn urls_fetch_through_the_same_path() { + let url = + serve_once("HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello"); + let dir = tempfile::tempdir().unwrap(); + let out = ReadFile + .run(&ctx(&dir), "t", json!({"path": url})) + .await + .unwrap(); + assert!(out.contains("200"), "{out}"); + assert!(out.contains("hello"), "{out}"); + } + + #[tokio::test] + async fn a_failing_url_is_an_error_carrying_the_status() { + let url = serve_once( + "HTTP/1.1 404 Not Found\r\nContent-Length: 4\r\nConnection: close\r\n\r\ngone", + ); + let dir = tempfile::tempdir().unwrap(); + let err = ReadFile + .run(&ctx(&dir), "t", json!({"path": url})) + .await + .unwrap_err(); + let msg = err.to_string(); + assert!(msg.contains("404"), "{msg}"); + assert!(msg.contains("gone"), "{msg}"); + } + + #[tokio::test] + async fn a_network_denying_sandbox_refuses_urls() { + let dir = tempfile::tempdir().unwrap(); + let sandbox = std::sync::Arc::new(bullpen_sandbox::Sandbox::strict(dir.path())); + let c = ToolCtx::new(dir.path()).with_sandbox(sandbox); + let err = ReadFile + .run(&c, "t", json!({"path": "http://127.0.0.1:1/never"})) + .await + .unwrap_err(); + assert!(err.to_string().contains("network"), "{err}"); + } + #[tokio::test] async fn sandbox_blocks_write_outside_workspace() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/tools/src/lib.rs b/crates/tools/src/lib.rs index 4727d16..ad01f62 100644 --- a/crates/tools/src/lib.rs +++ b/crates/tools/src/lib.rs @@ -4,10 +4,12 @@ //! application, never by the agent loop. The runtime — not the model — owns //! the parallel-safety decision via [`Tool::parallel_safe`]. +mod ask; mod bash; mod fs; mod search; +pub use ask::{Ask, Asker}; pub use bash::Bash; pub use fs::{EditFile, ReadFile, WriteFile}; pub use search::{Glob, Grep}; diff --git a/docs/TOOLS.md b/docs/TOOLS.md new file mode 100644 index 0000000..9356ff1 --- /dev/null +++ b/docs/TOOLS.md @@ -0,0 +1,92 @@ +# Tools: what exists, what's next + +bullpen grows its tool surface the same way it grows everything else: +durability first. A tool earns its place when it can honor the contract in +`crates/tools` — the runtime owns parallel-safety and replay-safety, effects +are idempotent on the provider-assigned `call_id`, and a crash mid-call +leaves state the next process can recover. + +This document maps a best-in-class agent tool surface onto that contract: +what bullpen ships today, and in what order the rest should land. + +## Shipped + +| Tool | Catalog role | Notes | +|---|---|---| +| `bash` | runtime shell | Serial, sandboxable (Seatbelt on macOS), timeout-bounded. | +| `read_file` | read | One path for files, directories, and http(s) URLs. Files are hashline output — every line carries a `line#hash` anchor — capped head+tail; directories render sorted listings; URLs fetch with a streamed cap and honor the sandbox's network policy. | +| `write_file` / `edit_file` | write / edit | Sandbox write-confinement applies. `edit_file` takes exact-string replacements or hashline patches — hunks addressed by anchors, spans via `to`, with stale-anchor recovery: a moved line is followed while its content hash is unique, a changed line fails with fresh context instead of misapplying. | +| `grep` | content search | Regex over the tree, `.gitignore`-aware. | +| `glob` | path find | Pattern lookup; reach for `grep` when you need content. | +| `agent` | task fan-out | The pen: durable child sessions, deterministic ids, reattach-on-replay, durable child budget. `worktree: true` isolates a work child in its own git worktree on its own branch; `background: true` dispatches and returns immediately. Inspect, isolated, and background children all run in parallel. | +| `job` | background coordination | The coordination plane, exposed to the model: `list` children with store-derived state (status + pid liveness), `wait` polls to a terminal state and returns the child's answer, `cancel` signals a background child — which finishes as failed and stays resumable. | +| `todo` | session plan | Durable todo list in the store; replay-safe via deterministic item ids; the store enforces one item in progress at a time. | +| `ask` | follow-up questions | Transport-agnostic: interactive runs answer from the terminal, detached runs (background, `--json`, piped) fail the call with the reason instead of blocking forever. Numbered options resolve to their text. | + +Two catalog entries cost nothing because they already exist under other +names: `find` is `glob`, `search` is `grep`, and `task` is the pen's +`agent` tool. + +## Coordination leftovers + +- **cross-process `job`** — today `cancel` reaches only background + children dispatched by the calling process (in-process cooperative + cancellation, so the child records its own terminal state). Signalling a + child owned by another process is a later, deliberate step: it needs a + protocol for the *other* process to finish its session cleanly. + +## Then: files & search, deepened + +- **`ast_grep` / `ast_edit`** — structural queries and previewed rewrites + by shelling out to [ast-grep]. Preview-then-apply maps onto intent + records: the preview is durable, the apply is a separate confirmed step + (the catalog's `resolve`). +- **richer `read`** — directories and URLs are in; archives, SQLite, and + PDFs remain, each an incremental, independently testable decoder behind + the existing tool. + +## Then: reaching outside the workspace + +Each of these wraps a proven external surface; the work is inputs, +sandbox policy, and output discipline, not invention. + +- **`github`** — `gh` CLI operations (repo, PR, issues, run-watch). +- **`web_search`** — provider-backed search (page retrieval already ships + as URL reads). `--sandbox-strict` (network cut) must disable it cleanly, + as it already does for URL reads. +- **`ssh`** — one remote command against a configured host; never + implicit, always named host allowlists. + +## Later: each a project of its own + +Worth doing only when the layers above are solid, and each behind its own +design doc: + +- **`lsp` / `debug`** — language-server navigation and DAP sessions. + Long-lived server processes need ownership like `SessionWorker` gives + runs: exclusive, generation-stamped, crash-detectable. +- **`eval`** — persistent Python/JavaScript cells. Kernel state is + process state — exactly what bullpen promises survives — so cells must + journal their inputs to be replayable into a fresh kernel. +- **`browser`** — CDP-driven tabs; the largest sandbox-policy surface. +- **memory & context** — `checkpoint` / `rewind` (transcript compaction is + already an anticipated entry kind in the store's tree design) and a + `retain` / `recall` memory bank. +- **media** — image inspection and generation, diagram rendering, TTS. + +## The bar for a new tool + +Before a tool merges it must answer, in code: + +1. **What does a crash mid-call leave behind?** If the answer is "state + the next process cannot interpret", it is not done. +2. **Is `replay_safe` honest?** Only `true` when re-execution with the + same input and `call_id` converges — deterministic derived ids are the + house pattern (the pen's child sessions, `todo`'s item ids). +3. **Is `parallel_safe` decided by the runtime?** Per-invocation, from the + input, never from model self-declaration. +4. **Does the sandbox still mean something?** Write confinement and + `--sandbox-strict` network cuts apply to the new capability or the tool + explains, loudly, why not. + +[ast-grep]: https://ast-grep.github.io