diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index f816fd3a..1e2cee9e 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -3,18 +3,25 @@ use std::sync::Arc; use futures::StreamExt; +use tokio::sync::Mutex; use uuid::Uuid; +use crate::agent::compaction::ContextCompactor; +use crate::agent::context_monitor::ContextMonitor; use crate::agent::self_repair::DefaultSelfRepair; +use crate::agent::session::{Session, ThreadState}; +use crate::agent::session_manager::SessionManager; +use crate::agent::submission::{Submission, SubmissionParser, SubmissionResult}; use crate::agent::{MessageIntent, RepairTask, Router, Scheduler}; use crate::channels::{ChannelManager, IncomingMessage, OutgoingResponse, StatusUpdate}; use crate::config::AgentConfig; use crate::context::ContextManager; use crate::error::Error; use crate::history::Store; -use crate::llm::{ChatMessage, LlmProvider, Reasoning, ReasoningContext}; +use crate::llm::{LlmProvider, Reasoning, ReasoningContext}; use crate::safety::SafetyLayer; use crate::tools::ToolRegistry; +use crate::workspace::Workspace; /// The main agent that coordinates all components. pub struct Agent { @@ -27,6 +34,9 @@ pub struct Agent { context_manager: Arc, scheduler: Arc, router: Router, + session_manager: Arc, + context_monitor: ContextMonitor, + workspace: Option>, } impl Agent { @@ -38,6 +48,7 @@ impl Agent { safety: Arc, tools: Arc, channels: ChannelManager, + workspace: Option>, ) -> Self { let context_manager = Arc::new(ContextManager::new(config.max_parallel_jobs)); @@ -60,6 +71,9 @@ impl Agent { context_manager, scheduler, router: Router::new(), + session_manager: Arc::new(SessionManager::new()), + context_monitor: ContextMonitor::new(), + workspace, } } @@ -123,54 +137,457 @@ impl Agent { truncate(&message.content, 100) ); - // Route the message - let intent = self.router.route(message); - tracing::debug!("Routed to intent: {:?}", intent); + // Parse submission type first + let submission = SubmissionParser::parse(&message.content); - // Send thinking status for non-trivial operations + // Resolve session and thread + let (session, thread_id) = self + .session_manager + .resolve_thread( + &message.user_id, + &message.channel, + message.thread_id.as_deref(), + ) + .await; + + // Process based on submission type + let result = match submission { + Submission::UserInput { content } => { + self.process_user_input(message, session, thread_id, &content) + .await + } + Submission::Undo => self.process_undo(session, thread_id).await, + Submission::Redo => self.process_redo(session, thread_id).await, + Submission::Interrupt => self.process_interrupt(session, thread_id).await, + 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::SwitchThread { thread_id: target } => { + self.process_switch_thread(message, target).await + } + Submission::Resume { checkpoint_id } => { + self.process_resume(session, thread_id, checkpoint_id).await + } + Submission::ExecApproval { .. } => { + // Not supported in simple chat flow + Ok(SubmissionResult::error( + "Approval flow not supported in this context", + )) + } + }; + + // Convert SubmissionResult to response string + match result? { + SubmissionResult::Response { content } => Ok(Some(content)), + SubmissionResult::Ok { message } => Ok(message), + SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))), + SubmissionResult::Interrupted => Ok(Some("Interrupted.".into())), + SubmissionResult::NeedApproval { .. } => { + Ok(Some("Approval required but not supported.".into())) + } + } + } + + async fn process_user_input( + &self, + message: &IncomingMessage, + session: Arc>, + thread_id: Uuid, + content: &str, + ) -> Result { + // First check thread state without holding lock during I/O + let thread_state = { + 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.state + }; + + // Check thread state + match thread_state { + ThreadState::Processing => { + return Ok(SubmissionResult::error( + "Turn in progress. Use /interrupt to cancel.", + )); + } + ThreadState::AwaitingApproval => { + return Ok(SubmissionResult::error( + "Waiting for approval. Use /interrupt to cancel.", + )); + } + ThreadState::Completed => { + return Ok(SubmissionResult::error( + "Thread completed. Use /thread new.", + )); + } + ThreadState::Idle | ThreadState::Interrupted => { + // Can proceed + } + } + + // Route for job commands (bypass turn system) + // Build a temporary message with the content to route + let temp_message = IncomingMessage { + content: content.to_string(), + ..message.clone() + }; + let intent = self.router.route(&temp_message); match &intent { - MessageIntent::Chat { .. } | MessageIntent::CreateJob { .. } => { - let _ = self - .channels - .send_status( - &message.channel, - StatusUpdate::Thinking("Processing...".into()), - ) - .await; + MessageIntent::CreateJob { .. } + | MessageIntent::CheckJobStatus { .. } + | MessageIntent::CancelJob { .. } + | MessageIntent::ListJobs { .. } + | MessageIntent::HelpJob { .. } + | MessageIntent::Command { .. } => { + return self.handle_job_or_command(intent, message).await; } _ => {} } - // Handle based on intent + // Auto-compact if needed BEFORE adding new turn + { + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + + 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 compactor = ContextCompactor::new(self.llm.clone()); + if let Err(e) = compactor + .compact(thread, strategy, self.workspace.as_deref()) + .await + { + tracing::warn!("Auto-compaction failed: {}", e); + } + } + } + + // Create checkpoint before turn + let undo_mgr = self.session_manager.get_undo_manager(thread_id).await; + { + let sess = session.lock().await; + let thread = sess + .threads + .get(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + + let mut mgr = undo_mgr.lock().await; + mgr.checkpoint( + thread.turn_number(), + thread.messages(), + format!("Before turn {}", thread.turn_number()), + ); + } + + // Start the turn and get messages + let turn_messages = { + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + thread.start_turn(content); + thread.messages() + }; + + // Send thinking status + let _ = self + .channels + .send_status( + &message.channel, + StatusUpdate::Thinking("Processing...".into()), + ) + .await; + + // Call LLM with thread context + let reasoning = Reasoning::new(self.llm.clone(), self.safety.clone()); + let context = ReasoningContext::new().with_messages(turn_messages); + let llm_result = reasoning.respond(&context).await; + + // Re-acquire lock and check if interrupted + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + + if thread.state == ThreadState::Interrupted { + let _ = self + .channels + .send_status(&message.channel, StatusUpdate::Status("Interrupted".into())) + .await; + return Ok(SubmissionResult::Interrupted); + } + + // Complete or fail the turn + match llm_result { + Ok(response) => { + thread.complete_turn(&response); + let _ = self + .channels + .send_status(&message.channel, StatusUpdate::Status("Done".into())) + .await; + Ok(SubmissionResult::response(response)) + } + Err(e) => { + thread.fail_turn(e.to_string()); + Ok(SubmissionResult::error(e.to_string())) + } + } + } + + /// Handle job-related intents without turn tracking. + async fn handle_job_or_command( + &self, + intent: MessageIntent, + message: &IncomingMessage, + ) -> Result { + // Send thinking status for non-trivial operations + if let MessageIntent::CreateJob { .. } = &intent { + let _ = self + .channels + .send_status( + &message.channel, + StatusUpdate::Thinking("Processing...".into()), + ) + .await; + } + let response = match intent { MessageIntent::CreateJob { title, description, category, - } => Some(self.handle_create_job(title, description, category).await?), - - MessageIntent::CheckJobStatus { job_id } => { - Some(self.handle_check_status(job_id).await?) - } - - MessageIntent::CancelJob { job_id } => Some(self.handle_cancel_job(&job_id).await?), - - MessageIntent::ListJobs { filter } => Some(self.handle_list_jobs(filter).await?), - - MessageIntent::HelpJob { job_id } => Some(self.handle_help_job(&job_id).await?), - - MessageIntent::Chat { content } => Some(self.handle_chat(message, &content).await?), - + } => self.handle_create_job(title, description, category).await?, + MessageIntent::CheckJobStatus { job_id } => self.handle_check_status(job_id).await?, + MessageIntent::CancelJob { job_id } => self.handle_cancel_job(&job_id).await?, + MessageIntent::ListJobs { filter } => self.handle_list_jobs(filter).await?, + MessageIntent::HelpJob { job_id } => self.handle_help_job(&job_id).await?, MessageIntent::Command { command, args } => { - self.handle_command(&command, &args).await? + match self.handle_command(&command, &args).await? { + Some(s) => s, + None => return Ok(SubmissionResult::Ok { message: None }), // Shutdown signal + } } - - MessageIntent::Unknown => Some( - "I'm not sure what you're asking. Try '/help' for available commands.".to_string(), - ), + _ => "Unknown intent".to_string(), }; + Ok(SubmissionResult::response(response)) + } - Ok(response) + async fn process_undo( + &self, + session: Arc>, + thread_id: Uuid, + ) -> Result { + let undo_mgr = self.session_manager.get_undo_manager(thread_id).await; + let mut mgr = undo_mgr.lock().await; + + if !mgr.can_undo() { + return Ok(SubmissionResult::ok_with_message("Nothing to undo.")); + } + + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + + // Save current state to redo, get previous checkpoint + let current_messages = thread.messages(); + let current_turn = thread.turn_number(); + + if let Some(checkpoint) = mgr.undo(current_turn, current_messages) { + // Extract values before consuming the reference + let turn_number = checkpoint.turn_number; + let messages = checkpoint.messages.clone(); + let undo_count = mgr.undo_count(); + // Restore thread from checkpoint + thread.restore_from_messages(messages); + Ok(SubmissionResult::ok_with_message(format!( + "Undone to turn {}. {} undo(s) remaining.", + turn_number, undo_count + ))) + } else { + Ok(SubmissionResult::error("Undo failed.")) + } + } + + async fn process_redo( + &self, + session: Arc>, + thread_id: Uuid, + ) -> Result { + let undo_mgr = self.session_manager.get_undo_manager(thread_id).await; + let mut mgr = undo_mgr.lock().await; + + if !mgr.can_redo() { + return Ok(SubmissionResult::ok_with_message("Nothing to redo.")); + } + + if let Some(checkpoint) = mgr.redo() { + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + thread.restore_from_messages(checkpoint.messages); + Ok(SubmissionResult::ok_with_message(format!( + "Redone to turn {}.", + checkpoint.turn_number + ))) + } else { + Ok(SubmissionResult::error("Redo failed.")) + } + } + + async fn process_interrupt( + &self, + session: Arc>, + thread_id: Uuid, + ) -> Result { + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + + match thread.state { + ThreadState::Processing | ThreadState::AwaitingApproval => { + thread.interrupt(); + Ok(SubmissionResult::ok_with_message("Interrupted.")) + } + _ => Ok(SubmissionResult::ok_with_message("Nothing to interrupt.")), + } + } + + async fn process_compact( + &self, + session: Arc>, + thread_id: Uuid, + ) -> Result { + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + + let messages = thread.messages(); + let usage = self.context_monitor.usage_percent(&messages); + let strategy = self + .context_monitor + .suggest_compaction(&messages) + .unwrap_or( + crate::agent::context_monitor::CompactionStrategy::Summarize { keep_recent: 5 }, + ); + + let compactor = ContextCompactor::new(self.llm.clone()); + match compactor + .compact(thread, strategy, self.workspace.as_deref()) + .await + { + Ok(result) => { + let mut msg = format!( + "Compacted: {} turns removed, {} → {} tokens (was {:.1}% full)", + result.turns_removed, result.tokens_before, result.tokens_after, usage + ); + if result.summary_written { + msg.push_str(", summary saved to workspace"); + } + Ok(SubmissionResult::ok_with_message(msg)) + } + Err(e) => Ok(SubmissionResult::error(format!("Compaction failed: {}", e))), + } + } + + async fn process_clear( + &self, + session: Arc>, + thread_id: Uuid, + ) -> Result { + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + thread.turns.clear(); + thread.state = ThreadState::Idle; + + // Clear undo history too + let undo_mgr = self.session_manager.get_undo_manager(thread_id).await; + undo_mgr.lock().await.clear(); + + Ok(SubmissionResult::ok_with_message("Thread cleared.")) + } + + async fn process_new_thread( + &self, + message: &IncomingMessage, + ) -> Result { + let session = self + .session_manager + .get_or_create_session(&message.user_id) + .await; + let mut sess = session.lock().await; + let thread = sess.create_thread(); + let thread_id = thread.id; + Ok(SubmissionResult::ok_with_message(format!( + "New thread: {}", + thread_id + ))) + } + + async fn process_switch_thread( + &self, + message: &IncomingMessage, + target_thread_id: Uuid, + ) -> Result { + let session = self + .session_manager + .get_or_create_session(&message.user_id) + .await; + let mut sess = session.lock().await; + + if sess.switch_thread(target_thread_id) { + Ok(SubmissionResult::ok_with_message(format!( + "Switched to thread {}", + target_thread_id + ))) + } else { + Ok(SubmissionResult::error("Thread not found.")) + } + } + + async fn process_resume( + &self, + session: Arc>, + thread_id: Uuid, + checkpoint_id: Uuid, + ) -> Result { + let undo_mgr = self.session_manager.get_undo_manager(thread_id).await; + let mut mgr = undo_mgr.lock().await; + + if let Some(checkpoint) = mgr.restore(checkpoint_id) { + let mut sess = session.lock().await; + let thread = sess + .threads + .get_mut(&thread_id) + .ok_or_else(|| Error::from(crate::error::JobError::NotFound { id: thread_id }))?; + thread.restore_from_messages(checkpoint.messages); + Ok(SubmissionResult::ok_with_message(format!( + "Resumed from checkpoint: {}", + checkpoint.description + ))) + } else { + Ok(SubmissionResult::error("Checkpoint not found.")) + } } async fn handle_create_job( @@ -307,42 +724,6 @@ impl Agent { } } - async fn handle_chat(&self, message: &IncomingMessage, content: &str) -> Result { - // Send thinking status - let _ = self - .channels - .send_status( - &message.channel, - StatusUpdate::Thinking("Generating response...".into()), - ) - .await; - - // Use LLM for general chat - let reasoning = Reasoning::new(self.llm.clone(), self.safety.clone()); - - let context = ReasoningContext::new().with_message(ChatMessage::user(content)); - - match reasoning.respond(&context).await { - Ok(response) => { - let _ = self - .channels - .send_status(&message.channel, StatusUpdate::Status("Done".into())) - .await; - Ok(response) - } - Err(e) => { - let _ = self - .channels - .send_status( - &message.channel, - StatusUpdate::Status(format!("Error: {}", e)), - ) - .await; - Err(e.into()) - } - } - } - async fn handle_command( &self, command: &str, @@ -350,15 +731,23 @@ impl Agent { ) -> Result, Error> { match command { "help" => Ok(Some( - r#"Available commands: - /job - Create a new job - /status [job_id] - Check job status - /cancel - Cancel a job - /list - List all jobs - /help - Help a stuck job - /quit - Exit the agent + r#"Commands: + /job - Create a job + /status [id] - Check job status + /cancel - Cancel a job + /list - List all jobs + /help - Help a stuck job -Or just chat naturally and I'll try to understand what you need!"# + /undo - Undo last turn + /redo - Redo undone turn + /compact - Compress context + /clear - Clear thread + /interrupt - Stop current turn + /thread new - New thread + /thread - Switch thread + /resume - Resume checkpoint + + /quit - Exit"# .to_string(), )), diff --git a/src/agent/mod.rs b/src/agent/mod.rs index 0875a6ff..abe4fdf4 100644 --- a/src/agent/mod.rs +++ b/src/agent/mod.rs @@ -17,6 +17,7 @@ mod router; mod scheduler; mod self_repair; pub mod session; +mod session_manager; pub mod submission; pub mod task; pub mod undo; @@ -30,7 +31,8 @@ pub use router::{MessageIntent, Router}; pub use scheduler::Scheduler; pub use self_repair::{RepairResult, RepairTask, SelfRepair, StuckJob}; pub use session::{Session, Thread, ThreadState, Turn, TurnState}; -pub use submission::{Submission, SubmissionResult}; +pub use session_manager::SessionManager; +pub use submission::{Submission, SubmissionParser, SubmissionResult}; pub use task::{Task, TaskContext, TaskHandler, TaskOutput, TaskStatus}; pub use undo::{Checkpoint, UndoManager}; pub use worker::Worker; diff --git a/src/agent/session.rs b/src/agent/session.rs index 615f803d..70846be1 100644 --- a/src/agent/session.rs +++ b/src/agent/session.rs @@ -230,6 +230,38 @@ impl Thread { } } } + + /// Restore thread state from a checkpoint's messages. + /// + /// Clears existing turns and rebuilds from message pairs. + /// Messages should alternate: user, assistant, user, assistant... + pub fn restore_from_messages(&mut self, messages: Vec) { + self.turns.clear(); + self.state = ThreadState::Idle; + + // Messages alternate: user, assistant, user, assistant... + let mut iter = messages.into_iter().peekable(); + let mut turn_number = 0; + + while let Some(msg) = iter.next() { + if msg.role == crate::llm::Role::User { + let mut turn = Turn::new(turn_number, &msg.content); + + // Check if next is assistant response + if let Some(next) = iter.peek() { + if next.role == crate::llm::Role::Assistant { + let response = iter.next().expect("peeked"); + turn.complete(&response.content); + } + } + + self.turns.push(turn); + turn_number += 1; + } + } + + self.updated_at = Utc::now(); + } } /// State of a turn. @@ -387,4 +419,48 @@ mod tests { assert_eq!(turn.tool_calls.len(), 1); assert!(turn.tool_calls[0].result.is_some()); } + + #[test] + fn test_restore_from_messages() { + let mut thread = Thread::new(Uuid::new_v4()); + + // First add some turns + thread.start_turn("Original message"); + thread.complete_turn("Original response"); + + // Now restore from different messages + let messages = vec![ + ChatMessage::user("Hello"), + ChatMessage::assistant("Hi there!"), + ChatMessage::user("How are you?"), + ChatMessage::assistant("I'm good!"), + ]; + + thread.restore_from_messages(messages); + + assert_eq!(thread.turns.len(), 2); + assert_eq!(thread.turns[0].user_input, "Hello"); + assert_eq!(thread.turns[0].response, Some("Hi there!".to_string())); + assert_eq!(thread.turns[1].user_input, "How are you?"); + assert_eq!(thread.turns[1].response, Some("I'm good!".to_string())); + assert_eq!(thread.state, ThreadState::Idle); + } + + #[test] + fn test_restore_from_messages_incomplete_turn() { + let mut thread = Thread::new(Uuid::new_v4()); + + // Messages with incomplete last turn (no assistant response) + let messages = vec![ + ChatMessage::user("Hello"), + ChatMessage::assistant("Hi there!"), + ChatMessage::user("How are you?"), + ]; + + thread.restore_from_messages(messages); + + assert_eq!(thread.turns.len(), 2); + assert_eq!(thread.turns[1].user_input, "How are you?"); + assert!(thread.turns[1].response.is_none()); + } } diff --git a/src/agent/session_manager.rs b/src/agent/session_manager.rs new file mode 100644 index 00000000..69787c9e --- /dev/null +++ b/src/agent/session_manager.rs @@ -0,0 +1,186 @@ +//! Session manager for multi-user, multi-thread conversation handling. +//! +//! Maps external channel thread IDs to internal UUIDs and manages undo state +//! for each thread. + +use std::collections::HashMap; +use std::sync::Arc; + +use tokio::sync::{Mutex, RwLock}; +use uuid::Uuid; + +use crate::agent::session::Session; +use crate::agent::undo::UndoManager; + +/// Key for mapping external thread IDs to internal ones. +#[derive(Clone, Hash, Eq, PartialEq)] +struct ThreadKey { + user_id: String, + channel: String, + external_thread_id: Option, +} + +/// Manages sessions, threads, and undo state for all users. +pub struct SessionManager { + sessions: RwLock>>>, + thread_map: RwLock>, + undo_managers: RwLock>>>, +} + +impl SessionManager { + /// Create a new session manager. + pub fn new() -> Self { + Self { + sessions: RwLock::new(HashMap::new()), + thread_map: RwLock::new(HashMap::new()), + undo_managers: RwLock::new(HashMap::new()), + } + } + + /// Get or create a session for a user. + pub async fn get_or_create_session(&self, user_id: &str) -> Arc> { + // Fast path: check if session exists + { + let sessions = self.sessions.read().await; + if let Some(session) = sessions.get(user_id) { + return Arc::clone(session); + } + } + + // Slow path: create new session + let mut sessions = self.sessions.write().await; + // Double-check after acquiring write lock + if let Some(session) = sessions.get(user_id) { + return Arc::clone(session); + } + + let session = Arc::new(Mutex::new(Session::new(user_id))); + sessions.insert(user_id.to_string(), Arc::clone(&session)); + session + } + + /// Resolve an external thread ID to an internal thread. + /// + /// Returns the session and thread ID. Creates both if they don't exist. + pub async fn resolve_thread( + &self, + user_id: &str, + channel: &str, + external_thread_id: Option<&str>, + ) -> (Arc>, Uuid) { + let session = self.get_or_create_session(user_id).await; + + let key = ThreadKey { + user_id: user_id.to_string(), + channel: channel.to_string(), + external_thread_id: external_thread_id.map(String::from), + }; + + // Check if we have a mapping + { + let thread_map = self.thread_map.read().await; + if let Some(&thread_id) = thread_map.get(&key) { + // Verify thread still exists in session + let sess = session.lock().await; + if sess.threads.contains_key(&thread_id) { + return (Arc::clone(&session), thread_id); + } + } + } + + // Create new thread (always create a new one for a new key) + let thread_id = { + let mut sess = session.lock().await; + let thread = sess.create_thread(); + thread.id + }; + + // Store mapping + { + let mut thread_map = self.thread_map.write().await; + thread_map.insert(key, thread_id); + } + + // Create undo manager for thread + { + let mut undo_managers = self.undo_managers.write().await; + undo_managers.insert(thread_id, Arc::new(Mutex::new(UndoManager::new()))); + } + + (session, thread_id) + } + + /// Get undo manager for a thread. + pub async fn get_undo_manager(&self, thread_id: Uuid) -> Arc> { + // Fast path + { + let managers = self.undo_managers.read().await; + if let Some(mgr) = managers.get(&thread_id) { + return Arc::clone(mgr); + } + } + + // Create if missing + let mut managers = self.undo_managers.write().await; + // Double-check + if let Some(mgr) = managers.get(&thread_id) { + return Arc::clone(mgr); + } + + let mgr = Arc::new(Mutex::new(UndoManager::new())); + managers.insert(thread_id, Arc::clone(&mgr)); + mgr + } +} + +impl Default for SessionManager { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_get_or_create_session() { + let manager = SessionManager::new(); + + let session1 = manager.get_or_create_session("user-1").await; + let session2 = manager.get_or_create_session("user-1").await; + + // Same user should get same session + assert!(Arc::ptr_eq(&session1, &session2)); + + let session3 = manager.get_or_create_session("user-2").await; + assert!(!Arc::ptr_eq(&session1, &session3)); + } + + #[tokio::test] + async fn test_resolve_thread() { + let manager = SessionManager::new(); + + let (session1, thread1) = manager.resolve_thread("user-1", "cli", None).await; + let (session2, thread2) = manager.resolve_thread("user-1", "cli", None).await; + + // Same channel+user should get same thread + assert!(Arc::ptr_eq(&session1, &session2)); + assert_eq!(thread1, thread2); + + // Different channel should get different thread + let (_, thread3) = manager.resolve_thread("user-1", "http", None).await; + assert_ne!(thread1, thread3); + } + + #[tokio::test] + async fn test_undo_manager() { + let manager = SessionManager::new(); + let (_, thread_id) = manager.resolve_thread("user-1", "cli", None).await; + + let undo1 = manager.get_undo_manager(thread_id).await; + let undo2 = manager.get_undo_manager(thread_id).await; + + assert!(Arc::ptr_eq(&undo1, &undo2)); + } +} diff --git a/src/agent/submission.rs b/src/agent/submission.rs index a35057ea..15598eae 100644 --- a/src/agent/submission.rs +++ b/src/agent/submission.rs @@ -6,6 +6,59 @@ use serde::{Deserialize, Serialize}; use uuid::Uuid; +/// Parses user input into Submission types. +pub struct SubmissionParser; + +impl SubmissionParser { + /// Parse message content into a Submission. + pub fn parse(content: &str) -> Submission { + let trimmed = content.trim(); + let lower = trimmed.to_lowercase(); + + // Control commands (exact match or prefix) + if lower == "/undo" { + return Submission::Undo; + } + if lower == "/redo" { + return Submission::Redo; + } + if lower == "/interrupt" || lower == "/stop" { + return Submission::Interrupt; + } + if lower == "/compact" { + return Submission::Compact; + } + if lower == "/clear" { + return Submission::Clear; + } + if lower == "/thread new" || lower == "/new" { + return Submission::NewThread; + } + + // /thread - switch thread + if let Some(rest) = lower.strip_prefix("/thread ") { + let rest = rest.trim(); + if rest != "new" { + if let Ok(id) = Uuid::parse_str(rest) { + return Submission::SwitchThread { thread_id: id }; + } + } + } + + // /resume - resume from checkpoint + if let Some(rest) = lower.strip_prefix("/resume ") { + if let Ok(id) = Uuid::parse_str(rest.trim()) { + return Submission::Resume { checkpoint_id: id }; + } + } + + // Default: user input + Submission::UserInput { + content: content.to_string(), + } + } +} + /// A submission to the agent. #[derive(Debug, Clone, Serialize, Deserialize)] pub enum Submission { @@ -200,4 +253,84 @@ mod tests { assert!(!undo.starts_turn()); assert!(undo.is_control()); } + + #[test] + fn test_parser_user_input() { + let submission = SubmissionParser::parse("Hello, how are you?"); + assert!( + matches!(submission, Submission::UserInput { content } if content == "Hello, how are you?") + ); + } + + #[test] + fn test_parser_undo() { + let submission = SubmissionParser::parse("/undo"); + assert!(matches!(submission, Submission::Undo)); + + let submission = SubmissionParser::parse("/UNDO"); + assert!(matches!(submission, Submission::Undo)); + } + + #[test] + fn test_parser_redo() { + let submission = SubmissionParser::parse("/redo"); + assert!(matches!(submission, Submission::Redo)); + } + + #[test] + fn test_parser_interrupt() { + let submission = SubmissionParser::parse("/interrupt"); + assert!(matches!(submission, Submission::Interrupt)); + + let submission = SubmissionParser::parse("/stop"); + assert!(matches!(submission, Submission::Interrupt)); + } + + #[test] + fn test_parser_compact() { + let submission = SubmissionParser::parse("/compact"); + assert!(matches!(submission, Submission::Compact)); + } + + #[test] + fn test_parser_clear() { + let submission = SubmissionParser::parse("/clear"); + assert!(matches!(submission, Submission::Clear)); + } + + #[test] + fn test_parser_new_thread() { + let submission = SubmissionParser::parse("/thread new"); + assert!(matches!(submission, Submission::NewThread)); + + let submission = SubmissionParser::parse("/new"); + assert!(matches!(submission, Submission::NewThread)); + } + + #[test] + fn test_parser_switch_thread() { + let uuid = Uuid::new_v4(); + let submission = SubmissionParser::parse(&format!("/thread {}", uuid)); + assert!(matches!(submission, Submission::SwitchThread { thread_id } if thread_id == uuid)); + } + + #[test] + fn test_parser_resume() { + let uuid = Uuid::new_v4(); + let submission = SubmissionParser::parse(&format!("/resume {}", uuid)); + assert!( + matches!(submission, Submission::Resume { checkpoint_id } if checkpoint_id == uuid) + ); + } + + #[test] + fn test_parser_invalid_commands_become_user_input() { + // Invalid UUID should become user input + let submission = SubmissionParser::parse("/thread not-a-uuid"); + assert!(matches!(submission, Submission::UserInput { .. })); + + // Unknown command should become user input + let submission = SubmissionParser::parse("/unknown"); + assert!(matches!(submission, Submission::UserInput { content } if content == "/unknown")); + } } diff --git a/src/channels/mod.rs b/src/channels/mod.rs index afde6c63..2806b63d 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -1,18 +1,14 @@ //! Multi-channel input system. //! -//! Channels receive messages from external sources (CLI, Slack, Telegram, HTTP) +//! Channels receive messages from external sources (CLI, HTTP, etc.) //! and convert them to a unified message format for the agent to process. mod channel; pub mod cli; mod http; mod manager; -mod slack; -mod telegram; pub use channel::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate}; pub use cli::TuiChannel; pub use http::HttpChannel; pub use manager::ChannelManager; -pub use slack::SlackChannel; -pub use telegram::TelegramChannel; diff --git a/src/channels/slack.rs b/src/channels/slack.rs deleted file mode 100644 index 1f6e8095..00000000 --- a/src/channels/slack.rs +++ /dev/null @@ -1,60 +0,0 @@ -//! Slack channel integration. -//! -//! TODO: Implement full Slack bot integration using slack-morphism. - -use async_trait::async_trait; - -use crate::channels::{Channel, IncomingMessage, MessageStream, OutgoingResponse}; -use crate::config::SlackConfig; -use crate::error::ChannelError; - -/// Slack channel for Slack bot integration. -pub struct SlackChannel { - #[allow(dead_code)] - config: SlackConfig, -} - -impl SlackChannel { - /// Create a new Slack channel. - pub fn new(config: SlackConfig) -> Self { - Self { config } - } -} - -#[async_trait] -impl Channel for SlackChannel { - fn name(&self) -> &str { - "slack" - } - - async fn start(&self) -> Result { - // TODO: Implement Slack socket mode connection - // 1. Connect via Slack Socket Mode - // 2. Listen for app_mention and direct message events - // 3. Convert Slack events to IncomingMessage - Err(ChannelError::StartupFailed { - name: "slack".to_string(), - reason: "Slack channel not yet implemented".to_string(), - }) - } - - async fn respond( - &self, - _msg: &IncomingMessage, - _response: OutgoingResponse, - ) -> Result<(), ChannelError> { - // TODO: Use Slack Web API to post message - // - If in thread, reply in thread - // - Support blocks for rich formatting - Err(ChannelError::SendFailed { - name: "slack".to_string(), - reason: "Slack channel not yet implemented".to_string(), - }) - } - - async fn health_check(&self) -> Result<(), ChannelError> { - Err(ChannelError::HealthCheckFailed { - name: "slack".to_string(), - }) - } -} diff --git a/src/channels/telegram.rs b/src/channels/telegram.rs deleted file mode 100644 index b6f03e19..00000000 --- a/src/channels/telegram.rs +++ /dev/null @@ -1,60 +0,0 @@ -//! Telegram channel integration. -//! -//! TODO: Implement full Telegram bot integration using teloxide. - -use async_trait::async_trait; - -use crate::channels::{Channel, IncomingMessage, MessageStream, OutgoingResponse}; -use crate::config::TelegramConfig; -use crate::error::ChannelError; - -/// Telegram channel for Telegram bot integration. -pub struct TelegramChannel { - #[allow(dead_code)] - config: TelegramConfig, -} - -impl TelegramChannel { - /// Create a new Telegram channel. - pub fn new(config: TelegramConfig) -> Self { - Self { config } - } -} - -#[async_trait] -impl Channel for TelegramChannel { - fn name(&self) -> &str { - "telegram" - } - - async fn start(&self) -> Result { - // TODO: Implement Telegram long polling or webhook - // 1. Use teloxide to connect to Telegram Bot API - // 2. Handle incoming messages - // 3. Convert to IncomingMessage format - Err(ChannelError::StartupFailed { - name: "telegram".to_string(), - reason: "Telegram channel not yet implemented".to_string(), - }) - } - - async fn respond( - &self, - _msg: &IncomingMessage, - _response: OutgoingResponse, - ) -> Result<(), ChannelError> { - // TODO: Use Telegram Bot API to send message - // - Reply to the same chat - // - Support reply_to_message_id for threaded replies - Err(ChannelError::SendFailed { - name: "telegram".to_string(), - reason: "Telegram channel not yet implemented".to_string(), - }) - } - - async fn health_check(&self) -> Result<(), ChannelError> { - Err(ChannelError::HealthCheckFailed { - name: "telegram".to_string(), - }) - } -} diff --git a/src/config.rs b/src/config.rs index b397c142..785be960 100644 --- a/src/config.rs +++ b/src/config.rs @@ -104,8 +104,6 @@ impl LlmConfig { #[derive(Debug, Clone)] pub struct ChannelsConfig { pub cli: CliConfig, - pub slack: Option, - pub telegram: Option, pub http: Option, } @@ -114,18 +112,6 @@ pub struct CliConfig { pub enabled: bool, } -#[derive(Debug, Clone)] -pub struct SlackConfig { - pub bot_token: SecretString, - pub app_token: SecretString, - pub signing_secret: SecretString, -} - -#[derive(Debug, Clone)] -pub struct TelegramConfig { - pub bot_token: SecretString, -} - #[derive(Debug, Clone)] pub struct HttpConfig { pub host: String, @@ -135,29 +121,6 @@ pub struct HttpConfig { impl ChannelsConfig { fn from_env() -> Result { - let slack = match ( - optional_env("SLACK_BOT_TOKEN")?, - optional_env("SLACK_APP_TOKEN")?, - optional_env("SLACK_SIGNING_SECRET")?, - ) { - (Some(bot_token), Some(app_token), Some(signing_secret)) => Some(SlackConfig { - bot_token: SecretString::from(bot_token), - app_token: SecretString::from(app_token), - signing_secret: SecretString::from(signing_secret), - }), - (None, None, None) => None, - _ => { - return Err(ConfigError::InvalidValue { - key: "SLACK_*".to_string(), - message: "all Slack environment variables must be set together".to_string(), - }); - } - }; - - let telegram = optional_env("TELEGRAM_BOT_TOKEN")?.map(|token| TelegramConfig { - bot_token: SecretString::from(token), - }); - let http = if optional_env("HTTP_PORT")?.is_some() || optional_env("HTTP_HOST")?.is_some() { Some(HttpConfig { host: optional_env("HTTP_HOST")?.unwrap_or_else(|| "0.0.0.0".to_string()), @@ -177,8 +140,6 @@ impl ChannelsConfig { Ok(Self { cli: CliConfig { enabled: true }, - slack, - telegram, http, }) } diff --git a/src/llm/reasoning.rs b/src/llm/reasoning.rs index b72b9782..dda52a34 100644 --- a/src/llm/reasoning.rs +++ b/src/llm/reasoning.rs @@ -39,6 +39,12 @@ impl ReasoningContext { self } + /// Set messages directly (for session-based context). + pub fn with_messages(mut self, messages: Vec) -> Self { + self.messages = messages; + self + } + /// Set available tools. pub fn with_tools(mut self, tools: Vec) -> Self { self.available_tools = tools; diff --git a/src/main.rs b/src/main.rs index 16c0679b..baddeaa0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -164,18 +164,23 @@ async fn main() -> anyhow::Result<()> { http_config.port ); } - - // TODO: Add Slack and Telegram channels when implemented - if config.channels.slack.is_some() { - tracing::warn!("Slack channel configured but not yet implemented"); - } - if config.channels.telegram.is_some() { - tracing::warn!("Telegram channel configured but not yet implemented"); - } } + // Create workspace for agent (shared with memory tools) + let workspace = store + .as_ref() + .map(|s| Arc::new(Workspace::new("default", s.pool()))); + // Create and run the agent - let agent = Agent::new(config.agent.clone(), store, llm, safety, tools, channels); + let agent = Agent::new( + config.agent.clone(), + store, + llm, + safety, + tools, + channels, + workspace, + ); tracing::info!("Agent initialized, starting main loop...");