From f3c85f57fc8176346b406ddaa5debefbdb0b025e Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Thu, 5 Feb 2026 19:28:13 -0800 Subject: [PATCH] Add proactivity features: memory CLI, session pruning, self-repair notifications, slash commands, status diagnostics, context warnings Closes the proactivity gap with six features: - Memory CLI (`ironclaw memory search/read/write/tree/status`) for direct workspace access without starting the full agent - Session pruning background task that evicts idle sessions (configurable TTL, default 7 days) - Self-repair notifications broadcast recovery results through channel manager instead of silent logging - `/heartbeat`, `/summarize`, `/suggest` slash commands for manual heartbeat trigger, thread summarization, and next-step suggestions - `ironclaw status` diagnostics command checking DB, session, secrets, embeddings, WASM tools, channels, heartbeat, and MCP servers - Context pressure warning that notifies users before auto-compaction fires Co-Authored-By: Claude Opus 4.6 --- .env.example | 8 +- src/agent/agent_loop.rs | 251 ++++++++++++++++++++++++++++++-- src/agent/session_manager.rs | 113 +++++++++++++++ src/agent/submission.rs | 42 ++++++ src/cli/memory.rs | 268 +++++++++++++++++++++++++++++++++++ src/cli/mod.rs | 13 ++ src/cli/status.rs | 193 +++++++++++++++++++++++++ src/config.rs | 12 ++ src/main.rs | 67 ++++++++- src/settings.rs | 10 ++ 10 files changed, 960 insertions(+), 17 deletions(-) create mode 100644 src/cli/memory.rs create mode 100644 src/cli/status.rs diff --git a/.env.example b/.env.example index c3a81008..1a80fc02 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,5 @@ # Database Configuration -DATABASE_URL=postgres://near_agent:password@localhost:5432/near_agent +DATABASE_URL=postgres://ironclaw:password@localhost:5432/ironclaw DATABASE_POOL_SIZE=10 # LLM Provider (NEAR AI) @@ -7,7 +7,7 @@ DATABASE_POOL_SIZE=10 # Session token is stored in ~/.near-agent/session.json and managed automatically. # On first run, the agent will open a browser for OAuth authentication. NEARAI_MODEL=claude-3-5-sonnet-20241022 -NEARAI_BASE_URL=https://api.near.ai +NEARAI_BASE_URL=https://cloud-api.near.ai NEARAI_AUTH_URL=https://private.near.ai # NEARAI_SESSION_PATH=~/.near-agent/session.json # optional, default shown @@ -28,7 +28,7 @@ HTTP_PORT=8080 HTTP_WEBHOOK_SECRET=your-webhook-secret # Agent Settings -AGENT_NAME=near-agent +AGENT_NAME=ironclaw AGENT_MAX_PARALLEL_JOBS=5 AGENT_JOB_TIMEOUT_SECS=3600 AGENT_STUCK_THRESHOLD_SECS=300 @@ -51,4 +51,4 @@ SAFETY_MAX_OUTPUT_LENGTH=100000 SAFETY_INJECTION_CHECK_ENABLED=true # Logging -RUST_LOG=near_agent=debug,tower_http=debug +RUST_LOG=ironclaw=debug,tower_http=debug diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index bdcc43b6..0ad965b5 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -32,13 +32,11 @@ use uuid::Uuid; use crate::agent::compaction::ContextCompactor; use crate::agent::context_monitor::ContextMonitor; use crate::agent::heartbeat::spawn_heartbeat; -use crate::agent::self_repair::DefaultSelfRepair; +use crate::agent::self_repair::{DefaultSelfRepair, RepairResult, SelfRepair}; use crate::agent::session::{PendingApproval, Session, ThreadState}; use crate::agent::session_manager::SessionManager; use crate::agent::submission::{Submission, SubmissionParser, SubmissionResult}; -use crate::agent::{ - HeartbeatConfig as AgentHeartbeatConfig, MessageIntent, RepairTask, Router, Scheduler, -}; +use crate::agent::{HeartbeatConfig as AgentHeartbeatConfig, MessageIntent, Router, Scheduler}; use crate::channels::{ChannelManager, IncomingMessage, OutgoingResponse, StatusUpdate}; use crate::config::{AgentConfig, HeartbeatConfig}; use crate::context::ContextManager; @@ -148,16 +146,98 @@ impl Agent { // Start channels let mut message_stream = self.channels.start_all().await?; - // Start self-repair task + // Start self-repair task with notification forwarding let repair = Arc::new(DefaultSelfRepair::new( self.context_manager.clone(), self.config.stuck_threshold, self.config.max_repair_attempts, )); - let repair_task = RepairTask::new(repair, self.config.repair_check_interval); - + let repair_interval = self.config.repair_check_interval; + let repair_channels = self.channels.clone(); let repair_handle = tokio::spawn(async move { - repair_task.run().await; + loop { + tokio::time::sleep(repair_interval).await; + + // Check stuck jobs + let stuck_jobs = repair.detect_stuck_jobs().await; + for job in stuck_jobs { + tracing::info!("Attempting to repair stuck job {}", job.job_id); + let result = repair.repair_stuck_job(&job).await; + let notification = match &result { + Ok(RepairResult::Success { message }) => { + tracing::info!("Repair succeeded: {}", message); + Some(format!( + "Job {} was stuck for {}s, recovery succeeded: {}", + job.job_id, + job.stuck_duration.as_secs(), + message + )) + } + Ok(RepairResult::Failed { message }) => { + tracing::error!("Repair failed: {}", message); + Some(format!( + "Job {} was stuck for {}s, recovery failed permanently: {}", + job.job_id, + job.stuck_duration.as_secs(), + message + )) + } + Ok(RepairResult::ManualRequired { message }) => { + tracing::warn!("Manual intervention needed: {}", message); + Some(format!( + "Job {} needs manual intervention: {}", + job.job_id, message + )) + } + Ok(RepairResult::Retry { message }) => { + tracing::warn!("Repair needs retry: {}", message); + None // Don't spam the user on retries + } + Err(e) => { + tracing::error!("Repair error: {}", e); + None + } + }; + + if let Some(msg) = notification { + let response = OutgoingResponse::text(format!("Self-Repair: {}", msg)); + let _ = repair_channels.broadcast_all("default", response).await; + } + } + + // Check broken tools + let broken_tools = repair.detect_broken_tools().await; + for tool in broken_tools { + tracing::info!("Attempting to repair broken tool: {}", tool.name); + match repair.repair_broken_tool(&tool).await { + Ok(RepairResult::Success { message }) => { + let response = OutgoingResponse::text(format!( + "Self-Repair: Tool '{}' repaired: {}", + tool.name, message + )); + let _ = repair_channels.broadcast_all("default", response).await; + } + Ok(result) => { + tracing::info!("Tool repair result: {:?}", result); + } + Err(e) => { + tracing::error!("Tool repair error: {}", e); + } + } + } + } + }); + + // Spawn session pruning task + let session_mgr = self.session_manager.clone(); + let session_idle_timeout = self.config.session_idle_timeout; + let pruning_handle = tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(600)); // Every 10 min + interval.tick().await; // Skip immediate first tick + loop { + interval.tick().await; + session_mgr.prune_stale_sessions(session_idle_timeout).await; + } }); // Spawn heartbeat if enabled @@ -272,6 +352,7 @@ impl Agent { // Cleanup tracing::info!("Agent shutting down..."); repair_handle.abort(); + pruning_handle.abort(); if let Some(handle) = heartbeat_handle { handle.abort(); } @@ -314,6 +395,9 @@ impl Agent { Submission::Compact => self.process_compact(session, thread_id).await, Submission::Clear => self.process_clear(session, thread_id).await, Submission::NewThread => self.process_new_thread(message).await, + Submission::Heartbeat => self.process_heartbeat().await, + Submission::Summarize => self.process_summarize(session, thread_id).await, + Submission::Suggest => self.process_suggest(session, thread_id).await, Submission::SwitchThread { thread_id: target } => { self.process_switch_thread(message, target).await } @@ -472,10 +556,21 @@ impl Agent { let messages = thread.messages(); if let Some(strategy) = self.context_monitor.suggest_compaction(&messages) { - tracing::info!( - "Context at {:.1}% capacity, auto-compacting", - self.context_monitor.usage_percent(&messages) - ); + let pct = self.context_monitor.usage_percent(&messages); + tracing::info!("Context at {:.1}% capacity, auto-compacting", pct); + + // Notify the user that compaction is happening + let _ = self + .channels + .send_status( + &message.channel, + StatusUpdate::Status(format!( + "Context at {:.0}% capacity, compacting...", + pct + )), + ) + .await; + let compactor = ContextCompactor::new(self.llm().clone()); if let Err(e) = compactor .compact(thread, strategy, self.workspace().map(|w| w.as_ref())) @@ -1427,6 +1522,134 @@ impl Agent { } } + /// Trigger a manual heartbeat check. + async fn process_heartbeat(&self) -> Result { + let Some(workspace) = self.workspace() else { + return Ok(SubmissionResult::error( + "Heartbeat requires a workspace (database must be connected).", + )); + }; + + let runner = crate::agent::HeartbeatRunner::new( + crate::agent::HeartbeatConfig::default(), + workspace.clone(), + self.llm().clone(), + ); + + match runner.check_heartbeat().await { + crate::agent::HeartbeatResult::Ok => Ok(SubmissionResult::ok_with_message( + "Heartbeat: all clear, nothing needs attention.", + )), + crate::agent::HeartbeatResult::NeedsAttention(msg) => Ok(SubmissionResult::response( + format!("Heartbeat findings:\n\n{}", msg), + )), + crate::agent::HeartbeatResult::Skipped => Ok(SubmissionResult::ok_with_message( + "Heartbeat skipped: no HEARTBEAT.md checklist found in workspace.", + )), + crate::agent::HeartbeatResult::Failed(err) => Ok(SubmissionResult::error(format!( + "Heartbeat failed: {}", + err + ))), + } + } + + /// Summarize the current thread's conversation. + async fn process_summarize( + &self, + session: Arc>, + thread_id: Uuid, + ) -> Result { + let messages = { + let sess = session.lock().await; + let thread = sess + .threads + .get(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + thread.messages() + }; + + if messages.is_empty() { + return Ok(SubmissionResult::ok_with_message( + "Nothing to summarize (empty thread).", + )); + } + + // Build a summary prompt with the conversation + let mut context = Vec::new(); + context.push(ChatMessage::system( + "Summarize the conversation so far in 3-5 concise bullet points. \ + Focus on decisions made, actions taken, and key outcomes. \ + Be brief and factual.", + )); + // Include the conversation messages (truncate to last 20 to avoid context overflow) + let start = if messages.len() > 20 { + messages.len() - 20 + } else { + 0 + }; + context.extend_from_slice(&messages[start..]); + context.push(ChatMessage::user("Summarize this conversation.")); + + let request = crate::llm::CompletionRequest::new(context) + .with_max_tokens(512) + .with_temperature(0.3); + + match self.llm().complete(request).await { + Ok(response) => Ok(SubmissionResult::response(format!( + "Thread Summary:\n\n{}", + response.content.trim() + ))), + Err(e) => Ok(SubmissionResult::error(format!("Summarize failed: {}", e))), + } + } + + /// Suggest next steps based on the current thread. + async fn process_suggest( + &self, + session: Arc>, + thread_id: Uuid, + ) -> Result { + let messages = { + let sess = session.lock().await; + let thread = sess + .threads + .get(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + thread.messages() + }; + + if messages.is_empty() { + return Ok(SubmissionResult::ok_with_message( + "Nothing to suggest from (empty thread).", + )); + } + + let mut context = Vec::new(); + context.push(ChatMessage::system( + "Based on the conversation so far, suggest 2-4 concrete next steps the user could take. \ + Be actionable and specific. Format as a numbered list.", + )); + let start = if messages.len() > 20 { + messages.len() - 20 + } else { + 0 + }; + context.extend_from_slice(&messages[start..]); + context.push(ChatMessage::user("What should I do next?")); + + let request = crate::llm::CompletionRequest::new(context) + .with_max_tokens(512) + .with_temperature(0.5); + + match self.llm().complete(request).await { + Ok(response) => Ok(SubmissionResult::response(format!( + "Suggested Next Steps:\n\n{}", + response.content.trim() + ))), + Err(e) => Ok(SubmissionResult::error(format!("Suggest failed: {}", e))), + } + } + async fn handle_command( &self, command: &str, @@ -1450,6 +1673,10 @@ impl Agent { /thread - Switch thread /resume - Resume checkpoint + /heartbeat - Run heartbeat check now + /summarize - Summarize current thread + /suggest - Suggest next steps + /quit - Exit"# .to_string(), )), diff --git a/src/agent/session_manager.rs b/src/agent/session_manager.rs index 69787c9e..dcf46343 100644 --- a/src/agent/session_manager.rs +++ b/src/agent/session_manager.rs @@ -131,6 +131,81 @@ impl SessionManager { managers.insert(thread_id, Arc::clone(&mgr)); mgr } + + /// Remove sessions that have been idle for longer than the given duration. + /// + /// Returns the number of sessions pruned. + pub async fn prune_stale_sessions(&self, max_idle: std::time::Duration) -> usize { + let cutoff = chrono::Utc::now() - chrono::TimeDelta::seconds(max_idle.as_secs() as i64); + + // Find stale session user_ids + let stale_users: Vec = { + let sessions = self.sessions.read().await; + sessions + .iter() + .filter_map(|(user_id, session)| { + // Try to lock; skip if contended (someone is actively using it) + let sess = session.try_lock().ok()?; + if sess.last_active_at < cutoff { + Some(user_id.clone()) + } else { + None + } + }) + .collect() + }; + + if stale_users.is_empty() { + return 0; + } + + // Collect thread IDs from stale sessions for cleanup + let mut stale_thread_ids: Vec = Vec::new(); + { + let sessions = self.sessions.read().await; + for user_id in &stale_users { + if let Some(session) = sessions.get(user_id) { + if let Ok(sess) = session.try_lock() { + stale_thread_ids.extend(sess.threads.keys()); + } + } + } + } + + // Remove sessions + let count = { + let mut sessions = self.sessions.write().await; + let before = sessions.len(); + for user_id in &stale_users { + sessions.remove(user_id); + } + before - sessions.len() + }; + + // Clean up thread mappings that point to stale sessions + { + let mut thread_map = self.thread_map.write().await; + thread_map.retain(|key, _| !stale_users.contains(&key.user_id)); + } + + // Clean up undo managers for stale threads + { + let mut undo_managers = self.undo_managers.write().await; + for thread_id in &stale_thread_ids { + undo_managers.remove(thread_id); + } + } + + if count > 0 { + tracing::info!( + "Pruned {} stale session(s) (idle > {}s)", + count, + max_idle.as_secs() + ); + } + + count + } } impl Default for SessionManager { @@ -183,4 +258,42 @@ mod tests { assert!(Arc::ptr_eq(&undo1, &undo2)); } + + #[tokio::test] + async fn test_prune_stale_sessions() { + let manager = SessionManager::new(); + + // Create two sessions and resolve threads (which updates last_active_at) + let (_, _thread_id) = manager.resolve_thread("user-active", "cli", None).await; + let (s2, _thread_id) = manager.resolve_thread("user-stale", "cli", None).await; + + // Backdate the stale session's last_active_at AFTER thread creation + { + let mut sess = s2.lock().await; + sess.last_active_at = chrono::Utc::now() - chrono::TimeDelta::seconds(86400 * 10); // 10 days ago + } + + // Prune with 7-day timeout + let pruned = manager + .prune_stale_sessions(std::time::Duration::from_secs(86400 * 7)) + .await; + assert_eq!(pruned, 1); + + // Active session should still exist + let sessions = manager.sessions.read().await; + assert!(sessions.contains_key("user-active")); + assert!(!sessions.contains_key("user-stale")); + } + + #[tokio::test] + async fn test_prune_no_stale_sessions() { + let manager = SessionManager::new(); + let _s1 = manager.get_or_create_session("user-1").await; + + // Nothing should be pruned when timeout is long + let pruned = manager + .prune_stale_sessions(std::time::Duration::from_secs(86400 * 365)) + .await; + assert_eq!(pruned, 0); + } } diff --git a/src/agent/submission.rs b/src/agent/submission.rs index f79c0b03..f1c36d68 100644 --- a/src/agent/submission.rs +++ b/src/agent/submission.rs @@ -31,6 +31,15 @@ impl SubmissionParser { if lower == "/clear" { return Submission::Clear; } + if lower == "/heartbeat" { + return Submission::Heartbeat; + } + if lower == "/summarize" || lower == "/summary" { + return Submission::Summarize; + } + if lower == "/suggest" { + return Submission::Suggest; + } if lower == "/thread new" || lower == "/new" { return Submission::NewThread; } @@ -139,6 +148,15 @@ pub enum Submission { /// Create a new thread. NewThread, + + /// Trigger a manual heartbeat check. + Heartbeat, + + /// Summarize the current thread. + Summarize, + + /// Suggest next steps based on the current thread. + Suggest, } impl Submission { @@ -202,6 +220,9 @@ impl Submission { | Self::Redo | Self::Clear | Self::NewThread + | Self::Heartbeat + | Self::Summarize + | Self::Suggest ) } } @@ -355,6 +376,27 @@ mod tests { ); } + #[test] + fn test_parser_heartbeat() { + let submission = SubmissionParser::parse("/heartbeat"); + assert!(matches!(submission, Submission::Heartbeat)); + } + + #[test] + fn test_parser_summarize() { + let submission = SubmissionParser::parse("/summarize"); + assert!(matches!(submission, Submission::Summarize)); + + let submission = SubmissionParser::parse("/summary"); + assert!(matches!(submission, Submission::Summarize)); + } + + #[test] + fn test_parser_suggest() { + let submission = SubmissionParser::parse("/suggest"); + assert!(matches!(submission, Submission::Suggest)); + } + #[test] fn test_parser_invalid_commands_become_user_input() { // Invalid UUID should become user input diff --git a/src/cli/memory.rs b/src/cli/memory.rs new file mode 100644 index 00000000..e10cf44f --- /dev/null +++ b/src/cli/memory.rs @@ -0,0 +1,268 @@ +//! Memory/workspace CLI commands. +//! +//! Exposes the workspace system for direct CLI use without starting the agent. + +use std::io::Read; +use std::sync::Arc; + +use clap::Subcommand; + +use crate::workspace::{EmbeddingProvider, SearchConfig, Workspace}; + +#[derive(Subcommand, Debug, Clone)] +pub enum MemoryCommand { + /// Search workspace memory (hybrid full-text + semantic) + Search { + /// Search query + query: String, + + /// Maximum number of results + #[arg(short, long, default_value = "5")] + limit: usize, + }, + + /// Read a file from the workspace + Read { + /// File path (e.g., "MEMORY.md", "daily/2024-01-15.md") + path: String, + }, + + /// Write content to a workspace file + Write { + /// File path (e.g., "notes/idea.md") + path: String, + + /// Content to write (omit to read from stdin) + content: Option, + + /// Append instead of overwrite + #[arg(short, long)] + append: bool, + }, + + /// Show workspace directory tree + Tree { + /// Root path to start from + #[arg(default_value = "")] + path: String, + + /// Maximum depth to traverse + #[arg(short, long, default_value = "3")] + depth: usize, + }, + + /// Show workspace status (document count, index health) + Status, +} + +/// Run a memory command. +pub async fn run_memory_command( + cmd: MemoryCommand, + pool: deadpool_postgres::Pool, + embeddings: Option>, +) -> anyhow::Result<()> { + let mut workspace = Workspace::new("default", pool); + if let Some(emb) = embeddings { + workspace = workspace.with_embeddings(emb); + } + + match cmd { + MemoryCommand::Search { query, limit } => search(&workspace, &query, limit).await, + MemoryCommand::Read { path } => read(&workspace, &path).await, + MemoryCommand::Write { + path, + content, + append, + } => write(&workspace, &path, content, append).await, + MemoryCommand::Tree { path, depth } => tree(&workspace, &path, depth).await, + MemoryCommand::Status => status(&workspace).await, + } +} + +async fn search(workspace: &Workspace, query: &str, limit: usize) -> anyhow::Result<()> { + let config = SearchConfig::default().with_limit(limit.min(50)); + let results = workspace.search_with_config(query, config).await?; + + if results.is_empty() { + println!("No results found for: {}", query); + return Ok(()); + } + + println!("Found {} result(s) for \"{}\":\n", results.len(), query); + + for (i, result) in results.iter().enumerate() { + let score_bar = score_indicator(result.score); + println!("{}. [{}] (score: {:.3})", i + 1, score_bar, result.score); + + // Show a content preview (first 200 chars) + let preview = truncate_content(&result.content, 200); + for line in preview.lines() { + println!(" {}", line); + } + println!(); + } + + Ok(()) +} + +async fn read(workspace: &Workspace, path: &str) -> anyhow::Result<()> { + match workspace.read(path).await { + Ok(doc) => { + println!("{}", doc.content); + } + Err(crate::error::WorkspaceError::DocumentNotFound { .. }) => { + anyhow::bail!("File not found: {}", path); + } + Err(e) => return Err(e.into()), + } + Ok(()) +} + +async fn write( + workspace: &Workspace, + path: &str, + content: Option, + append: bool, +) -> anyhow::Result<()> { + let content = match content { + Some(c) => c, + None => { + // Read from stdin + let mut buf = String::new(); + std::io::stdin().read_to_string(&mut buf)?; + buf + } + }; + + if append { + workspace.append(path, &content).await?; + println!("Appended to {}", path); + } else { + workspace.write(path, &content).await?; + println!("Wrote to {}", path); + } + + Ok(()) +} + +async fn tree(workspace: &Workspace, path: &str, max_depth: usize) -> anyhow::Result<()> { + let root = if path.is_empty() { "." } else { path }; + println!("{}/", root); + print_tree(workspace, path, "", max_depth, 0).await?; + Ok(()) +} + +async fn print_tree( + workspace: &Workspace, + path: &str, + prefix: &str, + max_depth: usize, + current_depth: usize, +) -> anyhow::Result<()> { + if current_depth >= max_depth { + return Ok(()); + } + + let entries = workspace.list(path).await?; + let count = entries.len(); + + for (i, entry) in entries.iter().enumerate() { + let is_last = i == count - 1; + let connector = if is_last { "└── " } else { "├── " }; + let child_prefix = if is_last { " " } else { "│ " }; + + if entry.is_directory { + println!("{}{}{}/", prefix, connector, entry.name()); + Box::pin(print_tree( + workspace, + &entry.path, + &format!("{}{}", prefix, child_prefix), + max_depth, + current_depth + 1, + )) + .await?; + } else { + println!("{}{}{}", prefix, connector, entry.name()); + } + } + + Ok(()) +} + +async fn status(workspace: &Workspace) -> anyhow::Result<()> { + let all_paths = workspace.list_all().await?; + let file_count = all_paths.len(); + + // Count directories by collecting unique parent paths + let mut dirs: std::collections::HashSet = std::collections::HashSet::new(); + for path in &all_paths { + if let Some(parent) = path.rsplit_once('/') { + dirs.insert(parent.0.to_string()); + } + } + + println!("Workspace Status"); + println!(" User: {}", workspace.user_id()); + println!(" Files: {}", file_count); + println!(" Directories: {}", dirs.len()); + + // Check key files + let key_files = [ + "MEMORY.md", + "HEARTBEAT.md", + "IDENTITY.md", + "SOUL.md", + "AGENTS.md", + "USER.md", + ]; + println!("\n Identity files:"); + for path in &key_files { + let exists = workspace.exists(path).await.unwrap_or(false); + let marker = if exists { "+" } else { "-" }; + println!(" [{}] {}", marker, path); + } + + Ok(()) +} + +fn truncate_content(s: &str, max_len: usize) -> String { + if s.len() <= max_len { + s.to_string() + } else { + format!("{}...", &s[..max_len]) + } +} + +fn score_indicator(score: f32) -> &'static str { + if score > 0.8_f32 { + "=====>" + } else if score > 0.5_f32 { + "====>" + } else if score > 0.3_f32 { + "===>" + } else if score > 0.1_f32 { + "==>" + } else { + "=>" + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_score_indicator() { + assert_eq!(score_indicator(0.9_f32), "=====>"); + assert_eq!(score_indicator(0.6_f32), "====>"); + assert_eq!(score_indicator(0.4_f32), "===>"); + assert_eq!(score_indicator(0.2_f32), "==>"); + assert_eq!(score_indicator(0.05_f32), "=>"); + } + + #[test] + fn test_truncate_content() { + assert_eq!(truncate_content("hello", 10), "hello"); + assert_eq!(truncate_content("hello world", 5), "hello..."); + } +} diff --git a/src/cli/mod.rs b/src/cli/mod.rs index 4d863059..fa3e9db1 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -6,13 +6,19 @@ //! - Managing configuration (`config list`, `config get`, `config set`) //! - Managing WASM tools (`tool install`, `tool list`, `tool remove`) //! - Managing MCP servers (`mcp add`, `mcp auth`, `mcp list`, `mcp test`) +//! - Querying workspace memory (`memory search`, `memory read`, `memory write`) +//! - Checking system health (`status`) mod config; mod mcp; +pub mod memory; +pub mod status; mod tool; pub use config::{ConfigCommand, run_config_command}; pub use mcp::{McpCommand, run_mcp_command}; +pub use memory::{MemoryCommand, run_memory_command}; +pub use status::run_status_command; pub use tool::{ToolCommand, run_tool_command}; use clap::{Parser, Subcommand}; @@ -79,6 +85,13 @@ pub enum Command { /// Manage MCP servers (hosted tool providers) #[command(subcommand)] Mcp(McpCommand), + + /// Query and manage workspace memory + #[command(subcommand)] + Memory(MemoryCommand), + + /// Show system health and diagnostics + Status, } impl Cli { diff --git a/src/cli/status.rs b/src/cli/status.rs new file mode 100644 index 00000000..0b4656c3 --- /dev/null +++ b/src/cli/status.rs @@ -0,0 +1,193 @@ +//! System health and diagnostics CLI command. +//! +//! Checks database connectivity, session validity, embeddings, +//! WASM runtime, tool count, and channel availability. + +use std::path::PathBuf; + +use crate::settings::Settings; + +/// Run the status command, printing system health info. +pub async fn run_status_command() -> anyhow::Result<()> { + let settings = Settings::load(); + + println!("IronClaw Status"); + println!("===============\n"); + + // Version + println!( + " Version: {} v{}", + env!("CARGO_PKG_NAME"), + env!("CARGO_PKG_VERSION") + ); + + // Database + let db_url_set = settings.database_url.is_some() || std::env::var("DATABASE_URL").is_ok(); + print!(" Database: "); + if db_url_set { + // Try to connect + match check_database().await { + Ok(()) => println!("connected"), + Err(e) => println!("error ({})", e), + } + } else { + println!("not configured"); + } + + // Session / Auth + print!(" Session: "); + let session_path = crate::llm::session::default_session_path(); + if session_path.exists() { + println!("found ({})", session_path.display()); + } else { + println!("not found (run `ironclaw setup`)"); + } + + // Secrets + print!(" Secrets: "); + let secrets_configured = settings.secrets_master_key_source != crate::settings::KeySource::None + || std::env::var("SECRETS_MASTER_KEY").is_ok() + || crate::secrets::keychain::has_master_key(); + if secrets_configured { + println!("configured ({:?})", settings.secrets_master_key_source); + } else { + println!("not configured"); + } + + // Embeddings + print!(" Embeddings: "); + let emb_enabled = settings.embeddings.enabled + || std::env::var("OPENAI_API_KEY").is_ok() + || std::env::var("EMBEDDING_ENABLED") + .map(|v| v == "true") + .unwrap_or(false); + if emb_enabled { + println!( + "enabled (provider: {}, model: {})", + settings.embeddings.provider, settings.embeddings.model + ); + } else { + println!("disabled"); + } + + // WASM tools + print!(" WASM Tools: "); + let tools_dir = settings + .wasm + .tools_dir + .clone() + .unwrap_or_else(default_tools_dir); + if tools_dir.exists() { + let count = count_wasm_files(&tools_dir); + println!("{} installed ({})", count, tools_dir.display()); + } else { + println!("directory not found ({})", tools_dir.display()); + } + + // WASM channels + print!(" Channels: "); + let channels_dir = settings + .channels + .wasm_channels_dir + .clone() + .unwrap_or_else(default_channels_dir); + let mut channel_info = vec!["cli".to_string()]; + if settings.channels.http_enabled { + channel_info.push(format!( + "http:{}", + settings.channels.http_port.unwrap_or(3000) + )); + } + if channels_dir.exists() { + let wasm_count = count_wasm_files(&channels_dir); + if wasm_count > 0 { + channel_info.push(format!("{} wasm", wasm_count)); + } + } + println!("{}", channel_info.join(", ")); + + // Heartbeat + print!(" Heartbeat: "); + let hb_enabled = settings.heartbeat.enabled + || std::env::var("HEARTBEAT_ENABLED") + .map(|v| v == "true") + .unwrap_or(false); + if hb_enabled { + println!("enabled (interval: {}s)", settings.heartbeat.interval_secs); + } else { + println!("disabled"); + } + + // MCP servers + print!(" MCP Servers: "); + match crate::tools::mcp::config::load_mcp_servers().await { + Ok(servers) => { + let enabled = servers.servers.iter().filter(|s| s.enabled).count(); + let total = servers.servers.len(); + println!("{} enabled / {} configured", enabled, total); + } + Err(_) => println!("none configured"), + } + + // Settings path + println!("\n Settings: {}", Settings::default_path().display()); + + Ok(()) +} + +async fn check_database() -> anyhow::Result<()> { + let _ = dotenvy::dotenv(); + let settings = Settings::load(); + let url = std::env::var("DATABASE_URL") + .ok() + .or(settings.database_url) + .ok_or_else(|| anyhow::anyhow!("no URL"))?; + + let config: deadpool_postgres::Config = deadpool_postgres::Config { + url: Some(url), + ..Default::default() + }; + let pool = config + .create_pool( + Some(deadpool_postgres::Runtime::Tokio1), + tokio_postgres::NoTls, + ) + .map_err(|e| anyhow::anyhow!("pool error: {}", e))?; + + let client = tokio::time::timeout(std::time::Duration::from_secs(5), pool.get()) + .await + .map_err(|_| anyhow::anyhow!("timeout"))? + .map_err(|e| anyhow::anyhow!("{}", e))?; + + client + .execute("SELECT 1", &[]) + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; + + Ok(()) +} + +fn count_wasm_files(dir: &std::path::Path) -> usize { + std::fs::read_dir(dir) + .map(|entries| { + entries + .filter_map(|e| e.ok()) + .filter(|e| e.path().extension().is_some_and(|ext| ext == "wasm")) + .count() + }) + .unwrap_or(0) +} + +fn default_tools_dir() -> PathBuf { + dirs::home_dir() + .unwrap_or_else(|| PathBuf::from(".")) + .join(".ironclaw") + .join("tools") +} + +fn default_channels_dir() -> PathBuf { + dirs::home_dir() + .unwrap_or_else(|| PathBuf::from(".")) + .join(".ironclaw") + .join("channels") +} diff --git a/src/config.rs b/src/config.rs index 797d058c..cc2a7225 100644 --- a/src/config.rs +++ b/src/config.rs @@ -415,6 +415,8 @@ pub struct AgentConfig { pub max_repair_attempts: u32, /// Whether to use planning before tool execution. pub use_planning: bool, + /// Session idle timeout. Sessions inactive longer than this are pruned. + pub session_idle_timeout: Duration, } impl AgentConfig { @@ -478,6 +480,16 @@ impl AgentConfig { message: format!("must be 'true' or 'false': {e}"), })? .unwrap_or(settings.agent.use_planning), + session_idle_timeout: Duration::from_secs( + optional_env("SESSION_IDLE_TIMEOUT_SECS")? + .map(|s| s.parse()) + .transpose() + .map_err(|e| ConfigError::InvalidValue { + key: "SESSION_IDLE_TIMEOUT_SECS".to_string(), + message: format!("must be a positive integer: {e}"), + })? + .unwrap_or(settings.agent.session_idle_timeout_secs), + ), }) } } diff --git a/src/main.rs b/src/main.rs index 0261fad2..c93f6477 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,7 +14,9 @@ use ironclaw::{ WasmChannelRuntime, WasmChannelRuntimeConfig, WasmChannelServer, }, }, - cli::{Cli, Command, run_mcp_command, run_tool_command}, + cli::{ + Cli, Command, run_mcp_command, run_memory_command, run_status_command, run_tool_command, + }, config::Config, context::ContextManager, history::Store, @@ -62,6 +64,69 @@ async fn main() -> anyhow::Result<()> { return run_mcp_command(mcp_cmd.clone()).await; } + Some(Command::Memory(mem_cmd)) => { + tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("warn")), + ) + .init(); + + // Memory commands need database (and optionally embeddings) + let _ = dotenvy::dotenv(); + let config = Config::from_env().map_err(|e| anyhow::anyhow!("{}", e))?; + let store = ironclaw::history::Store::new(&config.database).await?; + store.run_migrations().await?; + + // Set up embeddings if available + let session = ironclaw::llm::create_session_manager(ironclaw::llm::SessionConfig { + auth_base_url: config.llm.nearai.auth_base_url.clone(), + session_path: config.llm.nearai.session_path.clone(), + ..Default::default() + }) + .await; + + let embeddings: Option> = + if config.embeddings.enabled { + match config.embeddings.provider.as_str() { + "nearai" => Some(Arc::new( + ironclaw::workspace::NearAiEmbeddings::new( + &config.llm.nearai.base_url, + session, + ) + .with_model(&config.embeddings.model, 1536), + )), + _ => { + if let Some(api_key) = config.embeddings.openai_api_key() { + let dim = match config.embeddings.model.as_str() { + "text-embedding-3-large" => 3072, + _ => 1536, + }; + Some(Arc::new(ironclaw::workspace::OpenAiEmbeddings::with_model( + api_key, + &config.embeddings.model, + dim, + ))) + } else { + None + } + } + } + } else { + None + }; + + return run_memory_command(mem_cmd.clone(), store.pool(), embeddings).await; + } + Some(Command::Status) => { + let _ = dotenvy::dotenv(); + tracing_subscriber::fmt() + .with_env_filter( + EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("warn")), + ) + .init(); + + return run_status_command().await; + } Some(Command::Setup { skip_auth, channels_only, diff --git a/src/settings.rs b/src/settings.rs index bb594a19..adad7d3b 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -229,6 +229,11 @@ pub struct AgentSettings { /// Maximum repair attempts. #[serde(default = "default_max_repair_attempts")] pub max_repair_attempts: u32, + + /// Session idle timeout in seconds (default: 7 days). Sessions inactive + /// longer than this are pruned from memory. + #[serde(default = "default_session_idle_timeout")] + pub session_idle_timeout_secs: u64, } fn default_agent_name() -> String { @@ -251,6 +256,10 @@ fn default_repair_interval() -> u64 { 60 // 1 minute } +fn default_session_idle_timeout() -> u64 { + 7 * 24 * 3600 // 7 days +} + fn default_max_repair_attempts() -> u32 { 3 } @@ -269,6 +278,7 @@ impl Default for AgentSettings { use_planning: true, repair_check_interval_secs: default_repair_interval(), max_repair_attempts: default_max_repair_attempts(), + session_idle_timeout_secs: default_session_idle_timeout(), } } }