mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
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 <[email protected]>
This commit is contained in:
Illia Polosukhin
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
1c9f9db420
commit
f3c85f57fc
+239
-12
@@ -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<SubmissionResult, Error> {
|
||||
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<Mutex<Session>>,
|
||||
thread_id: Uuid,
|
||||
) -> Result<SubmissionResult, Error> {
|
||||
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<Mutex<Session>>,
|
||||
thread_id: Uuid,
|
||||
) -> Result<SubmissionResult, Error> {
|
||||
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 <id> - Switch thread
|
||||
/resume <id> - Resume checkpoint
|
||||
|
||||
/heartbeat - Run heartbeat check now
|
||||
/summarize - Summarize current thread
|
||||
/suggest - Suggest next steps
|
||||
|
||||
/quit - Exit"#
|
||||
.to_string(),
|
||||
)),
|
||||
|
||||
@@ -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<String> = {
|
||||
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<Uuid> = 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<String>,
|
||||
|
||||
/// 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<Arc<dyn EmbeddingProvider>>,
|
||||
) -> 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<String>,
|
||||
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<String> = 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...");
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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),
|
||||
),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+66
-1
@@ -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<Arc<dyn ironclaw::workspace::EmbeddingProvider>> =
|
||||
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,
|
||||
|
||||
@@ -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(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user