Extend support for session management

2026-08-25 14:53:34 +00:00 · 2026-02-03 08:59:59 -08:00
parent c9ebb117ab
commit dedda9c51d
11 changed files with 886 additions and 252 deletions
@@ -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<ContextManager>,
    scheduler: Arc<Scheduler>,
    router: Router,
+    session_manager: Arc<SessionManager>,
+    context_monitor: ContextMonitor,
+    workspace: Option<Arc<Workspace>>,
 }

 impl Agent {
@@ -38,6 +48,7 @@ impl Agent {
        safety: Arc<SafetyLayer>,
        tools: Arc<ToolRegistry>,
        channels: ChannelManager,
+        workspace: Option<Arc<Workspace>>,
    ) -> 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<Mutex<Session>>,
+        thread_id: Uuid,
+        content: &str,
+    ) -> Result<SubmissionResult, Error> {
+        // 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<SubmissionResult, Error> {
+        // 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<Mutex<Session>>,
+        thread_id: Uuid,
+    ) -> Result<SubmissionResult, Error> {
+        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<Mutex<Session>>,
+        thread_id: Uuid,
+    ) -> Result<SubmissionResult, Error> {
+        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<Mutex<Session>>,
+        thread_id: Uuid,
+    ) -> Result<SubmissionResult, Error> {
+        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<Mutex<Session>>,
+        thread_id: Uuid,
+    ) -> Result<SubmissionResult, Error> {
+        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<Mutex<Session>>,
+        thread_id: Uuid,
+    ) -> Result<SubmissionResult, Error> {
+        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<SubmissionResult, Error> {
+        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<SubmissionResult, Error> {
+        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<Mutex<Session>>,
+        thread_id: Uuid,
+        checkpoint_id: Uuid,
+    ) -> Result<SubmissionResult, Error> {
+        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<String, Error> {
-        // 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<Option<String>, Error> {
        match command {
            "help" => Ok(Some(
-                r#"Available commands:
-  /job <description>  - Create a new job
-  /status [job_id]    - Check job status
-  /cancel <job_id>    - Cancel a job
-  /list               - List all jobs
-  /help <job_id>      - Help a stuck job
-  /quit               - Exit the agent
+                r#"Commands:
+  /job <desc>     - Create a job
+  /status [id]    - Check job status
+  /cancel <id>    - Cancel a job
+  /list           - List all jobs
+  /help <job_id>  - 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 <id>    - Switch thread
+  /resume <id>    - Resume checkpoint
+
+  /quit           - Exit"#
                    .to_string(),
            )),

@@ -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;
@@ -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<ChatMessage>) {
+        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());
+    }
 }
@@ -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<String>,
+}
+
+/// Manages sessions, threads, and undo state for all users.
+pub struct SessionManager {
+    sessions: RwLock<HashMap<String, Arc<Mutex<Session>>>>,
+    thread_map: RwLock<HashMap<ThreadKey, Uuid>>,
+    undo_managers: RwLock<HashMap<Uuid, Arc<Mutex<UndoManager>>>>,
+}
+
+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<Mutex<Session>> {
+        // 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<Mutex<Session>>, 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<Mutex<UndoManager>> {
+        // 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));
+    }
+}
@@ -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 <uuid> - 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 <uuid> - 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"));
+    }
 }
@@ -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;
@@ -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<MessageStream, ChannelError> {
-        // 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(),
-        })
-    }
-}
@@ -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<MessageStream, ChannelError> {
-        // 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(),
-        })
-    }
-}
@@ -104,8 +104,6 @@ impl LlmConfig {
 #[derive(Debug, Clone)]
 pub struct ChannelsConfig {
    pub cli: CliConfig,
-    pub slack: Option<SlackConfig>,
-    pub telegram: Option<TelegramConfig>,
    pub http: Option<HttpConfig>,
 }

@@ -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<Self, ConfigError> {
-        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,
        })
    }
@@ -39,6 +39,12 @@ impl ReasoningContext {
        self
    }

+    /// Set messages directly (for session-based context).
+    pub fn with_messages(mut self, messages: Vec<ChatMessage>) -> Self {
+        self.messages = messages;
+        self
+    }
+
    /// Set available tools.
    pub fn with_tools(mut self, tools: Vec<ToolDefinition>) -> Self {
        self.available_tools = tools;
@@ -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...");