From 24996d7fbcb4879db4009b4920114d59dac6a4d5 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 04:56:21 +0000 Subject: [PATCH 1/6] feat(todo): give sessions a durable plan The `todo` tool tracks a session's plan in the store (schema v8), not the model's context window, so it survives crashes, resumes, and future compaction like everything else. Two choices carry the design, both house patterns from the pen: - Deterministic item ids (uuidv5 of 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, in the same transaction. Items resolve by id prefix, mirroring session resolution. Every action returns the rendered plan, so the model always acts on current state. docs/TOOLS.md maps the rest of a best-in-class tool catalog onto the durability contract and sequences what comes next (job, pen worktree isolation, ask, hashline edits, ast-grep, ...). Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01KoSjfcJPjXpkVLTS2Vxwwr --- crates/cli/src/main.rs | 5 + crates/harness/src/lib.rs | 2 + crates/harness/src/todo.rs | 325 +++++++++++++++++++++++++++++++++++++ crates/store/src/lib.rs | 249 +++++++++++++++++++++++++++- docs/TOOLS.md | 103 ++++++++++++ 5 files changed, 682 insertions(+), 2 deletions(-) create mode 100644 crates/harness/src/todo.rs create mode 100644 docs/TOOLS.md diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 82cc150..b62ad4b 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -592,6 +592,11 @@ async fn run( &session.id, pen_config, ))); + // 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, + ))); let mut agent = Agent::new(provider, registry, tool_ctx, config) .with_transcript(transcript, usage) .with_events(tx) diff --git a/crates/harness/src/lib.rs b/crates/harness/src/lib.rs index bba9c27..13f7d42 100644 --- a/crates/harness/src/lib.rs +++ b/crates/harness/src/lib.rs @@ -11,8 +11,10 @@ pub mod pen; #[cfg(test)] pub(crate) mod testutil; +pub mod todo; 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/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/store/src/lib.rs b/crates/store/src/lib.rs index 4fd98b0..b025aeb 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(()) } @@ -719,6 +744,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 +1252,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 +1281,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/docs/TOOLS.md b/docs/TOOLS.md new file mode 100644 index 0000000..78c646e --- /dev/null +++ b/docs/TOOLS.md @@ -0,0 +1,103 @@ +# 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 | Workspace-relative, output capped head+tail. | +| `write_file` / `edit_file` | write / edit | Sandbox write-confinement applies. | +| `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. Inspect children run in parallel. | +| `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. | + +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. + +## Next: coordination (the durable-multi-agent story) + +These extend machinery bullpen already has, and they are the tools that +make the thesis visible. + +- **`job`** — wait on or cancel background work. `bullpen run --bg` already + detaches sessions coordinated through the store; `job` exposes that plane + *to the model*: list a session's background children, block on one + finishing, cancel one. The store is the source of truth (status + pid + liveness), so a `job` call after a crash sees reality, not a stale + in-process handle. +- **pen worktree isolation** — `agent` gains the CLI's `--worktree` + behavior: a work-mode child in its own git worktree on a `bullpen/` + branch, making work children parallel-safe too. +- **`ask`** — structured follow-up questions for interactive runs. Needs an + interactivity channel in the CLI run loop; in non-interactive runs the + tool reports that no one is listening rather than blocking forever. + +## Then: files & search, deepened + +- **hashline `edit`** — content-hash anchors with stale-anchor recovery, + replacing brittle exact-string matching. Pairs naturally with the + durability model: an anchor is an intent that can be revalidated on + replay. +- **`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`** — one path for directories, archives, SQLite, PDFs, + and URLs. Each format is 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` / `fetch`** — provider-backed search plus page retrieval. + `--sandbox-strict` (network cut) must disable them cleanly. +- **`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 From a5dc841ab3f041034aac4f54bbcda0ab3fb6244e Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 05:09:07 +0000 Subject: [PATCH 2/6] feat(pen): isolated worktrees, background dispatch, and a `job` tool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pen's `agent` tool gains the two flags that complete the durable multi-agent story, plus the coordination plane to drive them: - `worktree: true` runs a work child in its own git worktree on its own `bullpen/` branch — the CLI's `--bg --worktree` behavior, now available per child. Placement mirrors the CLI resume path (record before create; locate decides on replay), so a replayed spawn reattaches to the same tree. Isolated children get a sandbox rebased onto their worktree with the linked-worktree git dirs widened in. Worktrees follow the store's directory, so isolated stores (tests, $BULLPEN_HOME) keep their worktrees beside their database. - `background: true` dispatches the child and returns immediately. The child runs in this process but coordinates through the store like everything else, so crashes leave the same recoverable state as any session. Cancellation is cooperative — a oneshot into the child's select loop — so a cancelled child records its own terminal state. - The `job` tool exposes the coordination plane to the model: `list` derives each child's state from stored status plus pid liveness, `wait` polls the store to a terminal state and returns the child's recorded answer, `cancel` signals a background child, which finishes as failed and stays resumable. Inspect, isolated, and background children are all parallel-safe; only a work child in the shared checkout stays serial. Plumbing: `pid_alive` moves from the CLI into `store::status` and the worktree module moves from the CLI into the harness, so both sides of the coordination plane share one definition of liveness and placement. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01KoSjfcJPjXpkVLTS2Vxwwr --- Cargo.lock | 2 + crates/cli/src/bg.rs | 12 +- crates/cli/src/main.rs | 13 +- crates/harness/Cargo.toml | 1 + crates/harness/src/job.rs | 384 +++++++++++++++ crates/harness/src/lib.rs | 3 + crates/harness/src/pen.rs | 612 ++++++++++++++++++++---- crates/harness/src/testutil.rs | 14 + crates/{cli => harness}/src/worktree.rs | 8 + crates/store/Cargo.toml | 1 + crates/store/src/lib.rs | 13 + crates/store/src/status.rs | 12 + docs/TOOLS.md | 22 +- 13 files changed, 973 insertions(+), 124 deletions(-) create mode 100644 crates/harness/src/job.rs rename crates/{cli => harness}/src/worktree.rs (98%) diff --git a/Cargo.lock b/Cargo.lock index ccc704f..ea7ad4d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -193,6 +193,7 @@ dependencies = [ name = "bullpen-harness" version = "0.1.0" dependencies = [ + "anyhow", "async-trait", "bullpen-agent", "bullpen-llm", @@ -233,6 +234,7 @@ version = "0.1.0" dependencies = [ "bullpen-llm", "fs2", + "libc", "rusqlite", "serde", "serde_json", 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 b62ad4b..339bf5a 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,11 @@ 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(), - &session.id, - pen_config, - ))); + 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(), 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 13f7d42..d1b829a 100644 --- a/crates/harness/src/lib.rs +++ b/crates/harness/src/lib.rs @@ -8,11 +8,14 @@ //! //! This crate is the future home of the pen (durable subagents). +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; 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/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 b025aeb..5c5e405 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -324,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( 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/docs/TOOLS.md b/docs/TOOLS.md index 78c646e..08d4fc3 100644 --- a/docs/TOOLS.md +++ b/docs/TOOLS.md @@ -18,30 +18,24 @@ what bullpen ships today, and in what order the rest should land. | `write_file` / `edit_file` | write / edit | Sandbox write-confinement applies. | | `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. Inspect children run in parallel. | +| `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. | 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. -## Next: coordination (the durable-multi-agent story) +## Next: coordination, completed -These extend machinery bullpen already has, and they are the tools that -make the thesis visible. - -- **`job`** — wait on or cancel background work. `bullpen run --bg` already - detaches sessions coordinated through the store; `job` exposes that plane - *to the model*: list a session's background children, block on one - finishing, cancel one. The store is the source of truth (status + pid - liveness), so a `job` call after a crash sees reality, not a stale - in-process handle. -- **pen worktree isolation** — `agent` gains the CLI's `--worktree` - behavior: a work-mode child in its own git worktree on a `bullpen/` - branch, making work children parallel-safe too. - **`ask`** — structured follow-up questions for interactive runs. Needs an interactivity channel in the CLI run loop; in non-interactive runs the tool reports that no one is listening rather than blocking forever. +- **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 From da402f33dd0e1269a2faf3d293e590758862f46d Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 05:12:23 +0000 Subject: [PATCH 3/6] feat(ask): follow-up questions for interactive runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `ask` tool puts one structured question to the human driving the run. It is transport-agnostic: the application injects an `Asker`, and the CLI's implementation prints to stderr — never into the answer stream on stdout — and reads one line from the terminal on a blocking thread. Options render as a numbered list, and the tool owns the number-to-option mapping so every transport gets it identically. A run with nobody on the other end (background, `--json`, piped stdin) registers the *detached* variant: the model still sees the tool, and a call fails immediately with the reason — decide and note the assumption — instead of blocking on input nobody will ever type, or surfacing a bare unknown-tool error. Also folds the new coordination surface into the README and marks the coordination tranche done in docs/TOOLS.md. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01KoSjfcJPjXpkVLTS2Vxwwr --- Cargo.lock | 1 + README.md | 15 ++- crates/cli/Cargo.toml | 1 + crates/cli/src/main.rs | 39 +++++++ crates/tools/src/ask.rs | 231 ++++++++++++++++++++++++++++++++++++++++ crates/tools/src/lib.rs | 2 + docs/TOOLS.md | 6 +- 7 files changed, 286 insertions(+), 9 deletions(-) create mode 100644 crates/tools/src/ask.rs diff --git a/Cargo.lock b/Cargo.lock index ea7ad4d..9277ebb 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", diff --git a/README.md b/README.md index 661e766..c254fac 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 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/main.rs b/crates/cli/src/main.rs index 339bf5a..ebf15ac 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -598,6 +598,14 @@ async fn run( Store::default_path(), &session.id, ))); + // 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) @@ -750,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/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/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 index 08d4fc3..8a869fc 100644 --- a/docs/TOOLS.md +++ b/docs/TOOLS.md @@ -21,16 +21,14 @@ what bullpen ships today, and in what order the rest should land. | `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. -## Next: coordination, completed +## Coordination leftovers -- **`ask`** — structured follow-up questions for interactive runs. Needs an - interactivity channel in the CLI run loop; in non-interactive runs the - tool reports that no one is listening rather than blocking forever. - **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 From a7fcc6d2903a5a6f80298558e1a0cd6df997d178 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 05:16:48 +0000 Subject: [PATCH 4/6] feat(fs): hashline reads and anchor-patched edits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `read_file` output becomes `line#hashcontent`: every line carries an anchor made of its position plus the first four hex chars of its SHA-256. `edit_file` keeps the exact-string mode and gains a `patch` mode addressed by those anchors — hunks of replace / insert_after / delete, spans via an inclusive `to`, anchor "0" to prepend at the top, several hunks per call applied bottom-up with overlaps rejected. An anchor is a claim about content, not just a position, which is what makes edits against a drifted file safe: - a *moved* line (hash found on exactly one line) is followed there and reported, instead of patching whatever now sits at the old number; - a *changed* line (hash on zero or several lines) fails the call with fresh hashline context around the site, so the model re-anchors without another read — and never misapplies the edit. Recovery trusts a hash only when it is unique in the file, so a four-hex-char collision degrades to an explicit error, never a wrong edit. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01KoSjfcJPjXpkVLTS2Vxwwr --- Cargo.lock | 1 + crates/tools/Cargo.toml | 1 + crates/tools/src/fs.rs | 518 +++++++++++++++++++++++++++++++++++++--- docs/TOOLS.md | 8 +- 4 files changed, 484 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9277ebb..38f3514 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -256,6 +256,7 @@ dependencies = [ "regex", "serde", "serde_json", + "sha2", "tempfile", "thiserror", "tokio", diff --git a/crates/tools/Cargo.toml b/crates/tools/Cargo.toml index c52ee42..f0bf22d 100644 --- a/crates/tools/Cargo.toml +++ b/crates/tools/Cargo.toml @@ -13,6 +13,7 @@ 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/fs.rs b/crates/tools/src/fs.rs index 37b22a5..f19df02 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] @@ -19,7 +48,9 @@ impl Tool for ReadFile { ToolSpec { name: "read_file".into(), description: "Read a file, optionally a line window. Output is 1-indexed \ - `linecontent` and capped at 256 KiB." + `line#hashcontent` and capped at 256 KiB; the \ + `line#hash` token is an anchor that `edit_file` patches \ + accept." .into(), input_schema: json!({ "type": "object", @@ -58,13 +89,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!( @@ -123,6 +149,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 +325,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 +458,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 +481,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] diff --git a/docs/TOOLS.md b/docs/TOOLS.md index 8a869fc..21f3ad9 100644 --- a/docs/TOOLS.md +++ b/docs/TOOLS.md @@ -14,8 +14,8 @@ what bullpen ships today, and in what order the rest should land. | Tool | Catalog role | Notes | |---|---|---| | `bash` | runtime shell | Serial, sandboxable (Seatbelt on macOS), timeout-bounded. | -| `read_file` | read | Workspace-relative, output capped head+tail. | -| `write_file` / `edit_file` | write / edit | Sandbox write-confinement applies. | +| `read_file` | read | Workspace-relative, output capped head+tail. Hashline output: every line carries a `line#hash` anchor. | +| `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. | @@ -37,10 +37,6 @@ names: `find` is `glob`, `search` is `grep`, and `task` is the pen's ## Then: files & search, deepened -- **hashline `edit`** — content-hash anchors with stale-anchor recovery, - replacing brittle exact-string matching. Pairs naturally with the - durability model: an anchor is an intent that can be revalidated on - replay. - **`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 From 8dc272e6f50c458fe07f31094c49cfc1c68061e7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 05:19:54 +0000 Subject: [PATCH 5/6] feat(fs): read directories and URLs through the one read path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `read_file` now reads whatever the path names. A directory renders a sorted listing — directories first, sizes for files — instead of a bare OS error. An http(s) URL is fetched with GET, the body bounded while it streams (2 MiB) rather than after it lands, with the status carried in the output and a non-success status carried in the error. The sandbox's network capability governs URL reads exactly as it governs shell commands: a sandbox that denies network refuses the fetch outright. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01KoSjfcJPjXpkVLTS2Vxwwr --- Cargo.lock | 2 + crates/tools/Cargo.toml | 2 + crates/tools/src/fs.rs | 195 ++++++++++++++++++++++++++++++++++++++-- docs/TOOLS.md | 13 +-- 4 files changed, 198 insertions(+), 14 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 38f3514..f9b3d1c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -251,9 +251,11 @@ dependencies = [ "async-trait", "bullpen-llm", "bullpen-sandbox", + "futures", "globset", "ignore", "regex", + "reqwest", "serde", "serde_json", "sha2", diff --git a/crates/tools/Cargo.toml b/crates/tools/Cargo.toml index f0bf22d..a40c0cb 100644 --- a/crates/tools/Cargo.toml +++ b/crates/tools/Cargo.toml @@ -8,7 +8,9 @@ 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 diff --git a/crates/tools/src/fs.rs b/crates/tools/src/fs.rs index f19df02..06db3e3 100644 --- a/crates/tools/src/fs.rs +++ b/crates/tools/src/fs.rs @@ -47,17 +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 \ - `line#hashcontent` and capped at 256 KiB; the \ - `line#hash` token is an anchor that `edit_file` patches \ - accept." + 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"] }), @@ -73,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())))?; @@ -102,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] @@ -699,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/docs/TOOLS.md b/docs/TOOLS.md index 21f3ad9..9356ff1 100644 --- a/docs/TOOLS.md +++ b/docs/TOOLS.md @@ -14,7 +14,7 @@ what bullpen ships today, and in what order the rest should land. | Tool | Catalog role | Notes | |---|---|---| | `bash` | runtime shell | Serial, sandboxable (Seatbelt on macOS), timeout-bounded. | -| `read_file` | read | Workspace-relative, output capped head+tail. Hashline output: every line carries a `line#hash` anchor. | +| `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. | @@ -41,9 +41,9 @@ names: `find` is `glob`, `search` is `grep`, and `task` is the pen's 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`** — one path for directories, archives, SQLite, PDFs, - and URLs. Each format is an incremental, independently testable decoder - behind the existing tool. +- **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 @@ -51,8 +51,9 @@ 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` / `fetch`** — provider-backed search plus page retrieval. - `--sandbox-strict` (network cut) must disable them cleanly. +- **`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. From c18f212ca1fee0fb6e8449292e904817edcf138a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 05:21:00 +0000 Subject: [PATCH 6/6] docs: truth pass over the grown tool surface ARCHITECTURE's crate table, the parallel-scheduling note, the harness module docs, and the README status row all predate the coordination and hashline work; bring each in line with what actually ships. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01KoSjfcJPjXpkVLTS2Vxwwr --- ARCHITECTURE.md | 6 +++--- README.md | 2 +- crates/harness/src/lib.rs | 4 +++- 3 files changed, 7 insertions(+), 5 deletions(-) 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/README.md b/README.md index c254fac..9de7d95 100644 --- a/README.md +++ b/README.md @@ -156,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/harness/src/lib.rs b/crates/harness/src/lib.rs index d1b829a..846856e 100644 --- a/crates/harness/src/lib.rs +++ b/crates/harness/src/lib.rs @@ -6,7 +6,9 @@ //! 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;