From 9afff347052bdad5746be36c4494526f86e8cf6c Mon Sep 17 00:00:00 2001 From: Yogthos Date: Fri, 29 May 2026 23:00:45 -0400 Subject: [PATCH] feat(bash): rework background shells to the Claude-Code model (unbounded + read/kill by id) Replaces the timeout-bounded, push-once background-shell design (#233) with Claude Code's proven model: background bash runs UNBOUNDED and the model reads its output and stops it explicitly by id. - New BackgroundShellStore (bg_shell.rs): per-shell live output buffer (unread, drained on read; hard-capped so a never-read flood can't OOM), status (Running/Exited/Killed/Failed), and the drain JoinHandle. Process-global (like the subagent /kill registry) so the bash tool, the new tools, the status bar, and session cleanup share one instance without threading it through every signature; tests inject their own store. - bash background=true: spawn_streaming_shell runs the command detached with no timeout (optional `timeout` = auto-kill-after-N), streaming stdout/stderr into the store; returns a shell id immediately. Permission + sandbox checks run before the spawn as before. PgKillGuard SIGKILLs the process group on abort. - New model-facing tools (mirroring Claude Code): `bash_output` (read new output + status by id) and `kill_shell` (kill by id). Registered in both tool paths, added to BUILTIN_TOOL_NAMES, exposed in the schema. - Reverted #233's TaskKind piggyback on BackgroundStore (push-once delivery doesn't fit a long-lived process): back to subagents-only + running_count(). - Status bar: agents:N from BackgroundStore.running_count(), shells:N from the shell store. /tasks now lists background shells with status. Session swap/end kills all background shells. Tests: store unit tests (drain/cap/kill/finish-first-wins/list), an end-to-end unbounded background bash (streams output, exits clean, count returns to 0), and status-bar per-kind badges. Full feature-matrix suite green at -D warnings (2150 passed). --- docs/features.md | 2 +- src/agent/agent_loop/integration.rs | 5 +- src/agent/builder.rs | 8 +- src/agent/tools/background.rs | 128 +++----- src/agent/tools/bash.rs | 245 +++++++++++---- src/agent/tools/bg_shell.rs | 458 ++++++++++++++++++++++++++++ src/agent/tools/mod.rs | 4 + src/agent/tools/task.rs | 9 +- src/agent/tools/task_status.rs | 35 +-- src/ui/mod.rs | 97 +++--- src/ui/slash/cmd_session.rs | 14 + src/ui/status.rs | 59 ++-- 12 files changed, 806 insertions(+), 258 deletions(-) create mode 100644 src/agent/tools/bg_shell.rs diff --git a/docs/features.md b/docs/features.md index a0fa3c8b..fd3f04d3 100644 --- a/docs/features.md +++ b/docs/features.md @@ -20,7 +20,7 @@ headline differentiators, see the top-level [README](../README.md). - **Prompts system**: switch between system prompt modes at runtime (`code`, `plan`, `review`, `debug`, etc.). See [prompts.md](prompts.md). - **Per-prompt tool restrictions**: each prompt (`prompts/.md`) can declare a `deny_tools` frontmatter list. The permission checker refuses those tools while the prompt is active — a real security boundary, not a prose gate. `plan`, `review`, and `review-security` ship with `edit`, `write`, `apply_patch`, `bash`, and `webfetch` denied. See [prompts.md](prompts.md). - **Subagent support**: `task` tool spawns a subagent for research or general analysis subtasks. -- **Background shells**: the `bash` tool accepts `background: true` to run a command detached — it returns immediately with a shell id and the output is delivered automatically when the command finishes (the same completion-notification channel background subagents use; `task_status` can poll a shell id too). Useful for long-running builds, servers, and watchers that shouldn't block the turn. Tracked in the shared background registry and capped per kind. The status bar shows live counts when any are running: `agents:N` (background subagents) and `shells:N` (background shells). +- **Background shells** (Claude-Code-style): the `bash` tool accepts `background: true` to run a command **detached and unbounded** (for dev servers, watch builds, long-running jobs) — it returns immediately with a shell id. The model reads accumulated output incrementally with the **`bash_output`** tool and stops the shell with **`kill_shell`** (both take the id); an optional `timeout` auto-kills after N seconds. Shells are tracked in a dedicated registry, capped at 8 concurrent, listed by `/tasks`, and killed when the session ends. The status bar shows live counts when any are running: `agents:N` (background subagents) and `shells:N` (background shells). - **Memory & self-improvement**: persistent per-project memory at `.dirge/memory/` (`MEMORY.md` facts + `PITFALLS.md`), injected into the system prompt as a frozen snapshot. The agent edits it via the `memory` tool. After an idle session, a unified post-session orchestrator runs (in order, fire-and-forget): a background review that extracts learnings into memory + skills, a skills curator and a memory curator (stale-detection + lifecycle + LLM consolidation, with audit reports under each store's `.curator_reports/`), and a cross-session pass that promotes sub-threshold patterns recurring across past sessions. - **MCP support**: connect MCP servers for extended tooling (optional compile-time feature). - **Git worktrees**: `/worktree` to create branch-per-task worktrees, `/wt-merge`, `/wt-exit`. diff --git a/src/agent/agent_loop/integration.rs b/src/agent/agent_loop/integration.rs index efcb8620..6423d068 100644 --- a/src/agent/agent_loop/integration.rs +++ b/src/agent/agent_loop/integration.rs @@ -1058,10 +1058,7 @@ mod tests { use crate::agent::tools::background::{BackgroundStore, TaskState}; let store = BackgroundStore::new(); - store.insert( - "sub-1".into(), - crate::agent::tools::background::TaskKind::Subagent, - ); + store.insert("sub-1".into()); let saw_reminder = Arc::new(Mutex::new(false)); let saw_clone = saw_reminder.clone(); diff --git a/src/agent/builder.rs b/src/agent/builder.rs index 95777f67..1ff0e7d2 100644 --- a/src/agent/builder.rs +++ b/src/agent/builder.rs @@ -376,8 +376,10 @@ pub async fn build_agent_inner( sandbox.clone(), cache.clone(), ) - .with_bg_store(bg_store.clone()), + .with_shell_store(Some(tools::bg_shell::global())), ), + Box::new(tools::BashOutputTool::new(tools::bg_shell::global())), + Box::new(tools::KillShellTool::new(tools::bg_shell::global())), Box::new(tools::GrepTool::with_cache( permission.clone(), ask_tx.clone(), @@ -759,11 +761,13 @@ pub async fn build_loop_tools( sandbox.clone(), cache.clone(), ) - .with_bg_store(bg_store.clone()), + .with_shell_store(Some(tools::bg_shell::global())), Some(ToolExecutionMode::Sequential), ) .await, ); + tools.push(wrap(tools::BashOutputTool::new(tools::bg_shell::global()), None).await); + tools.push(wrap(tools::KillShellTool::new(tools::bg_shell::global()), None).await); // Read-only batch. tools.push( diff --git a/src/agent/tools/background.rs b/src/agent/tools/background.rs index 7d20e10f..535079ce 100644 --- a/src/agent/tools/background.rs +++ b/src/agent/tools/background.rs @@ -25,16 +25,10 @@ const STORE_CAPACITY: usize = 32; /// tasks (audit M2). Without this cap a misbehaving LLM could spawn /// dozens of background tasks in parallel and burn the user's API /// budget. Hit by tracking the in-flight JoinHandle count via -/// `running_count_kind(Subagent)`; the `task` tool refuses new spawns +/// `running_count()`; the `task` tool refuses new background spawns /// when at-cap with a clear error rather than queueing. const MAX_CONCURRENT_SUBAGENTS: usize = 4; -/// Maximum number of concurrently running detached background shells -/// (the `bash` tool's `background` arg). Same rationale as the subagent -/// cap — a runaway model shouldn't be able to spawn unbounded detached -/// processes. Counted per-kind via `running_count_kind(TaskKind::Shell)`. -const MAX_CONCURRENT_BG_SHELLS: usize = 8; - /// Event surfaced on the UI lifecycle channel. /// /// `Started` fires when the parent spawns a background task; `Finished` fires @@ -92,21 +86,9 @@ pub enum TaskState { Failed(String), } -/// What kind of background work an entry represents. The store tracks -/// both background subagents (the `task` tool) and detached shells (the -/// `bash` tool's `background` arg) through the SAME delivery pipeline — -/// `kind` lets the status bar count them separately (`agents:N` / -/// `shells:N`) and lets each tool enforce its own concurrency cap. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum TaskKind { - Subagent, - Shell, -} - #[derive(Debug, Clone)] pub struct BackgroundTask { pub state: TaskState, - pub kind: TaskKind, } /// A completion event ready to be surfaced to the parent agent at its next @@ -137,9 +119,8 @@ impl BackgroundStore { /// Insert a new task in Running state. If the store is at capacity, the /// oldest task is evicted to make room. Inserting an existing id replaces - /// it in place without evicting anything. `kind` distinguishes a - /// background subagent from a detached shell (drives the per-kind counts). - pub fn insert(&self, id: String, kind: TaskKind) { + /// it in place without evicting anything. + pub fn insert(&self, id: String) { let mut inner = self.lock(); if !inner.tasks.contains_key(&id) && inner.tasks.len() >= STORE_CAPACITY { // Evict the oldest by insertion order. shift_remove preserves order @@ -150,7 +131,6 @@ impl BackgroundStore { id, BackgroundTask { state: TaskState::Running, - kind, }, ); } @@ -160,19 +140,12 @@ impl BackgroundStore { self.lock().tasks.get(id).cloned() } - /// Count of in-flight tasks of a specific `kind`. A task is "running" - /// while it still holds a live `JoinHandle`; `notify`/`cancel_all` - /// drop the handle, so an entry counts iff it is both Running and of - /// the requested kind. Drives the status bar's separate `agents:N` / - /// `shells:N` segments and each tool's per-kind concurrency cap. - pub fn running_count_kind(&self, kind: TaskKind) -> usize { - let inner = self.lock(); - inner - .handles - .keys() - .filter_map(|id| inner.tasks.get(id)) - .filter(|t| t.kind == kind) - .count() + /// Count of subagent tasks currently in flight. Equal to the number + /// of live `JoinHandle`s the store is tracking. Used by the `task` + /// tool to refuse a new background spawn at the `MAX_CONCURRENT_SUBAGENTS` + /// cap, and by the status bar's `agents:N` segment. + pub fn running_count(&self) -> usize { + self.lock().handles.len() } /// Compile-time cap on concurrent subagent spawns. @@ -180,11 +153,6 @@ impl BackgroundStore { MAX_CONCURRENT_SUBAGENTS } - /// Compile-time cap on concurrent detached background shells. - pub fn max_concurrent_shells() -> usize { - MAX_CONCURRENT_BG_SHELLS - } - /// Record a terminal state (Completed or Failed) and queue a notification /// for delivery. Truncates the payload to MAX_TASK_OUTPUT_CHARS. /// @@ -461,7 +429,7 @@ mod tests { #[test] fn insert_then_get_returns_running() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); let task = store.get("t1").expect("task present"); assert_eq!(task.state, TaskState::Running); } @@ -480,7 +448,7 @@ mod tests { #[tokio::test] async fn cancel_all_aborts_in_flight_tasks() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); // Spawn a long-running task and register its handle. let store_for_task = store.clone(); @@ -493,7 +461,7 @@ mod tests { // Also enqueue a stale pending notification to verify // cancel_all clears the queue. - store.insert("t_stale".into(), TaskKind::Subagent); + store.insert("t_stale".into()); store.notify("t_stale", TaskState::Completed("prev session".into())); assert_eq!(store.pending_len(), 1); @@ -523,7 +491,7 @@ mod tests { #[test] fn regression_get_is_read_only_after_completion() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Completed("done".into())); for _ in 0..3 { @@ -536,7 +504,7 @@ mod tests { #[test] fn notify_pushes_completed_to_pending_queue() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Completed("done".into())); assert_eq!(store.pending_len(), 1); } @@ -550,7 +518,7 @@ mod tests { #[test] fn regression_notify_relays_large_completed_payload() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); // Multi-line payload well past the 8 KiB byte threshold AND // the 200-line threshold so the relay fires AND its // head/tail summary elides the middle. @@ -577,7 +545,7 @@ mod tests { #[test] fn regression_notify_truncates_failed_error() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); let huge = "e".repeat(MAX_TASK_OUTPUT_CHARS * 2); let huge_len = huge.len(); store.notify("t1", TaskState::Failed(huge)); @@ -604,7 +572,7 @@ mod tests { #[test] fn small_completed_payload_passes_through_relay_verbatim() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); let small = "subagent answer: 42\nplus a couple more lines.\n".to_string(); store.notify("t1", TaskState::Completed(small.clone())); @@ -625,7 +593,7 @@ mod tests { #[test] fn notify_with_running_state_is_noop() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Running); assert_eq!(store.pending_len(), 0); // State unchanged. @@ -638,7 +606,7 @@ mod tests { #[test] fn regression_double_notify_enqueues_once() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Completed("first".into())); store.notify("t1", TaskState::Completed("second".into())); assert_eq!(store.pending_len(), 1); @@ -652,8 +620,8 @@ mod tests { #[test] fn drain_returns_pending_then_empties_queue() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); - store.insert("t2".into(), TaskKind::Subagent); + store.insert("t1".into()); + store.insert("t2".into()); store.notify("t1", TaskState::Completed("a".into())); store.notify("t2", TaskState::Failed("b".into())); @@ -673,7 +641,7 @@ mod tests { #[test] fn drain_is_empty_when_nothing_pending() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); // Insert alone doesn't enqueue — only notify() does. assert!(store.drain_notifications().is_empty()); } @@ -684,11 +652,11 @@ mod tests { fn regression_lru_evicts_oldest_at_capacity() { let store = BackgroundStore::new(); for i in 0..STORE_CAPACITY { - store.insert(format!("t{i}"), TaskKind::Subagent); + store.insert(format!("t{i}")); } assert_eq!(store.len(), STORE_CAPACITY); // One more push past capacity. - store.insert("overflow".into(), TaskKind::Subagent); + store.insert("overflow".into()); assert_eq!(store.len(), STORE_CAPACITY); // The oldest (t0) is gone; the newest is retained. assert!(store.get("t0").is_none()); @@ -702,10 +670,10 @@ mod tests { fn re_insert_existing_id_does_not_evict() { let store = BackgroundStore::new(); for i in 0..STORE_CAPACITY { - store.insert(format!("t{i}"), TaskKind::Subagent); + store.insert(format!("t{i}")); } // Re-insert at capacity: the existing id should just be reset. - store.insert("t5".into(), TaskKind::Subagent); + store.insert("t5".into()); assert_eq!(store.len(), STORE_CAPACITY); assert!(store.get("t0").is_some(), "oldest must NOT be evicted"); assert_eq!(store.get("t5").unwrap().state, TaskState::Running); @@ -717,9 +685,9 @@ mod tests { fn regression_notify_on_evicted_id_is_noop() { let store = BackgroundStore::new(); for i in 0..STORE_CAPACITY { - store.insert(format!("t{i}"), TaskKind::Subagent); + store.insert(format!("t{i}")); } - store.insert("overflow".into(), TaskKind::Subagent); // evicts t0 + store.insert("overflow".into()); // evicts t0 store.notify("t0", TaskState::Completed("late".into())); assert_eq!(store.pending_len(), 0); @@ -731,7 +699,7 @@ mod tests { fn clones_share_state() { let a = BackgroundStore::new(); let b = a.clone(); - a.insert("t1".into(), TaskKind::Subagent); + a.insert("t1".into()); b.notify("t1", TaskState::Completed("via b".into())); let drained = a.drain_notifications(); @@ -750,7 +718,7 @@ mod tests { #[test] fn prepend_passthrough_when_nothing_pending() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); // running, not pending + store.insert("t1".into()); // running, not pending let out = prepend_pending_notifications("hello", Some(&store)); assert_eq!(out, "hello"); } @@ -758,7 +726,7 @@ mod tests { #[test] fn prepend_formats_system_reminder() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Completed("the result".into())); let out = prepend_pending_notifications("user msg", Some(&store)); @@ -772,7 +740,7 @@ mod tests { #[test] fn prepend_includes_failed_tasks() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Failed("kaboom".into())); let out = prepend_pending_notifications("user msg", Some(&store)); @@ -786,7 +754,7 @@ mod tests { #[test] fn regression_prepend_drains_queue_once() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Completed("once".into())); let first = prepend_pending_notifications("msg", Some(&store)); @@ -800,7 +768,7 @@ mod tests { fn prepend_includes_all_pending_tasks_in_order() { let store = BackgroundStore::new(); for i in 0..3 { - store.insert(format!("t{i}"), TaskKind::Subagent); + store.insert(format!("t{i}")); store.notify(&format!("t{i}"), TaskState::Completed(format!("r{i}"))); } let out = prepend_pending_notifications("msg", Some(&store)); @@ -824,7 +792,7 @@ mod tests { async fn ui_sink_receives_completion_event() { let (tx, mut rx) = mpsc::unbounded_channel(); let store = BackgroundStore::with_ui_sink(tx); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Completed("done".into())); let notif = unwrap_finished(rx.recv().await.expect("event delivered")); @@ -836,7 +804,7 @@ mod tests { async fn ui_sink_receives_failure_event() { let (tx, mut rx) = mpsc::unbounded_channel(); let store = BackgroundStore::with_ui_sink(tx); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Failed("boom".into())); let notif = unwrap_finished(rx.recv().await.unwrap()); @@ -853,7 +821,7 @@ mod tests { async fn ui_sink_event_carries_relayed_payload() { let (tx, mut rx) = mpsc::unbounded_channel(); let store = BackgroundStore::with_ui_sink(tx); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); // Multi-line payload well over the 200-line threshold and // 8 KiB byte threshold so the relay's head/tail summary // elides the middle (single-line payloads still relay but @@ -889,7 +857,7 @@ mod tests { async fn ui_sink_does_not_receive_running_state() { let (tx, mut rx) = mpsc::unbounded_channel(); let store = BackgroundStore::with_ui_sink(tx); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Running); // Drain non-blockingly: nothing should be queued. @@ -911,12 +879,12 @@ mod tests { #[test] fn regression_drain_returns_snapshotted_state_after_eviction() { let store = BackgroundStore::new(); - store.insert("victim".into(), TaskKind::Subagent); + store.insert("victim".into()); store.notify("victim", TaskState::Completed("the result".into())); // Push enough new inserts to evict "victim" from the task map. for i in 0..STORE_CAPACITY { - store.insert(format!("filler{i}"), TaskKind::Subagent); + store.insert(format!("filler{i}")); } assert!(store.get("victim").is_none(), "victim must be evicted"); @@ -934,7 +902,7 @@ mod tests { async fn notify_started_fires_only_on_ui_sink() { let (tx, mut rx) = mpsc::unbounded_channel(); let store = BackgroundStore::with_ui_sink(tx); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify_started("t1"); let evt = rx.recv().await.expect("Started event delivered"); @@ -964,7 +932,7 @@ mod tests { async fn ui_sink_send_after_receiver_dropped_is_silent() { let (tx, rx) = mpsc::unbounded_channel(); let store = BackgroundStore::with_ui_sink(tx); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); drop(rx); // Must not panic. store.notify("t1", TaskState::Completed("payload".into())); @@ -985,7 +953,7 @@ mod tests { #[tokio::test] async fn subagent_completion_injects_followup_message() { let store = BackgroundStore::new(); - store.insert("abc123".into(), TaskKind::Subagent); + store.insert("abc123".into()); store.notify("abc123", TaskState::Completed("the answer is 42".into())); let hook = followup_from_background_store(store.clone()); @@ -1014,7 +982,7 @@ mod tests { #[tokio::test] async fn subagent_failure_injects_followup_with_error_marker() { let store = BackgroundStore::new(); - store.insert("xyz789".into(), TaskKind::Subagent); + store.insert("xyz789".into()); store.notify( "xyz789", TaskState::Failed("connection reset by peer".into()), @@ -1044,7 +1012,7 @@ mod tests { async fn followup_batches_multiple_completions_in_one_reminder() { let store = BackgroundStore::new(); for i in 0..3 { - store.insert(format!("t{i}"), TaskKind::Subagent); + store.insert(format!("t{i}")); store.notify( &format!("t{i}"), TaskState::Completed(format!("result-{i}")), @@ -1074,7 +1042,7 @@ mod tests { #[tokio::test] async fn followup_returns_empty_when_no_completions() { let store = BackgroundStore::new(); - store.insert("running".into(), TaskKind::Subagent); // inserted but not notified + store.insert("running".into()); // inserted but not notified let hook = followup_from_background_store(store); assert!(hook().await.is_empty()); } @@ -1087,7 +1055,7 @@ mod tests { #[tokio::test] async fn followup_drains_queue_once() { let store = BackgroundStore::new(); - store.insert("t1".into(), TaskKind::Subagent); + store.insert("t1".into()); store.notify("t1", TaskState::Completed("once".into())); let hook = followup_from_background_store(store); @@ -1109,7 +1077,7 @@ mod tests { let s = store.clone(); let id = format!("t{i}"); handles.push(std::thread::spawn(move || { - s.insert(id.clone(), TaskKind::Subagent); + s.insert(id.clone()); s.notify(&id, TaskState::Completed(format!("done-{i}"))); })); } diff --git a/src/agent/tools/bash.rs b/src/agent/tools/bash.rs index 5954c328..a9c2d2df 100644 --- a/src/agent/tools/bash.rs +++ b/src/agent/tools/bash.rs @@ -237,16 +237,128 @@ async fn run_with_timeout(cmd: Command, secs: u64) -> Result, +) -> tokio::task::JoinHandle<()> { + use crate::agent::tools::bg_shell::ShellStatus; + use std::process::Stdio; + tokio::spawn(async move { + let mut cmd = cmd; + cmd.stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + #[cfg(unix)] + cmd.process_group(0); + + let mut child = match cmd.spawn() { + Ok(c) => c, + Err(e) => { + store.finish(&id, ShellStatus::Failed(format!("failed to spawn: {e}"))); + return; + } + }; + let pid = child.id(); + #[cfg(unix)] + let _pgguard = pid.map(PgKillGuard::new); + + let stdout = child.stdout.take(); + let stderr = child.stderr.take(); + // Stream stdout + stderr line-by-line into the store in arrival + // order. The closure borrows `store`/`id`, so keep it scoped. + let drain = { + let store = store.clone(); + let id = id.clone(); + async move { + use tokio::io::AsyncBufReadExt; + let mut so = stdout.map(tokio::io::BufReader::new); + let mut se = stderr.map(tokio::io::BufReader::new); + loop { + let has_so = so.is_some(); + let has_se = se.is_some(); + if !has_so && !has_se { + break; + } + let mut so_buf = String::new(); + let mut se_buf = String::new(); + let so_fut = async { + match so.as_mut() { + Some(r) => r.read_line(&mut so_buf).await.map(Some), + None => Ok::<_, std::io::Error>(None), + } + }; + let se_fut = async { + match se.as_mut() { + Some(r) => r.read_line(&mut se_buf).await.map(Some), + None => Ok::<_, std::io::Error>(None), + } + }; + tokio::select! { + biased; + r = so_fut, if has_so => match r { + Ok(Some(0)) | Ok(None) | Err(_) => { so = None; } + Ok(Some(_)) => store.append(&id, &so_buf), + }, + r = se_fut, if has_se => match r { + Ok(Some(0)) | Ok(None) | Err(_) => { se = None; } + Ok(Some(_)) => store.append(&id, &se_buf), + }, + } + } + } + }; + + let wait = async { + drain.await; + child.wait().await + }; + + let status = match timeout { + Some(secs) => match tokio::time::timeout(Duration::from_secs(secs), wait).await { + Ok(Ok(st)) => ShellStatus::Exited(st.code().unwrap_or(-1)), + Ok(Err(e)) => ShellStatus::Failed(e.to_string()), + Err(_) => { + // Timed out: SIGKILL the whole group, then mark killed. + #[cfg(unix)] + if let Some(pid) = pid { + unsafe { + let _ = libc::kill(-(pid as libc::pid_t), libc::SIGKILL); + } + } + let _ = pid; + ShellStatus::Failed(format!("auto-killed after {secs}s timeout")) + } + }, + None => match wait.await { + Ok(st) => ShellStatus::Exited(st.code().unwrap_or(-1)), + Err(e) => ShellStatus::Failed(e.to_string()), + }, + }; + store.finish(&id, status); + }) +} + pub struct BashTool { pub permission: Option, pub ask_tx: Option, pub sandbox: Sandbox, cache: Option, - /// Shared background registry. When present, a `background: true` - /// command runs detached and is tracked here (counted as - /// `TaskKind::Shell`); when absent (e.g. some headless paths) the - /// `background` flag degrades gracefully to synchronous execution. - bg_store: Option, + /// Shared background-shell registry. When present, `background: true` + /// runs the command detached (unbounded) and tracks it here so the + /// model can read its output (`bash_output`) and stop it + /// (`kill_shell`). When absent (e.g. some headless paths) `background` + /// degrades gracefully to synchronous execution. + shell_store: Option, } impl BashTool { @@ -257,7 +369,7 @@ impl BashTool { ask_tx, sandbox, cache: None, - bg_store: None, + shell_store: None, } } @@ -272,18 +384,18 @@ impl BashTool { ask_tx, sandbox, cache: Some(cache), - bg_store: None, + shell_store: None, } } - /// Inject the shared background-task store so `background: true` - /// commands can run detached and be counted as background shells. - /// Chainable; `None` leaves the tool synchronous-only. - pub fn with_bg_store( + /// Inject the shared background-shell registry so `background: true` + /// commands run detached. Chainable; `None` leaves the tool + /// synchronous-only. + pub fn with_shell_store( mut self, - bg_store: Option, + shell_store: Option, ) -> Self { - self.bg_store = bg_store; + self.shell_store = shell_store; self } } @@ -335,56 +447,52 @@ impl Tool for BashTool { // pi (`bash.ts:76-81`) does this via `detached: true` + // `killProcessTree(pid)`. let background = args.background.unwrap_or(false); - // Background commands get a more generous default since they're - // meant for long-running work; explicit `timeout` still wins. - let secs = args.timeout.unwrap_or(if background { 600 } else { 120 }); - if secs == 0 { - return Err(ToolError::Msg("timeout must be > 0".to_string())); - } - // Detached/background path: spawn, register in the shared store as - // a Shell, and return immediately. Output is delivered later via - // the same background-completion notification subagents use. - // Degrades to synchronous if no store was injected. - if background && let Some(store) = &self.bg_store { - use crate::agent::tools::background::{TaskKind, TaskState}; - let running = store.running_count_kind(TaskKind::Shell); - let cap = crate::agent::tools::background::BackgroundStore::max_concurrent_shells(); + // Detached/background path (Claude-Code model): spawn UNBOUNDED, + // register in the shell store, and return immediately with an id. + // The model reads output with `bash_output` and stops it with + // `kill_shell`. `timeout`, if given, becomes an auto-kill-after-N. + // Degrades to synchronous if no shell store was injected. + if background && let Some(store) = &self.shell_store { + use crate::agent::tools::bg_shell::BackgroundShellStore; + if let Some(t) = args.timeout + && t == 0 + { + return Err(ToolError::Msg("timeout must be > 0".to_string())); + } + let running = store.running_count(); + let cap = BackgroundShellStore::max_concurrent(); if running >= cap { return Err(ToolError::Msg(format!( - "background shell cap reached ({running}/{cap} in flight). Wait for one to finish (its output arrives automatically, or check task_status) or run inline (background=false).", + "background shell cap reached ({running}/{cap} running). Stop one with kill_shell, or run inline (background=false).", ))); } let id = uuid::Uuid::new_v4().to_string(); - store.insert(id.clone(), TaskKind::Shell); - store.notify_started(&id); + store.register(id.clone(), command.clone()); // A backgrounded command may mutate the filesystem while it // runs; conservatively drop the per-turn read/grep/list cache. if let Some(ref cache) = self.cache { cache.clear(); } let wrapped = self.sandbox.wrap_command(&command); - let store_for_task = store.clone(); - let id_for_task = id.clone(); - let handle = tokio::spawn(async move { - let state = match run_with_timeout(wrapped, secs).await { - Ok(out) => { - let mut body = out.merged; - if out.exit_code != 0 { - body.push_str(&format!("\nExit code: {}", out.exit_code)); - } - TaskState::Completed(body) - } - Err(e) => TaskState::Failed(e.to_string()), - }; - store_for_task.notify(&id_for_task, state); - }); + let handle = spawn_streaming_shell(wrapped, store.clone(), id.clone(), args.timeout); store.attach_handle(&id, handle); + let timeout_note = match args.timeout { + Some(t) => format!(" (auto-killed after {t}s)"), + None => " (runs until it exits or you kill it)".to_string(), + }; return Ok(format!( - "background shell started — id: {id} (timeout {secs}s). It runs detached; its output is delivered automatically when it finishes — do NOT poll. To check sooner, use task_status with this id.", + "background shell started — id: {id}{timeout_note}. Read its output with bash_output (id: \"{id}\") and stop it with kill_shell (id: \"{id}\"). Output is NOT pushed to you — poll bash_output.", )); } + // Background requested but no store wired (headless): fall back to + // a bounded synchronous run. + let secs = args.timeout.unwrap_or(120); + if secs == 0 { + return Err(ToolError::Msg("timeout must be > 0".to_string())); + } + let output = run_with_timeout(self.sandbox.wrap_command(&command), secs).await?; // F12: `merged` already contains stdout + stderr in arrival @@ -652,18 +760,19 @@ mod tests { /// registers a `TaskKind::Shell` in the store, and delivers the /// command's output via the store's completion path once it finishes. #[tokio::test] - async fn background_bash_registers_shell_and_delivers_output() { + async fn background_bash_registers_shell_and_streams_output() { use crate::agent::tools::BashArgs; - use crate::agent::tools::background::{BackgroundStore, TaskKind, TaskState}; + use crate::agent::tools::bg_shell::{BackgroundShellStore, ShellStatus}; - let store = BackgroundStore::new(); + let store = BackgroundShellStore::new(); let tool = BashTool::new(None, None, crate::sandbox::Sandbox::new(false)) - .with_bg_store(Some(store.clone())); + .with_shell_store(Some(store.clone())); + // Unbounded background run (timeout: None) — Claude-Code model. let res = tool .call(BashArgs { command: "echo bg-hello".to_string(), - timeout: Some(10), + timeout: None, background: Some(true), }) .await @@ -673,32 +782,38 @@ mod tests { "expected an immediate start message, got: {res}" ); - // Parse the id out of "… id: (timeout …". + // Parse the id out of "… id: (…". let id = res .split("id: ") .nth(1) - .and_then(|s| s.split(' ').next()) + .and_then(|s| s.split(['(', ' ']).next()) .expect("id in start message") .to_string(); - // Poll until the background command finishes and its output lands. - let mut completed = None; - for _ in 0..100 { - if let Some(task) = store.get(&id) - && let TaskState::Completed(out) = task.state - { - completed = Some(out); - break; + // Poll bash_output's underlying read until the shell exits, and + // accumulate streamed output. + let mut out = String::new(); + let mut exited = false; + for _ in 0..200 { + if let Some((chunk, status)) = store.read_new(&id) { + out.push_str(&chunk); + if !status.is_running() { + assert!( + matches!(status, ShellStatus::Exited(0)), + "status: {status:?}" + ); + exited = true; + break; + } } tokio::time::sleep(std::time::Duration::from_millis(20)).await; } - let out = completed.expect("background shell should complete"); + assert!(exited, "background shell should exit"); assert!( out.contains("bg-hello"), - "expected command output, got: {out}" + "expected streamed output, got: {out}" ); - // Handle dropped on completion → no longer counted as a running shell. - assert_eq!(store.running_count_kind(TaskKind::Shell), 0); + assert_eq!(store.running_count(), 0); } /// Test helper: build a single op-based rule (tool-agnostic). diff --git a/src/agent/tools/bg_shell.rs b/src/agent/tools/bg_shell.rs new file mode 100644 index 00000000..e6fbb0a7 --- /dev/null +++ b/src/agent/tools/bg_shell.rs @@ -0,0 +1,458 @@ +//! Background-shell registry — the Claude-Code-style model for detached +//! `bash` commands: a command started with `background: true` runs +//! UNBOUNDED (no timeout kill), its output streams into a per-shell +//! buffer the model pulls incrementally via the `bash_output` tool, and +//! it is stopped explicitly via the `kill_shell` tool (or killed en-masse +//! when the session ends). +//! +//! This is distinct from the background-SUBAGENT store +//! (`background.rs`), which uses a push-once completion notification — +//! the wrong fit for long-lived processes (dev servers, watchers) that +//! never "complete" and whose output must be readable while running. +//! +//! Memory model: each entry holds only the UNREAD output. `read_new` +//! returns and clears it, so a model that polls regularly keeps the +//! buffer small; an unread buffer is hard-capped so a never-read flood +//! can't OOM. + +use indexmap::IndexMap; +use rig::completion::ToolDefinition; +use rig::tool::Tool; +use serde::Deserialize; +use std::sync::{Arc, Mutex}; +use tokio::task::JoinHandle; + +use crate::agent::tools::ToolError; + +/// Process-global background-shell registry. There is exactly one +/// interactive session per process, so the bash tool (which spawns +/// shells), the `bash_output`/`kill_shell` tools, the status bar, and +/// session-end cleanup all share this one instance — avoiding threading +/// it through every builder/UI signature. (Same pattern dirge already +/// uses for the subagent `/kill` abort registry.) Tests inject their own +/// store instead of touching the global, so they stay isolated. +static GLOBAL: std::sync::LazyLock = + std::sync::LazyLock::new(BackgroundShellStore::new); + +/// A clone (cheap — `Arc` inside) of the process-global shell registry. +pub fn global() -> BackgroundShellStore { + GLOBAL.clone() +} + +/// Hard cap on a single shell's UNREAD output buffer. Past this the +/// buffer stops growing and a one-time truncation marker is appended; +/// the model should `bash_output` regularly to drain it. +const MAX_UNREAD_BYTES: usize = 1024 * 1024; + +/// Max number of shells retained (running + finished-but-unread). Oldest +/// by insertion order is evicted past this — generous for any session. +const STORE_CAPACITY: usize = 32; + +/// Max concurrently RUNNING background shells. A runaway model shouldn't +/// be able to spawn unbounded detached processes. +const MAX_CONCURRENT_SHELLS: usize = 8; + +#[derive(Debug, Clone, PartialEq)] +pub enum ShellStatus { + Running, + Exited(i32), + /// Killed via `kill_shell` / session end. + Killed, + /// Failed to spawn or drain (carries the error). + Failed(String), +} + +impl ShellStatus { + pub fn is_running(&self) -> bool { + matches!(self, ShellStatus::Running) + } + /// One-word label for the model-facing output / `/tasks` listing. + pub fn label(&self) -> String { + match self { + ShellStatus::Running => "running".to_string(), + ShellStatus::Exited(code) => format!("exited({code})"), + ShellStatus::Killed => "killed".to_string(), + ShellStatus::Failed(e) => format!("failed: {e}"), + } + } +} + +struct ShellEntry { + /// The command line, for the `/tasks` listing and tool feedback. + command: String, + /// Output produced since the last `read_new` (drained on read). + unread: String, + /// True once `unread` hit the cap and we stopped appending. + truncated: bool, + status: ShellStatus, + /// Drain task handle. Aborting it drops the child + its + /// `PgKillGuard`, SIGKILLing the whole process group. `None` once the + /// shell has reached a terminal state. + handle: Option>, +} + +/// One row of the `/tasks` listing / model-facing status. +#[derive(Debug, Clone, PartialEq)] +pub struct ShellInfo { + pub id: String, + pub command: String, + pub status: ShellStatus, +} + +/// Thread-safe registry of background shells. Cloneable (`Arc` inside) so +/// the bash tool, the `bash_output`/`kill_shell` tools, and the UI all +/// share one instance. +#[derive(Debug, Clone, Default)] +pub struct BackgroundShellStore { + inner: Arc>>, +} + +// Manual Debug for ShellEntry (JoinHandle isn't Debug-friendly to print). +impl std::fmt::Debug for ShellEntry { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ShellEntry") + .field("command", &self.command) + .field("unread_len", &self.unread.len()) + .field("truncated", &self.truncated) + .field("status", &self.status) + .field("has_handle", &self.handle.is_some()) + .finish() + } +} + +impl BackgroundShellStore { + pub fn new() -> Self { + Self::default() + } + + /// Compile-time cap on concurrently running shells. + pub fn max_concurrent() -> usize { + MAX_CONCURRENT_SHELLS + } + + fn lock(&self) -> std::sync::MutexGuard<'_, IndexMap> { + self.inner.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// Register a freshly-started shell in `Running` state. Evicts the + /// oldest entry if at capacity. + pub fn register(&self, id: String, command: String) { + let mut map = self.lock(); + if !map.contains_key(&id) && map.len() >= STORE_CAPACITY { + map.shift_remove_index(0); + } + map.insert( + id, + ShellEntry { + command, + unread: String::new(), + truncated: false, + status: ShellStatus::Running, + handle: None, + }, + ); + } + + /// Attach the drain-task handle so `kill`/`kill_all` can abort it. + pub fn attach_handle(&self, id: &str, handle: JoinHandle<()>) { + let mut map = self.lock(); + match map.get_mut(id) { + // Only keep the handle while the shell is still tracked AND + // running; a race where the drain finished first leaves the + // terminal status intact and just drops the (already-done) + // handle. + Some(e) if e.status.is_running() => e.handle = Some(handle), + _ => handle.abort(), + } + } + + /// Append a chunk of streamed output. No-op for an unknown id. + /// Bounded: once the unread buffer hits the cap, further output is + /// dropped with a one-time marker (the model should poll to drain). + pub fn append(&self, id: &str, chunk: &str) { + let mut map = self.lock(); + let Some(e) = map.get_mut(id) else { + return; + }; + if e.unread.len() + chunk.len() <= MAX_UNREAD_BYTES { + e.unread.push_str(chunk); + } else if !e.truncated { + e.truncated = true; + let room = MAX_UNREAD_BYTES.saturating_sub(e.unread.len()); + // Push a UTF-8-safe prefix of the chunk up to the cap, then a marker. + let mut take = room.min(chunk.len()); + while take > 0 && !chunk.is_char_boundary(take) { + take -= 1; + } + e.unread.push_str(&chunk[..take]); + e.unread.push_str( + "\n…[background shell output exceeded the unread-buffer cap; call bash_output more often to drain it]", + ); + } + } + + /// Record a terminal status and drop the drain handle. No-op if the + /// id is unknown or already terminal (first terminal wins, so a + /// `kill` racing a natural exit doesn't clobber the real exit code). + pub fn finish(&self, id: &str, status: ShellStatus) { + let mut map = self.lock(); + if let Some(e) = map.get_mut(id) + && e.status.is_running() + { + e.status = status; + e.handle = None; + } + } + + /// Return output produced since the last read (clearing it) plus the + /// current status. `None` if the id is unknown. + pub fn read_new(&self, id: &str) -> Option<(String, ShellStatus)> { + let mut map = self.lock(); + let e = map.get_mut(id)?; + let out = std::mem::take(&mut e.unread); + e.truncated = false; + Some((out, e.status.clone())) + } + + /// Kill a running shell by id: abort its drain task (which drops the + /// child and SIGKILLs the process group) and mark it `Killed`. + /// Returns true if a running shell was found and killed. + pub fn kill(&self, id: &str) -> bool { + let mut map = self.lock(); + let Some(e) = map.get_mut(id) else { + return false; + }; + if !e.status.is_running() { + return false; + } + if let Some(h) = e.handle.take() { + h.abort(); + } + e.status = ShellStatus::Killed; + true + } + + /// Number of shells currently running (drives the status bar). + pub fn running_count(&self) -> usize { + self.lock() + .values() + .filter(|e| e.status.is_running()) + .count() + } + + /// Snapshot of every tracked shell, newest last. + pub fn list(&self) -> Vec { + self.lock() + .iter() + .map(|(id, e)| ShellInfo { + id: id.clone(), + command: e.command.clone(), + status: e.status.clone(), + }) + .collect() + } + + /// Abort every running shell — called on session swap / shutdown so + /// detached processes don't outlive the session that started them. + pub fn kill_all(&self) { + let mut map = self.lock(); + for e in map.values_mut() { + if e.status.is_running() { + if let Some(h) = e.handle.take() { + h.abort(); + } + e.status = ShellStatus::Killed; + } + } + } +} + +// ── Model-facing tools ────────────────────────────────────────────── + +#[derive(Deserialize)] +pub struct BashOutputArgs { + pub id: String, +} + +/// `bash_output` — read output produced by a background shell since the +/// last read, plus its current status. Mirrors Claude Code's BashOutput. +pub struct BashOutputTool { + store: BackgroundShellStore, +} + +impl BashOutputTool { + pub fn new(store: BackgroundShellStore) -> Self { + Self { store } + } +} + +impl Tool for BashOutputTool { + const NAME: &'static str = "bash_output"; + type Error = ToolError; + type Args = BashOutputArgs; + type Output = String; + + async fn definition(&self, _prompt: String) -> ToolDefinition { + ToolDefinition { + name: "bash_output".to_string(), + description: "Read new output from a background shell (one started with bash(background=true)). Returns the output produced since your last call plus the shell's status (running / exited(code) / killed / failed). Poll this to follow a long-running command; call kill_shell to stop it.".to_string(), + parameters: serde_json::json!({ + "type": "object", + "properties": { + "id": { "type": "string", "description": "The background shell id returned by bash(background=true)." } + }, + "required": ["id"] + }), + } + } + + async fn call(&self, args: BashOutputArgs) -> Result { + match self.store.read_new(&args.id) { + Some((out, status)) => { + let body = if out.is_empty() { + "(no new output)".to_string() + } else { + out + }; + Ok(format!( + "[shell {} — {}]\n{}", + args.id, + status.label(), + body + )) + } + None => Err(ToolError::Msg(format!( + "no background shell with id {:?} (it may have been evicted)", + args.id + ))), + } + } +} + +#[derive(Deserialize)] +pub struct KillShellArgs { + pub id: String, +} + +/// `kill_shell` — stop a running background shell by id (SIGKILLs its +/// process group). Mirrors Claude Code's KillShell / TaskStop. +pub struct KillShellTool { + store: BackgroundShellStore, +} + +impl KillShellTool { + pub fn new(store: BackgroundShellStore) -> Self { + Self { store } + } +} + +impl Tool for KillShellTool { + const NAME: &'static str = "kill_shell"; + type Error = ToolError; + type Args = KillShellArgs; + type Output = String; + + async fn definition(&self, _prompt: String) -> ToolDefinition { + ToolDefinition { + name: "kill_shell".to_string(), + description: "Stop a running background shell (one started with bash(background=true)) by id. Kills the whole process group. No-op if it already exited.".to_string(), + parameters: serde_json::json!({ + "type": "object", + "properties": { + "id": { "type": "string", "description": "The background shell id to kill." } + }, + "required": ["id"] + }), + } + } + + async fn call(&self, args: KillShellArgs) -> Result { + if self.store.kill(&args.id) { + Ok(format!("killed background shell {}", args.id)) + } else { + Ok(format!( + "no running background shell with id {:?} (already exited or unknown)", + args.id + )) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn read_new_drains_unread_and_reports_status() { + let s = BackgroundShellStore::new(); + s.register("a".into(), "sleep 1".into()); + s.append("a", "line1\n"); + s.append("a", "line2\n"); + let (out, st) = s.read_new("a").unwrap(); + assert_eq!(out, "line1\nline2\n"); + assert_eq!(st, ShellStatus::Running); + // Second read sees only new output. + s.append("a", "line3\n"); + let (out2, _) = s.read_new("a").unwrap(); + assert_eq!(out2, "line3\n"); + // Nothing new. + let (out3, _) = s.read_new("a").unwrap(); + assert_eq!(out3, ""); + } + + #[test] + fn read_new_unknown_id_is_none() { + let s = BackgroundShellStore::new(); + assert!(s.read_new("nope").is_none()); + } + + #[test] + fn finish_sets_terminal_and_first_terminal_wins() { + let s = BackgroundShellStore::new(); + s.register("a".into(), "x".into()); + assert_eq!(s.running_count(), 1); + s.finish("a", ShellStatus::Exited(0)); + let (_, st) = s.read_new("a").unwrap(); + assert_eq!(st, ShellStatus::Exited(0)); + assert_eq!(s.running_count(), 0); + // A late kill/finish does not clobber the recorded exit. + s.finish("a", ShellStatus::Killed); + assert_eq!(s.read_new("a").unwrap().1, ShellStatus::Exited(0)); + } + + #[test] + fn kill_marks_killed_only_when_running() { + let s = BackgroundShellStore::new(); + s.register("a".into(), "x".into()); + assert!(s.kill("a")); + assert_eq!(s.read_new("a").unwrap().1, ShellStatus::Killed); + // Already terminal → kill is a no-op. + assert!(!s.kill("a")); + assert!(!s.kill("unknown")); + } + + #[test] + fn unread_buffer_is_capped() { + let s = BackgroundShellStore::new(); + s.register("a".into(), "flood".into()); + let chunk = "x".repeat(100_000); + for _ in 0..20 { + s.append("a", &chunk); + } + let (out, _) = s.read_new("a").unwrap(); + assert!(out.len() <= MAX_UNREAD_BYTES + 200, "len was {}", out.len()); + assert!(out.contains("exceeded the unread-buffer cap")); + } + + #[test] + fn list_reports_all_shells() { + let s = BackgroundShellStore::new(); + s.register("a".into(), "cmd-a".into()); + s.register("b".into(), "cmd-b".into()); + s.finish("b", ShellStatus::Exited(1)); + let rows = s.list(); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].command, "cmd-a"); + assert_eq!(rows[1].status, ShellStatus::Exited(1)); + } +} diff --git a/src/agent/tools/mod.rs b/src/agent/tools/mod.rs index 43c18699..81cd6016 100644 --- a/src/agent/tools/mod.rs +++ b/src/agent/tools/mod.rs @@ -1,6 +1,7 @@ pub(crate) mod apply_patch; pub(crate) mod background; mod bash; +pub(crate) mod bg_shell; pub(crate) mod cache; pub(crate) mod edit; mod find_files; @@ -30,6 +31,7 @@ pub(crate) mod write; pub use apply_patch::ApplyPatchTool; pub use bash::BashTool; +pub use bg_shell::{BashOutputTool, KillShellTool}; pub use cache::ToolCache; pub use edit::EditTool; pub use find_files::FindFilesTool; @@ -88,6 +90,8 @@ pub const BUILTIN_TOOL_NAMES: &[&str] = &[ "skill", "task", "task_status", + "bash_output", + "kill_shell", "tool_search", "question", "webfetch", diff --git a/src/agent/tools/task.rs b/src/agent/tools/task.rs index f144ddab..b92563bf 100644 --- a/src/agent/tools/task.rs +++ b/src/agent/tools/task.rs @@ -320,9 +320,7 @@ impl Tool for TaskTool { // concurrency cap. The agent gets a clear refusal it // can act on (wait for an existing task to finish, then // retry) rather than fanning out unbounded. - let running = self - .bg_store - .running_count_kind(crate::agent::tools::background::TaskKind::Subagent); + let running = self.bg_store.running_count(); let cap = BackgroundStore::max_concurrent(); if running >= cap { return Err(ToolError::Msg(format!( @@ -331,10 +329,7 @@ impl Tool for TaskTool { ))); } let task_id = Uuid::new_v4().to_string(); - self.bg_store.insert( - task_id.clone(), - crate::agent::tools::background::TaskKind::Subagent, - ); + self.bg_store.insert(task_id.clone()); self.bg_store.notify_started(&task_id); // dirge-ov2 Phase D: announce the subagent so the UI diff --git a/src/agent/tools/task_status.rs b/src/agent/tools/task_status.rs index 5d14b1f0..44f8faf0 100644 --- a/src/agent/tools/task_status.rs +++ b/src/agent/tools/task_status.rs @@ -162,10 +162,7 @@ mod tests { #[tokio::test] async fn test_task_status_running() { let store = BackgroundStore::new(); - store.insert( - "test-task".to_string(), - crate::agent::tools::background::TaskKind::Subagent, - ); + store.insert("test-task".to_string()); let tool = TaskStatusTool::new(store); let result = tool .call(TaskStatusArgs { @@ -180,10 +177,7 @@ mod tests { #[tokio::test] async fn test_task_status_completed() { let store = BackgroundStore::new(); - store.insert( - "test-task".to_string(), - crate::agent::tools::background::TaskKind::Subagent, - ); + store.insert("test-task".to_string()); store.notify("test-task", TaskState::Completed("result text".to_string())); let tool = TaskStatusTool::new(store); let result = tool @@ -200,10 +194,7 @@ mod tests { #[tokio::test] async fn test_task_status_failed() { let store = BackgroundStore::new(); - store.insert( - "test-task".to_string(), - crate::agent::tools::background::TaskKind::Subagent, - ); + store.insert("test-task".to_string()); store.notify("test-task", TaskState::Failed("error message".to_string())); let tool = TaskStatusTool::new(store); let result = tool @@ -220,10 +211,7 @@ mod tests { #[tokio::test] async fn test_task_status_wait_completed() { let store = BackgroundStore::new(); - store.insert( - "test-task".to_string(), - crate::agent::tools::background::TaskKind::Subagent, - ); + store.insert("test-task".to_string()); // Update to completed after a short delay let store_clone = store.clone(); @@ -281,10 +269,7 @@ mod tests { #[tokio::test] async fn status_lookup_is_idempotent() { let store = BackgroundStore::new(); - store.insert( - "t1".into(), - crate::agent::tools::background::TaskKind::Subagent, - ); + store.insert("t1".into()); store.notify("t1", TaskState::Completed("payload".into())); let tool = TaskStatusTool::new(store); @@ -306,10 +291,7 @@ mod tests { #[tokio::test] async fn wait_returns_on_failure() { let store = BackgroundStore::new(); - store.insert( - "t1".into(), - crate::agent::tools::background::TaskKind::Subagent, - ); + store.insert("t1".into()); let store_clone = store.clone(); tokio::spawn(async move { @@ -359,10 +341,7 @@ mod tests { #[tokio::test(start_paused = true)] async fn wait_returns_timeout_message_when_task_stays_running() { let store = BackgroundStore::new(); - store.insert( - "forever".into(), - crate::agent::tools::background::TaskKind::Subagent, - ); + store.insert("forever".into()); let tool = TaskStatusTool::new(store); // Drive the future with a parallel timer that advances past the cap. diff --git a/src/ui/mod.rs b/src/ui/mod.rs index 2df71de4..acca47b2 100644 --- a/src/ui/mod.rs +++ b/src/ui/mod.rs @@ -188,6 +188,10 @@ pub async fn run_interactive( input.load_history_entry(content); } } + // The process-global background-shell registry — shared with the + // `bash`/`bash_output`/`kill_shell` tools so the status bar's + // `shells:N` count reflects the same shells the model spawned. + let shell_store = Some(crate::agent::tools::bg_shell::global()); let mut is_running = false; // Plain-text messages typed while the agent is running are pushed here // instead of being rejected. The loop polls this queue at turn boundaries @@ -473,6 +477,7 @@ pub async fn run_interactive( context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), + shell_store.as_ref(), ), interjection_queue.lock().unwrap().len(), ), @@ -710,6 +715,11 @@ pub async fn run_interactive( if let Some(store) = bg_store.as_ref() { store.cancel_all(); } + // Likewise stop any detached background shells — they + // belong to the previous session and shouldn't outlive it. + if let Some(store) = shell_store.as_ref() { + store.kill_all(); + } // Repaint chat from the (possibly fresh) session so // the user sees the new state. The agent runtime // keeps the same model — reset_to_new / switch_session @@ -730,7 +740,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -764,7 +774,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -778,7 +788,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -787,7 +797,7 @@ pub async fn run_interactive( input.handle_paste(&text); renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -800,7 +810,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -816,7 +826,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -826,7 +836,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -901,7 +911,7 @@ pub async fn run_interactive( )?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; } else { @@ -922,7 +932,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -957,7 +967,7 @@ pub async fn run_interactive( renderer.write_line(msg, c_error())?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -974,7 +984,7 @@ pub async fn run_interactive( } renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; if rewind_picker.active { @@ -989,7 +999,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1002,7 +1012,7 @@ pub async fn run_interactive( rewind_picker.draw()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1011,7 +1021,7 @@ pub async fn run_interactive( renderer.write_line("Press Esc again to rewind...", theme::dim())?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1031,7 +1041,7 @@ pub async fn run_interactive( )?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1102,7 +1112,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1155,7 +1165,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1168,7 +1178,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1178,7 +1188,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1188,7 +1198,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1197,7 +1207,7 @@ pub async fn run_interactive( renderer.scroll_to_bottom()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1210,7 +1220,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; if let Some(ref picker) = input.picker { @@ -1247,7 +1257,7 @@ pub async fn run_interactive( } renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1285,7 +1295,7 @@ pub async fn run_interactive( )?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1306,7 +1316,7 @@ pub async fn run_interactive( )?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1361,7 +1371,7 @@ pub async fn run_interactive( } renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1403,7 +1413,7 @@ pub async fn run_interactive( )?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -1741,7 +1751,7 @@ pub async fn run_interactive( } renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; if let Some(ref picker) = input.picker { @@ -2463,7 +2473,7 @@ pub async fn run_interactive( } renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; if let Some(ref picker) = input.picker { @@ -2589,6 +2599,7 @@ pub async fn run_interactive( context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), + shell_store.as_ref(), ), interjection_queue.lock().unwrap().len(), ), @@ -2670,6 +2681,7 @@ pub async fn run_interactive( context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), + shell_store.as_ref(), ), interjection_queue.lock().unwrap().len(), ), @@ -2692,7 +2704,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -2941,7 +2953,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; if let Some(ref picker) = input.picker { @@ -3026,6 +3038,7 @@ pub async fn run_interactive( context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), + shell_store.as_ref(), ), interjection_queue.lock().unwrap().len(), ), @@ -3089,7 +3102,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; } @@ -3318,7 +3331,7 @@ pub async fn run_interactive( is_running = true; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; } @@ -3471,7 +3484,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; @@ -3546,7 +3559,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; let ev = user_rx.recv().await; @@ -3636,7 +3649,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; if let Some(ref picker) = input.picker { @@ -3700,7 +3713,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -3770,7 +3783,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -3816,7 +3829,7 @@ pub async fn run_interactive( } renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; } @@ -3856,7 +3869,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; continue; @@ -3929,7 +3942,7 @@ pub async fn run_interactive( renderer.render_viewport()?; renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; if let Some(ref picker) = input.picker { @@ -3939,7 +3952,7 @@ pub async fn run_interactive( _ = tokio::time::sleep(tokio::time::Duration::from_millis(200)), if is_running => { renderer.draw_bottom( &input, - &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref()), interjection_queue.lock().unwrap().len()), + &with_queue(StatusLine::render(session, is_running, 0, loop_label.as_deref(), context.current_prompt_name.as_deref(), perm_mode().as_deref(), bg_store.as_ref(), shell_store.as_ref()), interjection_queue.lock().unwrap().len()), is_running, )?; if let Some(ref picker) = input.picker { diff --git a/src/ui/slash/cmd_session.rs b/src/ui/slash/cmd_session.rs index debf2e60..f9038778 100644 --- a/src/ui/slash/cmd_session.rs +++ b/src/ui/slash/cmd_session.rs @@ -216,6 +216,20 @@ pub(super) async fn cmd_tasks(ctx: &mut SlashCtx<'_>) -> anyhow::Result<()> { theme::dim(), )?; } + + // Background shells (bash background=true). Listed with their status + // so the user can see what's running and kill it via the model's + // kill_shell tool (or it auto-cleans on session end). + let shells = crate::agent::tools::bg_shell::global().list(); + if !shells.is_empty() { + ctx.renderer.write_line("background shells:", c_result())?; + for s in &shells { + ctx.renderer.write_line( + &format!(" [{}] {} — {}", s.status.label(), s.id, s.command), + c_result(), + )?; + } + } Ok(()) } diff --git a/src/ui/status.rs b/src/ui/status.rs index 169097af..bfabd221 100644 --- a/src/ui/status.rs +++ b/src/ui/status.rs @@ -1,6 +1,7 @@ use std::path::{Path, PathBuf}; -use crate::agent::tools::background::{BackgroundStore, TaskKind}; +use crate::agent::tools::background::BackgroundStore; +use crate::agent::tools::bg_shell::BackgroundShellStore; use crate::session::Session; pub struct StatusLine; @@ -58,6 +59,7 @@ impl StatusLine { prompt_name: Option<&str>, perm_mode: Option<&str>, bg_store: Option<&BackgroundStore>, + shell_store: Option<&BackgroundShellStore>, ) -> String { let state = if is_running { "running" } else { "ready" }; let wd_path = Path::new(&session.working_dir); @@ -110,13 +112,8 @@ impl StatusLine { // Active background work, counted per kind. Each badge is shown // only when non-zero, like the other conditional badges, so the // bar stays quiet during normal single-agent work. - let (active_agents, active_shells) = match bg_store { - Some(s) => ( - s.running_count_kind(TaskKind::Subagent), - s.running_count_kind(TaskKind::Shell), - ), - None => (0, 0), - }; + let active_agents = bg_store.map(|s| s.running_count()).unwrap_or(0); + let active_shells = shell_store.map(|s| s.running_count()).unwrap_or(0); let agents_badge = if active_agents > 0 { format!(" | agents:{}", active_agents) } else { @@ -151,39 +148,43 @@ impl StatusLine { #[cfg(test)] mod tests { use super::StatusLine; - use crate::agent::tools::background::{BackgroundStore, TaskKind}; + use crate::agent::tools::background::BackgroundStore; + use crate::agent::tools::bg_shell::BackgroundShellStore; use crate::session::Session; - /// Build a store with `agents` running subagents and `shells` running - /// shells (each needs a live handle to count, so attach a never-ending - /// spawned task per entry). - fn store_with(agents: usize, shells: usize) -> BackgroundStore { + /// A subagent store with `agents` running subagents (each needs a + /// live handle to count, so attach a never-ending spawned task). + fn agent_store(agents: usize) -> BackgroundStore { let store = BackgroundStore::new(); - let mut n = 0; - for (kind, count) in [(TaskKind::Subagent, agents), (TaskKind::Shell, shells)] { - for _ in 0..count { - let id = format!("id{n}"); - n += 1; - store.insert(id.clone(), kind); - let h = tokio::runtime::Handle::try_current() - .ok() - .map(|_| tokio::spawn(async { std::future::pending::<()>().await })); - if let Some(h) = h { - store.attach_handle(&id, h); - } + for n in 0..agents { + let id = format!("a{n}"); + store.insert(id.clone()); + if tokio::runtime::Handle::try_current().is_ok() { + store.attach_handle(&id, tokio::spawn(std::future::pending::<()>())); } } store } - fn render(store: &BackgroundStore) -> String { + /// A shell store with `shells` running shells. + fn shell_store(shells: usize) -> BackgroundShellStore { + let store = BackgroundShellStore::new(); + for n in 0..shells { + store.register(format!("s{n}"), "cmd".to_string()); + } + store + } + + fn render(agents: usize, shells: usize) -> String { let session = Session::new("openrouter", "test-model", 100_000); - StatusLine::render(&session, false, 0, None, None, None, Some(store)) + let a = agent_store(agents); + let s = shell_store(shells); + StatusLine::render(&session, false, 0, None, None, None, Some(&a), Some(&s)) } #[tokio::test] async fn badges_hidden_when_nothing_active() { - let line = render(&store_with(0, 0)); + let line = render(0, 0); assert!( !line.contains("agents:"), "no agents badge expected: {line}" @@ -196,7 +197,7 @@ mod tests { #[tokio::test] async fn agents_and_shells_counted_separately() { - let line = render(&store_with(2, 3)); + let line = render(2, 3); assert!(line.contains("agents:2"), "expected agents:2 in: {line}"); assert!(line.contains("shells:3"), "expected shells:3 in: {line}"); }