diff --git a/src/agent/mod.rs b/src/agent/mod.rs index 0874701..da8aecb 100644 --- a/src/agent/mod.rs +++ b/src/agent/mod.rs @@ -24,7 +24,7 @@ use crate::user::api::ApiClient; use crate::agent::context as journal; use crate::user::log::ConversationLog; use crate::user::api::StreamEvent; -use crate::agent::tools::{ProcessTracker, ToolCall, ToolDef, FunctionCall, summarize_args}; +use crate::agent::tools::{ToolCall, ToolDef, FunctionCall, summarize_args}; use crate::user::types::*; use crate::user::ui_channel::{ContextSection, SharedContextState, StatusInfo, StreamTarget, UiMessage, UiSender}; @@ -59,8 +59,6 @@ pub struct Agent { tool_defs: Vec, /// Last known prompt token count from the API (tracks context size). last_prompt_tokens: u32, - /// Shared process tracker for bash tool — lets TUI show/kill running commands. - pub process_tracker: ProcessTracker, /// Current reasoning effort level ("none", "low", "high"). pub reasoning_effort: String, /// Persistent conversation log — append-only record of all messages. @@ -125,7 +123,6 @@ impl Agent { client, tool_defs, last_prompt_tokens: 0, - process_tracker: ProcessTracker::new(), reasoning_effort: "none".to_string(), conversation_log, tokenizer, @@ -259,7 +256,7 @@ impl Agent { } done }; - for mut entry in finished { + for entry in finished { if let Ok((call, output)) = entry.handle.await { self.apply_tool_result(&call, output, ui_tx, &mut bg_ds); } @@ -334,9 +331,8 @@ impl Agent { .unwrap_or(false); let call_id = call.id.clone(); let call_name = call.function.name.clone(); - let tracker = self.process_tracker.clone(); let handle = tokio::spawn(async move { - let output = tools::dispatch(&call.function.name, &args, &tracker).await; + let output = tools::dispatch(&call.function.name, &args).await; (call, output) }); self.active_tools.lock().unwrap().push( @@ -510,7 +506,7 @@ impl Agent { }; if !pending.is_empty() { self.push_message(msg.clone()); - for mut entry in pending { + for entry in pending { if let Ok((call, output)) = entry.handle.await { self.apply_tool_result(&call, output, ui_tx, &mut ds); } @@ -586,9 +582,8 @@ impl Agent { let call_id = call.id.clone(); let call_name = call.function.name.clone(); let call = call.clone(); - let tracker = self.process_tracker.clone(); let handle = tokio::spawn(async move { - let output = tools::dispatch(&call.function.name, &args, &tracker).await; + let output = tools::dispatch(&call.function.name, &args).await; (call, output) }); self.active_tools.lock().unwrap().push( diff --git a/src/agent/tools/bash.rs b/src/agent/tools/bash.rs index e525851..e5dab6c 100644 --- a/src/agent/tools/bash.rs +++ b/src/agent/tools/bash.rs @@ -12,7 +12,7 @@ use serde_json::json; use std::process::Stdio; use tokio::io::AsyncReadExt; -use super::{ToolDef, ProcessTracker, default_timeout}; +use super::{ToolDef, default_timeout}; /// RAII guard that SIGTERMs the process group on drop. /// Ensures child processes are cleaned up when a task is aborted. @@ -55,7 +55,7 @@ pub fn definition() -> ToolDef { ) } -pub async fn run_bash(args: &serde_json::Value, tracker: &ProcessTracker) -> Result { +pub async fn run_bash(args: &serde_json::Value) -> Result { let a: Args = serde_json::from_value(args.clone()) .context("invalid bash arguments")?; let command = &a.command; @@ -73,7 +73,6 @@ pub async fn run_bash(args: &serde_json::Value, tracker: &ProcessTracker) -> Res let pid = child.id().unwrap_or(0); let kill_guard = KillOnDrop(pid); - tracker.register(pid, command).await; // Take ownership of stdout/stderr handles before waiting, // so we can still kill the child on timeout. @@ -139,14 +138,12 @@ pub async fn run_bash(args: &serde_json::Value, tracker: &ProcessTracker) -> Res Err(anyhow::anyhow!("Command failed: {}", e)) } Err(_) => { - // Timeout — kill the process group - tracker.kill(pid).await; + // Timeout — KillOnDrop will SIGTERM the process group Err(anyhow::anyhow!("Command timed out after {}s: {}", timeout_secs, command)) } }; // Process completed normally — defuse the kill guard std::mem::forget(kill_guard); - tracker.unregister(pid).await; result } diff --git a/src/agent/tools/mod.rs b/src/agent/tools/mod.rs index 70a5402..4ea4de8 100644 --- a/src/agent/tools/mod.rs +++ b/src/agent/tools/mod.rs @@ -19,9 +19,7 @@ mod vision; pub mod working_stack; use serde::{Serialize, Deserialize}; -use std::sync::Arc; use std::time::Instant; -use tokio::sync::Mutex; fn default_timeout() -> u64 { 120 } @@ -110,14 +108,6 @@ impl ToolOutput { } } -/// Info about a running child process, visible to the TUI. -#[derive(Debug, Clone)] -pub struct ProcessInfo { - pub pid: u32, - pub command: String, - pub started: Instant, -} - /// A tool call in flight — metadata for TUI + JoinHandle for /// result collection and cancellation. pub struct ActiveToolCall { @@ -129,53 +119,6 @@ pub struct ActiveToolCall { pub handle: tokio::task::JoinHandle<(ToolCall, ToolOutput)>, } -/// Shared tracker for running child processes. Allows the TUI to -/// display what's running and kill processes by PID. -#[derive(Debug, Clone, Default)] -pub struct ProcessTracker { - inner: Arc>>, -} - -impl ProcessTracker { - pub fn new() -> Self { - Self::default() - } - - async fn register(&self, pid: u32, command: &str) { - self.inner.lock().await.push(ProcessInfo { - pid, - command: if command.len() > 120 { - format!("{}...", &command[..120]) - } else { - command.to_string() - }, - started: Instant::now(), - }); - } - - async fn unregister(&self, pid: u32) { - self.inner.lock().await.retain(|p| p.pid != pid); - } - - /// Snapshot of currently running processes. - pub async fn list(&self) -> Vec { - self.inner.lock().await.clone() - } - - /// Kill a process by PID. Returns true if the signal was sent. - pub async fn kill(&self, pid: u32) -> bool { - // SIGTERM the process group (negative PID kills the group) - let ret = unsafe { libc::kill(-(pid as i32), libc::SIGTERM) }; - if ret != 0 { - // Try just the process - unsafe { libc::kill(pid as i32, libc::SIGTERM) }; - } - // Don't unregister — let the normal exit path do that - // so the tool result says "killed by user" - true - } -} - /// Truncate output if it exceeds max length, appending a truncation notice. pub fn truncate_output(mut s: String, max: usize) -> String { if s.len() > max { @@ -197,7 +140,6 @@ pub fn truncate_output(mut s: String, max: usize) -> String { pub async fn dispatch( name: &str, args: &serde_json::Value, - tracker: &ProcessTracker, ) -> ToolOutput { // Agent-specific tools let rich_result = match name { @@ -211,7 +153,7 @@ pub async fn dispatch( return result.unwrap_or_else(ToolOutput::error); } - if let Some(output) = dispatch_shared(name, args, tracker, None).await { + if let Some(output) = dispatch_shared(name, args, None).await { return output; } @@ -224,7 +166,6 @@ pub async fn dispatch( pub async fn dispatch_shared( name: &str, args: &serde_json::Value, - tracker: &ProcessTracker, provenance: Option<&str>, ) -> Option { // Memory and journal tools @@ -241,7 +182,7 @@ pub async fn dispatch_shared( "read_file" => read::read_file(args), "write_file" => write::write_file(args), "edit_file" => edit::edit_file(args), - "bash" => bash::run_bash(args, tracker).await, + "bash" => bash::run_bash(args).await, "grep" => grep::grep(args), "glob" => glob::glob_search(args), _ => return None, diff --git a/src/bin/consciousness.rs b/src/bin/consciousness.rs index f7f053c..d9c8b9a 100644 --- a/src/bin/consciousness.rs +++ b/src/bin/consciousness.rs @@ -33,7 +33,7 @@ use clap::Parser; use poc_memory::dbglog; use poc_memory::user::*; -use poc_memory::agent::{tools, Agent, TurnResult}; +use poc_memory::agent::{Agent, TurnResult}; use poc_memory::user::api::ApiClient; use poc_memory::user::tui::HotkeyAction; use poc_memory::config::{self, AppConfig, SessionConfig}; @@ -114,7 +114,6 @@ enum Command { struct Session { agent: Arc>, config: SessionConfig, - process_tracker: tools::ProcessTracker, ui_tx: ui_channel::UiSender, turn_tx: mpsc::Sender<(Result, StreamTarget)>, // DMN state @@ -140,7 +139,6 @@ impl Session { fn new( agent: Arc>, config: SessionConfig, - process_tracker: tools::ProcessTracker, ui_tx: ui_channel::UiSender, turn_tx: mpsc::Sender<(Result, StreamTarget)>, ) -> Self { @@ -149,7 +147,6 @@ impl Session { Self { agent, config, - process_tracker, ui_tx, turn_tx, dmn: if dmn::is_off() { @@ -581,13 +578,17 @@ impl Session { /// Interrupt: kill processes, abort current turn, clear pending queue. async fn interrupt(&mut self) { - let procs = self.process_tracker.list().await; - for p in &procs { - self.process_tracker.kill(p.pid).await; - } - // Only abort the turn if no processes are running — let SIGTERM'd - // processes exit normally so run_bash can unregister them. - if procs.is_empty() { + // Abort all active tool calls (KillOnDrop sends SIGTERM) + let count = { + let agent = self.agent.lock().await; + let mut tools = agent.active_tools.lock().unwrap(); + let count = tools.len(); + for entry in tools.drain(..) { + entry.handle.abort(); + } + count + }; + if count == 0 { if let Some(handle) = self.turn_handle.take() { handle.abort(); self.turn_in_progress = false; @@ -599,7 +600,7 @@ impl Session { } } self.pending_input = None; - let killed = procs.len(); + let killed = count; if killed > 0 || self.turn_in_progress { let _ = self.ui_tx.send(UiMessage::Info(format!( "(interrupted — killed {} process(es), turn aborted)", @@ -639,28 +640,25 @@ impl Session { } } - /// Show and kill running processes (Ctrl+K). + /// Show and kill running tool calls (Ctrl+K). async fn kill_processes(&mut self) { - let procs = self.process_tracker.list().await; - if procs.is_empty() { + let active_tools = self.agent.lock().await.active_tools.clone(); + let mut tools = active_tools.lock().unwrap(); + if tools.is_empty() { let _ = self .ui_tx - .send(UiMessage::Info("(no running processes)".into())); + .send(UiMessage::Info("(no running tool calls)".into())); } else { - for p in &procs { - let elapsed = p.started.elapsed(); + for entry in tools.drain(..) { + let elapsed = entry.started.elapsed(); let _ = self.ui_tx.send(UiMessage::Info(format!( - " killing pid {} ({:.0}s): {}", - p.pid, + " killing {} ({:.0}s): {}", + entry.name, elapsed.as_secs_f64(), - p.command + entry.detail ))); - self.process_tracker.kill(p.pid).await; + entry.handle.abort(); } - let _ = self.ui_tx.send(UiMessage::Info(format!( - "Killed {} process(es)", - procs.len() - ))); } } @@ -877,7 +875,6 @@ async fn run(cli: cli::CliArgs) -> Result<()> { // Keep a reference to the process tracker outside the agent lock // so Ctrl+K can kill processes even when the agent is busy. - let process_tracker = agent.lock().await.process_tracker.clone(); // Restore conversation from the append-only log { @@ -912,7 +909,6 @@ async fn run(cli: cli::CliArgs) -> Result<()> { let mut session = Session::new( agent, config, - process_tracker, ui_tx.clone(), turn_tx, ); @@ -994,7 +990,7 @@ async fn run(cli: cli::CliArgs) -> Result<()> { // Render tick — update periodic state _ = render_interval.tick() => { - let new_count = session.process_tracker.list().await.len() as u32; + let new_count = session.agent.lock().await.active_tools.lock().unwrap().len() as u32; if new_count != app.running_processes { app.running_processes = new_count; dirty = true; diff --git a/src/subconscious/api.rs b/src/subconscious/api.rs index 56c8b1d..82feb53 100644 --- a/src/subconscious/api.rs +++ b/src/subconscious/api.rs @@ -9,7 +9,7 @@ use crate::user::api::ApiClient; use crate::user::types::*; -use crate::agent::tools::{self as agent_tools, ProcessTracker, ToolOutput}; +use crate::agent::tools::{self as agent_tools}; use std::sync::OnceLock; @@ -55,7 +55,6 @@ pub async fn call_api_with_tools( .filter(|t| tools.iter().any(|w| w == &t.function.name)) .collect() }; - let tracker = ProcessTracker::new(); // Provenance tracks which agent:phase is making writes. // Updated between steps by the bail function via set_provenance(). let first_phase = phases.first().map(|s| s.as_str()).unwrap_or(""); @@ -175,7 +174,7 @@ pub async fn call_api_with_tools( }; let prov = provenance.borrow().clone(); - let output = match agent_tools::dispatch_shared(&call.function.name, &args, &tracker, Some(&prov)).await { + let output = match agent_tools::dispatch_shared(&call.function.name, &args, Some(&prov)).await { Some(out) => out, None => ToolOutput::error(format!("Unknown tool: {}", call.function.name)), };