session.rs

//! Session runner - manages the lifecycle of a conversation session.
//!
//! This module provides the core session management that connects
//! the TUI/CLI to the agent loop.

use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use async_channel::{Receiver, Sender, unbounded};
use chrono::Utc;
use tokio_stream::StreamExt;
use tracing::{info, warn};

use cortex_protocol::{
    AgentMessageDeltaEvent, AgentMessageEvent, ConversationId, ErrorEvent, Event, EventMsg,
    ExecApprovalRequestEvent, ExecCommandBeginEvent, ExecCommandEndEvent,
    ExecCommandOutputDeltaEvent, ExecCommandSource, ExecOutputStream, Op, ParsedCommand,
    SessionConfiguredEvent, Submission, TaskCompleteEvent, TaskStartedEvent, TokenCountEvent,
    TokenUsage, TokenUsageInfo, UserMessageEvent,
};

use crate::client::types::MessageContent;
use crate::client::{
    CompletionRequest, Message, MessageRole, ModelClient, ResponseEvent, ToolCall,
    ToolDefinition as ClientToolDefinition, create_client,
};
use crate::config::Config;
use crate::error::{CortexError, Result};
use crate::rollout::reader::{RolloutItem, get_events, get_session_meta};
use crate::rollout::recorder::SessionMeta;
use crate::rollout::{RolloutRecorder, SESSIONS_SUBDIR, get_rollout_path, read_rollout};
use crate::skills::{render_skills_section, try_get_skills_manager};
use crate::summarization::SummarizationStrategy;
use crate::tools::context::ToolOutputChunk;
use crate::tools::{ToolContext, ToolRouter};

/// Simple token counter for session context tracking.
/// Uses approximation-based counting (4 chars per token).
pub struct TokenCounter;

impl Default for TokenCounter {
    fn default() -> Self {
        Self::new()
    }
}

impl TokenCounter {
    /// Create a new token counter.
    pub fn new() -> Self {
        Self
    }

    /// Count tokens in messages (approximate).
    pub async fn count_messages(&self, _model: &str, messages: &[Message]) -> Result<usize> {
        let mut total = 0usize;
        for msg in messages {
            // Base overhead per message (~4 tokens)
            total += 4;
            // Count content tokens
            let text = match &msg.content {
                MessageContent::Text(t) => t.len(),
                MessageContent::Parts(parts) => parts
                    .iter()
                    .map(|p| match p {
                        crate::client::types::ContentPart::Text { text, .. } => text.len(),
                        _ => 85, // Image tokens approximation
                    })
                    .sum(),
                MessageContent::ToolResult { content, .. } => content.len(),
                MessageContent::ToolCalls(calls) => {
                    calls.iter().map(|c| c.name.len() + c.arguments.len()).sum()
                }
            };
            // Approximate: 4 chars per token
            total += (text as f64 / 4.0).ceil() as usize;
        }
        // Message separator overhead
        total += messages.len() * 3;
        Ok(total)
    }

    /// Count tokens in tool definitions (approximate).
    pub async fn count_tools(&self, _model: &str, tools: &[ClientToolDefinition]) -> Result<usize> {
        let mut total = 0usize;
        for tool in tools {
            let json = serde_json::to_string(tool).unwrap_or_default();
            // Approximate: 4 chars per token + 10 overhead
            total += (json.len() as f64 / 4.0).ceil() as usize + 10;
        }
        // Base overhead for having tools
        if !tools.is_empty() {
            total += 9;
        }
        Ok(total)
    }
}

/// A running session that handles conversation with the model.
pub struct Session {
    /// Session configuration.
    config: Config,
    /// Conversation ID.
    conversation_id: ConversationId,
    /// Model client.
    client: Box<dyn ModelClient>,
    /// Tool router.
    tool_router: ToolRouter,
    /// Conversation messages.
    messages: Vec<Message>,
    /// Submission receiver (from UI).
    submission_rx: Receiver<Submission>,
    /// Event sender (to UI).
    event_tx: Sender<Event>,
    /// Current turn ID.
    turn_id: u64,
    /// Token usage tracking.
    total_usage: TokenUsage,
    /// Token counter for precise context window tracking.
    token_counter: Arc<TokenCounter>,
    /// Whether session is running.
    running: bool,
    /// Current agent profile.
    pub current_agent_profile: Option<crate::agent::AgentProfile>,
    /// Rollout recorder for persistence.
    recorder: Option<RolloutRecorder>,
    /// Pending approval requests (call_id -> (tool_call, args)).
    pending_approvals: std::collections::HashMap<String, PendingToolCall>,
    /// Cancellation flag for interrupting current request.
    cancelled: Arc<AtomicBool>,
    /// Snapshot manager for undo/redo.
    snapshot_manager: Arc<crate::tasks::snapshot::SnapshotManager>,
    /// Undo history.
    undo_history: crate::tasks::undo::UndoHistory,
    /// Redo history.
    redo_history: crate::tasks::undo::RedoHistory,
    /// Current turn's undo actions.
    current_undo_actions: Vec<crate::tasks::undo::UndoAction>,
    /// Share service for generating public URLs.
    share_service: crate::share_service::ShareService,
    /// LSP integration.
    lsp: Arc<crate::integrations::LspIntegration>,
}

/// A tool call waiting for approval.
#[derive(Debug, Clone)]
struct PendingToolCall {
    tool_name: String,
    arguments: serde_json::Value,
    tool_call_id: String,
}

/// Handle for interacting with a session.
#[derive(Clone)]
pub struct SessionHandle {
    /// Submission sender.
    pub submission_tx: Sender<Submission>,
    /// Event receiver.
    pub event_rx: Receiver<Event>,
    /// Conversation ID.
    pub conversation_id: ConversationId,
    /// Cancellation flag.
    pub cancelled: Arc<AtomicBool>,
}

impl Session {
    /// Create a new session with channels.
    pub fn new(config: Config) -> Result<(Self, SessionHandle)> {
        let (submission_tx, submission_rx) = unbounded();
        let (event_tx, event_rx) = unbounded();

        let conversation_id = ConversationId::new();
        let cancelled = Arc::new(AtomicBool::new(false));

        // Get API key from environment or config
        let api_key = std::env::var("OPENAI_API_KEY")
            .or_else(|_| std::env::var("ANTHROPIC_API_KEY"))
            .or_else(|_| std::env::var("OPENROUTER_API_KEY"))
            .unwrap_or_default();

        let client = create_client(
            &config.model_provider_id,
            &config.model,
            &api_key,
            Some(config.model_provider.base_url.as_str()),
        )?;

        let mut tool_router = ToolRouter::new();

        // Initialize rollout recorder
        let mut recorder = RolloutRecorder::new(&config.cortex_home, conversation_id)?;
        recorder.init()?;

        // Record session metadata
        let meta = SessionMeta {
            id: conversation_id,
            parent_id: None,
            fork_point: None,
            timestamp: Utc::now().to_rfc3339(),
            cwd: config.cwd.clone(),
            model: config.model.clone(),
            cli_version: env!("CARGO_PKG_VERSION").to_string(),
            instructions: config.user_instructions.clone(),
        };
        recorder.record_meta(&meta)?;

        // Initialize with system prompt (including skills if available)
        let skills_section = load_skills_section(&config.cwd);
        let mut messages = Vec::new();
        messages.push(Message::system(build_system_prompt_with_skills(
            &config,
            skills_section.as_deref(),
        )));

        // Load current agent profile if set
        let current_agent_profile = if let Some(agent_name) = &config.current_agent {
            let profiles = crate::agent::AgentProfile::load_all().unwrap_or_default();
            profiles.get(agent_name).cloned()
        } else {
            None
        };

        // Initialize snapshot manager
        let snapshot_dir = config
            .cortex_home
            .join("snapshots")
            .join(conversation_id.to_string());
        let snapshot_manager =
            Arc::new(crate::tasks::snapshot::SnapshotManager::new(50).with_storage(snapshot_dir));

        // Initialize LSP integration
        let lsp = Arc::new(crate::integrations::LspIntegration::new(true));
        let lsp_clone = lsp.clone();
        let cwd_clone = config.cwd.clone();
        tokio::spawn(async move {
            if let Err(e) = lsp_clone.init(&cwd_clone).await {
                warn!("Failed to initialize LSP in session: {}", e);
            }
        });

        tool_router.set_lsp(lsp.clone());

        let session = Self {
            config,
            conversation_id,
            client,
            tool_router,
            messages,
            submission_rx,
            event_tx,
            turn_id: 0,
            total_usage: TokenUsage::default(),
            token_counter: Arc::new(TokenCounter),
            running: true,
            current_agent_profile,
            recorder: Some(recorder),
            pending_approvals: std::collections::HashMap::new(),
            cancelled: cancelled.clone(),
            snapshot_manager,
            undo_history: crate::tasks::undo::UndoHistory::new(50),
            redo_history: crate::tasks::undo::RedoHistory::new(50),
            current_undo_actions: Vec::new(),
            share_service: crate::share_service::ShareService::new(),
            lsp,
        };

        let handle = SessionHandle {
            submission_tx,
            event_rx,
            conversation_id: session.conversation_id,
            cancelled,
        };

        Ok((session, handle))
    }

    /// Run the session loop.
    pub async fn run(&mut self) -> Result<()> {
        // Emit session configured event
        self.emit(EventMsg::SessionConfigured(Box::new(
            SessionConfiguredEvent {
                session_id: self.conversation_id,
                parent_session_id: None,
                model: self.config.model.clone(),
                model_provider_id: self.config.model_provider_id.clone(),
                approval_policy: self.config.approval_policy,
                sandbox_policy: self.config.sandbox_policy.clone(),
                cwd: self.config.cwd.clone(),
                reasoning_effort: None,
                history_log_id: 0,
                history_entry_count: 0,
                initial_messages: None,
                rollout_path: PathBuf::new(),
            },
        )))
        .await;

        while self.running {
            // Check cancellation flag periodically using a timeout
            let submission = tokio::select! {
                result = self.submission_rx.recv() => {
                    match result {
                        Ok(s) => s,
                        Err(_) => break,
                    }
                }
                _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
                    // Check if we should exit due to cancellation
                    if self.cancelled.load(Ordering::SeqCst) {
                        tracing::info!("Session detected cancellation, exiting run loop");
                        self.running = false;
                        break;
                    }
                    continue;
                }
            };

            if let Err(e) = self.handle_submission(submission).await {
                self.emit(EventMsg::Error(ErrorEvent {
                    message: e.to_string(),
                    cortex_error_info: None,
                }))
                .await;
            }
        }

        Ok(())
    }

    async fn handle_submission(&mut self, submission: Submission) -> Result<()> {
        match submission.op {
            Op::Shutdown => {
                self.running = false;
                self.emit(EventMsg::ShutdownComplete).await;
            }
            Op::Interrupt => {
                // Set cancellation flag to stop current request
                self.cancelled.store(true, Ordering::SeqCst);
                self.emit(EventMsg::TurnAborted(cortex_protocol::TurnAbortedEvent {
                    reason: cortex_protocol::TurnAbortReason::Interrupted,
                }))
                .await;
            }
            Op::UserInput { items } => {
                self.handle_user_input(&submission.id, items).await?;
            }
            Op::UserTurn { items, .. } => {
                self.handle_user_input(&submission.id, items).await?;
            }
            Op::Compact => {
                self.handle_compact().await?;
            }
            Op::Undo => {
                self.handle_undo().await?;
            }
            Op::Redo => {
                self.handle_redo().await?;
            }
            Op::ForkSession {
                fork_point_message_id,
                message_index,
            } => {
                self.handle_fork_session(fork_point_message_id, message_index)
                    .await?;
            }
            Op::ExecApproval { id, decision } => {
                self.handle_exec_approval(&id, decision).await?;
            }
            Op::OverrideTurnContext {
                cwd,
                approval_policy,
                sandbox_policy,
                model,
                effort,
                summary,
            } => {
                if let Some(cwd) = cwd {
                    info!("Updating session CWD to: {:?}", cwd);
                    self.config.cwd = cwd;
                }
                if let Some(policy) = approval_policy {
                    self.config.approval_policy = policy;
                }
                if let Some(policy) = sandbox_policy {
                    self.config.sandbox_policy = policy;
                }
                if let Some(model) = model {
                    self.config.model = model;
                }
                if let Some(effort) = effort {
                    self.config.reasoning_effort = effort.map(|e| match e {
                        cortex_protocol::ReasoningEffort::Low => {
                            crate::config::ReasoningEffort::Low
                        }
                        cortex_protocol::ReasoningEffort::Medium => {
                            crate::config::ReasoningEffort::Medium
                        }
                        cortex_protocol::ReasoningEffort::High => {
                            crate::config::ReasoningEffort::High
                        }
                    });
                }
                if let Some(summary) = summary {
                    self.config.reasoning_summary = match summary {
                        cortex_protocol::ReasoningSummary::None => {
                            crate::config::ReasoningSummary::None
                        }
                        cortex_protocol::ReasoningSummary::Brief => {
                            crate::config::ReasoningSummary::Brief
                        }
                        cortex_protocol::ReasoningSummary::Detailed => {
                            crate::config::ReasoningSummary::Detailed
                        }
                        cortex_protocol::ReasoningSummary::Auto => {
                            crate::config::ReasoningSummary::Auto
                        }
                    };
                }
            }
            Op::SwitchAgent { name } => {
                info!("Switching agent to: {}", name);
                self.config.current_agent = Some(name.clone());
                // Load and set the agent profile
                let profiles = crate::agent::AgentProfile::load_all().unwrap_or_default();
                self.current_agent_profile = profiles.get(&name).cloned();

                // Update system prompt in existing message history
                if let Some(msg) = self.messages.first_mut()
                    && matches!(msg.role, crate::client::MessageRole::System)
                {
                    *msg = Message::system(build_system_prompt(&self.config));
                }
            }
            Op::Share => {
                let url = self
                    .share_service
                    .share(&self.conversation_id.to_string(), &self.messages)
                    .await?;
                self.emit(EventMsg::SessionShared(
                    cortex_protocol::SessionSharedEvent { url },
                ))
                .await;
            }
            Op::Unshare => {
                self.share_service
                    .unshare(&self.conversation_id.to_string())
                    .await?;
                self.emit(EventMsg::SessionUnshared(
                    cortex_protocol::SessionUnsharedEvent { success: true },
                ))
                .await;
            }
            Op::ReloadMcpServers => {
                info!("Reloading MCP servers...");
            }
            Op::EnableMcpServer { name } => {
                info!("Enabling MCP server: {}...", name);
            }
            Op::DisableMcpServer { name } => {
                info!("Disabling MCP server: {}...", name);
            }
            _ => {}
        }
        Ok(())
    }

    async fn handle_user_input(
        &mut self,
        _submission_id: &str,
        items: Vec<cortex_protocol::UserInput>,
    ) -> Result<()> {
        tracing::info!("Session handling user input: {} items", items.len());

        // Reset cancellation flag at start of each turn
        self.cancelled.store(false, Ordering::SeqCst);

        self.turn_id += 1;
        let turn_id = self.turn_id.to_string();

        // Extract text from user input
        let user_text: String = items
            .iter()
            .filter_map(|item| {
                if let cortex_protocol::UserInput::Text { text } = item {
                    Some(text.clone())
                } else {
                    None
                }
            })
            .collect::<Vec<_>>()
            .join("\n");

        if user_text.is_empty() {
            tracing::warn!("Session received empty user input");
            return Ok(());
        }

        tracing::debug!("User message: {}", user_text);

        // Track messages for redo
        let _msg_start_idx = self.messages.len();

        // Emit user message event
        self.emit(EventMsg::UserMessage(UserMessageEvent {
            id: None,
            parent_id: None,
            message: user_text.clone(),
            images: None,
        }))
        .await;

        // Add user message to history
        self.messages.push(Message::user(&user_text));

        // Emit task started
        self.emit(EventMsg::TaskStarted(TaskStartedEvent {
            model_context_window: self.config.model_context_window,
        }))
        .await;

        // Fast git-based snapshot (uses git write-tree, instant)
        // Only track if the cwd is a git repository (has .git directory)
        let is_git_repo = self.config.cwd.join(".git").exists()
            || std::process::Command::new("git")
                .args(["rev-parse", "--is-inside-work-tree"])
                .current_dir(&self.config.cwd)
                .output()
                .map(|o| o.status.success())
                .unwrap_or(false);

        let pre_snapshot_hash = if is_git_repo {
            match crate::git_snapshot::GitSnapshot::new(
                &self.config.cortex_home,
                &self.conversation_id.to_string(),
                self.config.cwd.clone(),
            ) {
                Ok(gs) => gs.track().ok(),
                Err(e) => {
                    tracing::warn!("Failed to create git snapshot: {}", e);
                    None
                }
            }
        } else {
            tracing::debug!("Skipping snapshot - cwd is not a git repository");
            None
        };

        // Run the agent loop until complete
        tracing::info!("Starting agent loop for turn {}...", turn_id);
        if let Err(e) = self.run_agent_loop(&turn_id).await {
            tracing::error!("Agent loop failed: {}", e);
            // Emit error event so TUI can display it
            self.emit(EventMsg::Error(ErrorEvent {
                message: e.to_string(),
                cortex_error_info: None,
            }))
            .await;
            // Emit TaskComplete to reset TUI state
            self.emit(EventMsg::TaskComplete(TaskCompleteEvent {
                last_agent_message: None,
            }))
            .await;
            return Err(e);
        }
        tracing::info!("Agent loop completed for turn {}", turn_id);

        // Fast git-based diff (if we have a pre-snapshot)
        if let Some(hash) = pre_snapshot_hash
            && let Ok(gs) = crate::git_snapshot::GitSnapshot::new(
                &self.config.cortex_home,
                &self.conversation_id.to_string(),
                self.config.cwd.clone(),
            )
        {
            // Get diff for display
            if let Ok(diff_text) = gs.diff(&hash)
                && !diff_text.is_empty()
            {
                use cortex_protocol::TurnDiffEvent;
                self.emit(EventMsg::TurnDiff(TurnDiffEvent {
                    unified_diff: diff_text,
                }))
                .await;
            }
        }

        // Create undo task
        let undo_task = crate::tasks::undo::UndoTask::new(
            format!("undo_{turn_id}"),
            crate::tasks::undo::UndoTarget::Turn(turn_id.clone()),
        );

        self.undo_history.push(undo_task);

        Ok(())
    }

    async fn handle_compact(&mut self) -> Result<()> {
        let strategy = SummarizationStrategy::default();
        let (to_summarize, to_keep) = strategy.split_messages(&self.messages);

        if to_summarize.is_empty() {
            return Ok(());
        }

        // Build summarization prompt
        let prompt = strategy.build_summarization_prompt(&to_summarize);

        // Call model to summarize
        let request = CompletionRequest {
            model: "gpt-4o-mini".to_string(), // Use a cheaper model for summarization
            messages: prompt,
            max_tokens: Some(strategy.target_summary_tokens as u32),
            temperature: Some(0.3),
            ..Default::default()
        };

        let mut stream = self.client.complete(request).await?;
        let mut summary = String::new();

        while let Some(event) = stream.next().await {
            if let ResponseEvent::Delta(delta) = event? {
                summary.push_str(&delta);
            }
        }

        // Replace messages
        let mut new_messages = Vec::new();
        if strategy.preserve_system
            && !self.messages.is_empty()
            && self.messages[0].role == MessageRole::System
        {
            new_messages.push(self.messages[0].clone());
        }

        new_messages.push(Message::system(format!(
            "[Conversation Summary]\n\n{}",
            summary
        )));

        new_messages.extend(to_keep);

        self.messages = new_messages;

        // Emit event or log
        tracing::info!("Context compacted using summarization strategy");

        Ok(())
    }

    /// Capture a snapshot of the current workspace.
    async fn capture_snapshot(
        &self,
        description: &str,
    ) -> Result<crate::tasks::snapshot::Snapshot> {
        let mut snapshot = self.snapshot_manager.create(description).await;
        snapshot.turn_id = Some(self.turn_id.to_string());

        // Walk workspace and capture files
        // For performance, we only capture files that are likely to be changed
        // or we could use a more sophisticated approach.
        // For now, let's capture everything in CWD except ignored patterns.
        let mut it = walkdir::WalkDir::new(&self.config.cwd).into_iter();
        loop {
            // Check for cancellation periodically
            if self.cancelled.load(Ordering::SeqCst) {
                tracing::info!("Snapshot capture cancelled by user");
                return Err(CortexError::Cancelled);
            }

            let entry = match it.next() {
                None => break,
                Some(Err(_)) => continue,
                Some(Ok(e)) => e,
            };

            let path = entry.path();
            if path.is_dir() {
                let name = path.file_name().and_then(|n| n.to_str()).unwrap_or("");
                if name == ".git"
                    || name == "target"
                    || name == "node_modules"
                    || name == ".factory"
                {
                    it.skip_current_dir();
                }
                continue;
            }

            // Skip large files or binary files if needed
            // For now, just capture
            if let Err(e) = snapshot.capture_file(path) {
                tracing::warn!("Failed to capture file {}: {}", path.display(), e);
            }
        }

        Ok(snapshot)
    }

    async fn run_agent_loop(&mut self, _turn_id: &str) -> Result<()> {
        let max_iterations = 200;
        let mut iteration = 0;

        loop {
            // Check for cancellation
            if self.cancelled.load(Ordering::SeqCst) {
                tracing::info!("Agent loop cancelled by user");
                break;
            }

            iteration += 1;
            if iteration > max_iterations {
                self.emit(EventMsg::Error(ErrorEvent {
                    message: "Maximum iterations reached".to_string(),
                    cortex_error_info: None,
                }))
                .await;
                break;
            }

            // Build completion request
            let tool_defs = self.tool_router.get_tool_definitions();

            // Calculate precise context tokens
            let client_tools: Vec<ClientToolDefinition> = tool_defs
                .iter()
                .map(|t| {
                    ClientToolDefinition::function(&t.name, &t.description, t.parameters.clone())
                })
                .collect();

            let context_tokens = self
                .token_counter
                .count_messages(&self.config.model, &self.messages)
                .await
                .unwrap_or(0);
            let tools_tokens = self
                .token_counter
                .count_tools(&self.config.model, &client_tools)
                .await
                .unwrap_or(0);
            let total_context_tokens = context_tokens + tools_tokens;

            // Emit token count event before starting request
            self.emit(EventMsg::TokenCount(TokenCountEvent {
                info: Some(TokenUsageInfo {
                    total_token_usage: self.total_usage.clone(),
                    last_token_usage: TokenUsage::default(),
                    model_context_window: self.config.model_context_window,
                    context_tokens: total_context_tokens as i64,
                }),
                rate_limits: None,
            }))
            .await;

            let tools = client_tools;

            let request = CompletionRequest {
                model: self.config.model.clone(),
                messages: self.messages.clone(),
                max_tokens: Some(4096),
                temperature: Some(0.7),
                tools,
                stream: true,
            };

            // Get streaming response
            let mut stream = self.client.complete(request).await?;

            let mut full_content = String::new();
            let mut tool_calls: Vec<ToolCall> = Vec::new();

            // Process stream
            while let Some(event) = stream.next().await {
                // Check for cancellation during streaming
                if self.cancelled.load(Ordering::SeqCst) {
                    tracing::info!("Stream processing cancelled by user");
                    return Ok(());
                }

                match event? {
                    ResponseEvent::Delta(delta) => {
                        full_content.push_str(&delta);
                        self.emit(EventMsg::AgentMessageDelta(AgentMessageDeltaEvent {
                            delta,
                        }))
                        .await;
                    }
                    ResponseEvent::ToolCall(tc) => {
                        tracing::info!("Session received tool call: {} (id: {})", tc.name, tc.id);
                        tool_calls.push(ToolCall {
                            id: tc.id,
                            call_type: "function".to_string(),
                            function: crate::client::FunctionCall {
                                name: tc.name,
                                arguments: tc.arguments,
                            },
                        });
                    }
                    ResponseEvent::Done(response) => {
                        // Update token usage
                        self.total_usage.input_tokens += response.usage.input_tokens;
                        self.total_usage.output_tokens += response.usage.output_tokens;
                        self.total_usage.total_tokens += response.usage.total_tokens;

                        self.emit(EventMsg::TokenCount(TokenCountEvent {
                            info: Some(TokenUsageInfo {
                                total_token_usage: self.total_usage.clone(),
                                last_token_usage: TokenUsage {
                                    input_tokens: response.usage.input_tokens,
                                    output_tokens: response.usage.output_tokens,
                                    total_tokens: response.usage.total_tokens,
                                    ..Default::default()
                                },
                                model_context_window: self.config.model_context_window,
                                context_tokens: response.usage.input_tokens
                                    + response.usage.output_tokens,
                            }),
                            rate_limits: None,
                        }))
                        .await;
                    }
                    ResponseEvent::Error(e) => {
                        self.emit(EventMsg::Error(ErrorEvent {
                            message: e,
                            cortex_error_info: None,
                        }))
                        .await;
                        return Ok(());
                    }
                    _ => {}
                }
            }

            // Emit full message if we have content
            if !full_content.is_empty() {
                self.emit(EventMsg::AgentMessage(AgentMessageEvent {
                    id: None,
                    parent_id: None,
                    message: full_content.clone(),
                }))
                .await;
            }

            // Add assistant message to history
            let mut assistant_msg = Message::assistant(&full_content);
            if !tool_calls.is_empty() {
                assistant_msg.tool_calls = Some(tool_calls.clone());
            }
            self.messages.push(assistant_msg);

            // If no tool calls, we're done
            if tool_calls.is_empty() {
                break;
            }

            // Execute tool calls
            tracing::info!("Processing {} tool calls", tool_calls.len());
            for tool_call in tool_calls {
                // Check for cancellation before each tool
                if self.cancelled.load(Ordering::SeqCst) {
                    tracing::info!("Tool execution cancelled by user");
                    return Ok(());
                }

                let tool_name = &tool_call.function.name;
                tracing::info!("Processing tool call: {} (id: {})", tool_name, tool_call.id);
                let args: serde_json::Value = serde_json::from_str(&tool_call.function.arguments)
                    .unwrap_or(serde_json::Value::Null);

                // Check agent permissions first
                // Use a local boolean to avoid borrowing self mutably later
                let mut agent_denied = false;
                if let Some(profile) = &self.current_agent_profile
                    && let crate::agent::ToolPermission::Deny = profile.can_use_tool(tool_name)
                {
                    agent_denied = true;
                }

                if agent_denied {
                    let profile_name = self
                        .current_agent_profile
                        .as_ref()
                        .map(|p| p.name.clone())
                        .unwrap_or_default();

                    tracing::warn!(
                        "Agent '{}' denied access to tool '{}'",
                        profile_name,
                        tool_name
                    );

                    self.emit(EventMsg::Error(ErrorEvent {
                        message: format!(
                            "Agent '{}' is not allowed to use tool '{}'.",
                            profile_name, tool_name
                        ),
                        cortex_error_info: None,
                    }))
                    .await;

                    // Add rejection message as tool result
                    self.messages.push(Message::tool_result(
                        &tool_call.id,
                        format!(
                            "Permission denied: Agent '{}' is not allowed to use tool '{}'.",
                            profile_name, tool_name
                        ),
                    ));
                    continue;
                }

                // Check if approval is needed for shell commands
                let needs_approval = if tool_name == "Execute" {
                    if let Some(cmd_array) = args.get("command").and_then(|c| c.as_array()) {
                        let cmd: Vec<String> = cmd_array
                            .iter()
                            .filter_map(|v| v.as_str().map(std::string::ToString::to_string))
                            .collect();

                        let analysis = crate::safety::analyze_command(&cmd, &self.config.cwd);
                        let requires = crate::safety::requires_approval(
                            &analysis,
                            &self.config.approval_policy,
                        );

                        if requires {
                            // Emit approval request
                            self.emit(EventMsg::ExecApprovalRequest(ExecApprovalRequestEvent {
                                call_id: tool_call.id.clone(),
                                turn_id: self.turn_id.to_string(),
                                command: cmd.clone(),
                                cwd: self.config.cwd.clone(),
                                sandbox_assessment: None,
                            }))
                            .await;
                            true
                        } else {
                            false
                        }
                    } else {
                        false
                    }
                } else {
                    false
                };

                // If approval needed, store pending and wait for user response
                tracing::info!("needs_approval for {}: {}", tool_name, needs_approval);
                if needs_approval {
                    tracing::info!("Tool {} requires approval, storing pending", tool_name);
                    self.pending_approvals.insert(
                        tool_call.id.clone(),
                        PendingToolCall {
                            tool_name: tool_name.clone(),
                            arguments: args,
                            tool_call_id: tool_call.id.clone(),
                        },
                    );
                    // Return early - we'll continue when approval comes
                    return Ok(());
                }
                tracing::info!("Tool {} does NOT require approval, executing", tool_name);

                // Handle PatchApply events
                if tool_name == "ApplyPatch"
                    && let Some(patch) = args.get("patch").and_then(|p| p.as_str())
                    && let Ok(file_changes) =
                        crate::tools::handlers::apply_patch::parse_unified_diff(patch)
                {
                    let mut protocol_changes = std::collections::HashMap::new();
                    for change in file_changes {
                        if let Some(path) = change.new_path.or(change.old_path) {
                            let protocol_change = if change.is_new_file {
                                cortex_protocol::FileChange::Add {
                                    content: String::new(),
                                }
                            } else if change.is_deleted {
                                cortex_protocol::FileChange::Delete {
                                    content: String::new(),
                                }
                            } else {
                                cortex_protocol::FileChange::Update {
                                    unified_diff: String::new(),
                                    move_path: None,
                                }
                            };
                            protocol_changes.insert(path, protocol_change);
                        }
                    }
                    self.emit(EventMsg::PatchApplyBegin(
                        cortex_protocol::PatchApplyBeginEvent {
                            call_id: tool_call.id.clone(),
                            turn_id: self.turn_id.to_string(),
                            auto_approved: true,
                            changes: protocol_changes,
                        },
                    ))
                    .await;
                }

                // Get command for event (for shell tools, parse the command array)
                let command_for_event: Vec<String> = if tool_name == "Execute" {
                    args.get("command")
                        .and_then(|c| c.as_array())
                        .map(|arr| {
                            arr.iter()
                                .filter_map(|v| v.as_str().map(String::from))
                                .collect()
                        })
                        .unwrap_or_else(|| vec![tool_name.clone()])
                } else {
                    vec![tool_name.clone()]
                };

                // Emit ExecCommandBegin event
                let exec_start = std::time::Instant::now();
                self.emit(EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
                    call_id: tool_call.id.clone(),
                    turn_id: self.turn_id.to_string(),
                    command: command_for_event.clone(),
                    cwd: self.config.cwd.clone(),
                    parsed_cmd: vec![ParsedCommand {
                        program: command_for_event.first().cloned().unwrap_or_default(),
                        args: command_for_event.iter().skip(1).cloned().collect(),
                    }],
                    source: ExecCommandSource::Agent,
                    interaction_input: None,
                    tool_name: Some(tool_name.clone()),
                    tool_arguments: Some(args.clone()),
                }))
                .await;

                // Create channel for streaming output
                let (output_tx, mut output_rx) =
                    tokio::sync::mpsc::channel::<(String, ToolOutputChunk)>(100);

                // Execute tool with streaming context
                let context = ToolContext::new(self.config.cwd.clone())
                    .with_sandbox_policy(self.config.sandbox_policy.clone())
                    .with_turn_id(self.turn_id.to_string())
                    .with_conversation_id(self.conversation_id.to_string())
                    .with_call_id(tool_call.id.clone())
                    .with_output_sender(output_tx)
                    .with_lsp(self.lsp.clone());

                // Clone event sender for the streaming task
                let event_tx = self.event_tx.clone();
                let turn_id = self.turn_id;

                // Spawn task to forward output chunks as events
                let streaming_task = tokio::spawn(async move {
                    while let Some((call_id, chunk)) = output_rx.recv().await {
                        let (stream, data) = match chunk {
                            ToolOutputChunk::Stdout(s) => (ExecOutputStream::Stdout, s),
                            ToolOutputChunk::Stderr(s) => (ExecOutputStream::Stderr, s),
                        };

                        // Encode chunk as base64
                        use base64::Engine;
                        let chunk_b64 =
                            base64::engine::general_purpose::STANDARD.encode(data.as_bytes());

                        let event = Event {
                            id: turn_id.to_string(),
                            msg: EventMsg::ExecCommandOutputDelta(ExecCommandOutputDeltaEvent {
                                call_id,
                                stream,
                                chunk: chunk_b64,
                            }),
                        };

                        let _ = event_tx.send(event).await;
                    }
                });

                tracing::info!("About to execute tool {} via tool_router", tool_name);
                let result = self
                    .tool_router
                    .execute(tool_name, args.clone(), &context)
                    .await;
                tracing::info!(
                    "Tool {} execution completed: {:?}",
                    tool_name,
                    result.is_ok()
                );

                // Drop the context to close the output channel
                drop(context);

                // Wait for streaming to finish (will complete now that sender is dropped)
                let _ = streaming_task.await;
                tracing::info!("Streaming task completed for tool {}", tool_name);

                let (result_text, exit_code, metadata) = match &result {
                    Ok(r) => {
                        let meta = r.metadata.as_ref().and_then(|m| m.data.clone());
                        (r.output.clone(), if r.success { 0 } else { 1 }, meta)
                    }
                    Err(e) => (format!("Error: {e}"), 1, None),
                };

                // Emit ExecCommandEnd event
                tracing::info!("Emitting ExecCommandEnd for tool {}", tool_call.id);
                let duration_ms = exec_start.elapsed().as_millis() as u64;
                self.emit(EventMsg::ExecCommandEnd(Box::new(ExecCommandEndEvent {
                    call_id: tool_call.id.clone(),
                    turn_id: self.turn_id.to_string(),
                    command: command_for_event.clone(),
                    cwd: self.config.cwd.clone(),
                    parsed_cmd: vec![ParsedCommand {
                        program: command_for_event.first().cloned().unwrap_or_default(),
                        args: command_for_event.iter().skip(1).cloned().collect(),
                    }],
                    source: ExecCommandSource::Agent,
                    interaction_input: None,
                    stdout: result_text.clone(),
                    stderr: String::new(),
                    aggregated_output: result_text.clone(),
                    exit_code,
                    duration_ms,
                    formatted_output: result_text.clone(),
                    metadata,
                })))
                .await;

                // Add tool result to messages
                self.messages
                    .push(Message::tool_result(&tool_call.id, &result_text));

                // Handle PatchApplyEnd
                if tool_name == "ApplyPatch" {
                    self.emit(EventMsg::PatchApplyEnd(
                        cortex_protocol::PatchApplyEndEvent {
                            call_id: tool_call.id.clone(),
                            turn_id: self.turn_id.to_string(),
                            stdout: result_text.clone(),
                            stderr: String::new(),
                            success: exit_code == 0,
                            changes: std::collections::HashMap::new(), // TODO: populate if needed
                        },
                    ))
                    .await;
                }
            }
        }

        // Emit task complete
        let last_msg = self
            .messages
            .last()
            .and_then(|m| m.content.as_text())
            .map(std::string::ToString::to_string);

        self.emit(EventMsg::TaskComplete(TaskCompleteEvent {
            last_agent_message: last_msg,
        }))
        .await;

        Ok(())
    }

    async fn handle_undo(&mut self) -> Result<()> {
        use cortex_protocol::{UndoCompletedEvent, UndoStartedEvent};

        self.emit(EventMsg::UndoStarted(UndoStartedEvent {
            message: Some("Undoing last turn...".to_string()),
        }))
        .await;

        let mut success = false;
        if let Some(undo_task) = self.undo_history.pop() {
            // Revert file changes
            let mut all_success = true;
            for action in &undo_task.actions {
                match action.execute() {
                    Ok(result) => {
                        if !result.is_success() {
                            tracing::error!("Undo action failed: {}", result.description());
                            all_success = false;
                        }
                    }
                    Err(e) => {
                        tracing::error!("Undo action error: {}", e);
                        all_success = false;
                    }
                }
            }

            if all_success {
                // Remove messages from history
                while let Some(msg) = self.messages.last() {
                    if matches!(msg.role, crate::client::MessageRole::User) {
                        break;
                    }
                    self.messages.pop();
                }
                if let Some(msg) = self.messages.last()
                    && matches!(msg.role, crate::client::MessageRole::User)
                {
                    self.messages.pop();
                }

                // Push to redo history
                self.redo_history.push(undo_task);
                success = true;
            }
        }

        self.emit(EventMsg::UndoCompleted(UndoCompletedEvent {
            success,
            message: if success {
                Some("Undo successful".to_string())
            } else {
                Some("Undo failed".to_string())
            },
        }))
        .await;

        Ok(())
    }

    async fn handle_redo(&mut self) -> Result<()> {
        use cortex_protocol::{RedoCompletedEvent, RedoStartedEvent};

        self.emit(EventMsg::RedoStarted(RedoStartedEvent {
            message: Some("Redoing last undone turn...".to_string()),
        }))
        .await;

        let mut success = false;
        if let Some(redo_task) = self.redo_history.pop() {
            // 1. Re-apply file changes if we have a forward diff
            if let Some(diff) = &redo_task.forward_diff
                && let Err(e) = diff.apply(&self.config.cwd).await
            {
                tracing::error!("Redo action error: {}", e);
                // Continue anyway? For now, yes, but log it.
            }

            // 2. Re-apply messages
            let turn_messages = redo_task.messages.clone();
            self.messages.extend(turn_messages.clone());

            // 3. Emit events for the re-applied messages
            for msg in turn_messages {
                match msg.role {
                    crate::client::MessageRole::User => {
                        if let Some(text) = msg.content.as_text() {
                            self.emit(EventMsg::UserMessage(UserMessageEvent {
                                id: None,
                                parent_id: None,
                                message: text.to_string(),
                                images: None,
                            }))
                            .await;
                        }
                    }
                    crate::client::MessageRole::Assistant => {
                        if let Some(text) = msg.content.as_text() {
                            self.emit(EventMsg::AgentMessage(AgentMessageEvent {
                                id: None,
                                parent_id: None,
                                message: text.to_string(),
                            }))
                            .await;
                        }
                    }
                    _ => {}
                }
            }

            // 4. Push back to undo history
            self.undo_history.push(redo_task);
            success = true;
        }

        self.emit(EventMsg::RedoCompleted(RedoCompletedEvent {
            success,
            message: if success {
                Some("Redo successful".to_string())
            } else {
                Some("No turns to redo".to_string())
            },
        }))
        .await;

        Ok(())
    }

    async fn handle_fork_session(
        &mut self,
        fork_point_message_id: Option<String>,
        message_index: Option<usize>,
    ) -> Result<()> {
        let original_id = self.conversation_id;
        let mut fork_index = self.messages.len();

        if let Some(idx) = message_index {
            if idx < self.messages.len() {
                fork_index = idx + 1;
            }
        } else if let Some(id) = &fork_point_message_id {
            // Try to parse as index first
            if let Ok(idx) = id.parse::<usize>()
                && idx < self.messages.len()
            {
                fork_index = idx + 1;
            }
        }

        // Truncate messages
        self.messages.truncate(fork_index);

        // New session ID
        self.conversation_id = ConversationId::new();

        // New recorder
        let mut recorder = RolloutRecorder::new(&self.config.cortex_home, self.conversation_id)?;
        recorder.init()?;

        // Record meta
        let meta = SessionMeta {
            id: self.conversation_id,
            parent_id: Some(original_id),
            fork_point: Some(fork_index.to_string()),
            timestamp: Utc::now().to_rfc3339(),
            cwd: self.config.cwd.clone(),
            model: self.config.model.clone(),
            cli_version: env!("CARGO_PKG_VERSION").to_string(),
            instructions: self.config.user_instructions.clone(),
        };
        recorder.record_meta(&meta)?;
        self.recorder = Some(recorder);

        // Emit SessionConfigured
        self.emit(EventMsg::SessionConfigured(Box::new(
            SessionConfiguredEvent {
                session_id: self.conversation_id,
                parent_session_id: Some(original_id),
                model: self.config.model.clone(),
                model_provider_id: self.config.model_provider_id.clone(),
                approval_policy: self.config.approval_policy,
                sandbox_policy: self.config.sandbox_policy.clone(),
                cwd: self.config.cwd.clone(),
                reasoning_effort: None,
                history_log_id: 0,
                history_entry_count: self.messages.len(),
                initial_messages: None,
                rollout_path: PathBuf::new(),
            },
        )))
        .await;

        Ok(())
    }

    async fn handle_exec_approval(
        &mut self,
        call_id: &str,
        decision: cortex_protocol::ReviewDecision,
    ) -> Result<()> {
        use cortex_protocol::ReviewDecision;

        if let Some(pending) = self.pending_approvals.remove(call_id) {
            match decision {
                ReviewDecision::Approved | ReviewDecision::ApprovedForSession => {
                    // Execute the approved tool
                    let context = ToolContext::new(self.config.cwd.clone())
                        .with_sandbox_policy(self.config.sandbox_policy.clone())
                        .with_turn_id(self.turn_id.to_string())
                        .with_conversation_id(self.conversation_id.to_string())
                        .with_lsp(self.lsp.clone());

                    let result = self
                        .tool_router
                        .execute(&pending.tool_name, pending.arguments, &context)
                        .await;

                    let result_text = match result {
                        Ok(r) => r.output,
                        Err(e) => format!("Error: {e}"),
                    };

                    // Add tool result to messages
                    self.messages
                        .push(Message::tool_result(&pending.tool_call_id, &result_text));

                    // Continue the agent loop
                    self.run_agent_loop(&self.turn_id.to_string()).await?;
                }
                ReviewDecision::Denied | ReviewDecision::Abort => {
                    // Add rejection message as tool result
                    self.messages.push(Message::tool_result(
                        &pending.tool_call_id,
                        "Command was rejected by user.",
                    ));
                }
            }
        }
        Ok(())
    }

    async fn emit(&mut self, msg: EventMsg) {
        // Skip rollout recording for delta events (too frequent, causes latency)
        let skip_recording = matches!(msg, EventMsg::AgentMessageDelta(_));

        let event = Event {
            id: self.turn_id.to_string(),
            msg,
        };

        // Record event to rollout file (skip deltas for performance)
        if !skip_recording && let Some(recorder) = &mut self.recorder {
            let _ = recorder.record_event(&event);
        }

        // Non-blocking send - never wait
        let _ = self.event_tx.try_send(event);
    }

    /// Resume a session from a rollout file.
    pub fn resume(
        config: Config,
        conversation_id: ConversationId,
    ) -> Result<(Self, SessionHandle)> {
        let (submission_tx, submission_rx) = unbounded();
        let (event_tx, event_rx) = unbounded();

        let rollout_path = get_rollout_path(&config.cortex_home, &conversation_id);
        let entries = read_rollout(&rollout_path)?;

        // Get API key
        let api_key = std::env::var("OPENAI_API_KEY")
            .or_else(|_| std::env::var("ANTHROPIC_API_KEY"))
            .or_else(|_| std::env::var("OPENROUTER_API_KEY"))
            .unwrap_or_default();

        let client = create_client(
            &config.model_provider_id,
            &config.model,
            &api_key,
            Some(config.model_provider.base_url.as_str()),
        )?;

        let mut tool_router = ToolRouter::new();

        // Rebuild messages from events
        let mut messages = Vec::new();
        messages.push(Message::system(build_system_prompt(&config)));

        let events = get_events(&entries);
        for event_msg in events {
            match event_msg {
                EventMsg::UserMessage(e) => {
                    messages.push(Message::user(&e.message));
                }
                EventMsg::AgentMessage(e) => {
                    messages.push(Message::assistant(&e.message));
                }
                EventMsg::UndoCompleted(e) => {
                    if e.success {
                        // Replicate handle_undo logic
                        while let Some(msg) = messages.last() {
                            if matches!(msg.role, crate::client::MessageRole::User) {
                                break;
                            }
                            messages.pop();
                        }
                        if let Some(msg) = messages.last()
                            && matches!(msg.role, crate::client::MessageRole::User)
                        {
                            messages.pop();
                        }
                    }
                }
                _ => {}
            }
        }

        // Initialize rollout recorder
        let mut recorder = RolloutRecorder::new(&config.cortex_home, conversation_id)?;
        recorder.init()?;

        // Initialize LSP integration
        let lsp = Arc::new(crate::integrations::LspIntegration::new(true));
        let lsp_clone = lsp.clone();
        let cwd_clone = config.cwd.clone();
        tokio::spawn(async move {
            if let Err(e) = lsp_clone.init(&cwd_clone).await {
                warn!("Failed to initialize LSP in resumed session: {}", e);
            }
        });

        tool_router.set_lsp(lsp.clone());

        let cancelled = Arc::new(AtomicBool::new(false));

        let session = Self {
            conversation_id,
            client,
            tool_router,
            messages,
            submission_rx,
            event_tx,
            turn_id: 0,
            total_usage: TokenUsage::default(),
            token_counter: Arc::new(TokenCounter),
            running: true,
            // Resume session: load profile if configured
            current_agent_profile: {
                if let Some(agent_name) = &config.current_agent {
                    let profiles = crate::agent::AgentProfile::load_all().unwrap_or_default();
                    profiles.get(agent_name).cloned()
                } else {
                    None
                }
            },
            config,
            recorder: Some(recorder),
            pending_approvals: std::collections::HashMap::new(),
            cancelled: cancelled.clone(),
            snapshot_manager: Arc::new(crate::tasks::snapshot::SnapshotManager::new(50)),
            undo_history: crate::tasks::undo::UndoHistory::new(50),
            redo_history: crate::tasks::undo::RedoHistory::new(50),
            current_undo_actions: Vec::new(),
            share_service: crate::share_service::ShareService::new(),
            lsp,
        };

        let handle = SessionHandle {
            submission_tx,
            event_rx,
            conversation_id,
            cancelled,
        };

        Ok((session, handle))
    }

    /// Fork a session from an existing conversation.
    pub fn fork(
        config: Config,
        original_conversation_id: ConversationId,
        message_index: usize,
    ) -> Result<(Self, SessionHandle)> {
        let (submission_tx, submission_rx) = unbounded();
        let (event_tx, event_rx) = unbounded();

        let new_conversation_id = ConversationId::new();

        // Get API key
        let api_key = std::env::var("OPENAI_API_KEY")
            .or_else(|_| std::env::var("ANTHROPIC_API_KEY"))
            .or_else(|_| std::env::var("OPENROUTER_API_KEY"))
            .unwrap_or_default();

        let client = create_client(
            &config.model_provider_id,
            &config.model,
            &api_key,
            Some(config.model_provider.base_url.as_str()),
        )?;

        let mut tool_router = ToolRouter::new();

        // Rebuild messages from original rollout up to index
        let rollout_path = get_rollout_path(&config.cortex_home, &original_conversation_id);
        let entries = read_rollout(&rollout_path)?;

        let mut messages = Vec::new();
        messages.push(Message::system(build_system_prompt(&config)));

        let events = get_events(&entries);

        // Count messages (User + Assistant) to decide where to stop
        let mut count = 0;
        for event_msg in events {
            let mut added = false;
            match event_msg {
                EventMsg::UserMessage(e) => {
                    messages.push(Message::user(&e.message));
                    added = true;
                }
                EventMsg::AgentMessage(e) => {
                    messages.push(Message::assistant(&e.message));
                    added = true;
                }
                _ => {}
            }
            if added {
                if count >= message_index {
                    break;
                }
                count += 1;
            }
        }

        // Initialize new rollout recorder
        let mut recorder = RolloutRecorder::new(&config.cortex_home, new_conversation_id)?;
        recorder.init()?;

        // Record session metadata
        let meta = SessionMeta {
            id: new_conversation_id,
            parent_id: Some(original_conversation_id),
            fork_point: Some(message_index.to_string()),
            timestamp: Utc::now().to_rfc3339(),
            cwd: config.cwd.clone(),
            model: config.model.clone(),
            cli_version: env!("CARGO_PKG_VERSION").to_string(),
            instructions: config.user_instructions.clone(),
        };
        recorder.record_meta(&meta)?;

        // Record the copied events to the new rollout
        let mut user_msg_count = 0;
        for msg in &messages {
            match msg.role {
                crate::client::MessageRole::User => {
                    user_msg_count += 1;
                    if let Some(text) = msg.content.as_text() {
                        recorder.record_event(&Event {
                            id: "0".to_string(),
                            msg: EventMsg::UserMessage(UserMessageEvent {
                                id: None,
                                parent_id: None,
                                message: text.to_string(),
                                images: None,
                            }),
                        })?;
                    }
                }
                crate::client::MessageRole::Assistant => {
                    if let Some(text) = msg.content.as_text() {
                        recorder.record_event(&Event {
                            id: "0".to_string(),
                            msg: EventMsg::AgentMessage(AgentMessageEvent {
                                id: None,
                                parent_id: None,
                                message: text.to_string(),
                            }),
                        })?;
                    }
                }
                _ => {}
            }
        }

        // Initialize snapshot manager
        let snapshot_dir = config
            .cortex_home
            .join("snapshots")
            .join(new_conversation_id.to_string());
        let snapshot_manager =
            Arc::new(crate::tasks::snapshot::SnapshotManager::new(50).with_storage(snapshot_dir));

        // Initialize LSP integration
        let lsp = Arc::new(crate::integrations::LspIntegration::new(true));
        let lsp_clone = lsp.clone();
        let cwd_clone = config.cwd.clone();
        tokio::spawn(async move {
            if let Err(e) = lsp_clone.init(&cwd_clone).await {
                warn!("Failed to initialize LSP in forked session: {}", e);
            }
        });

        tool_router.set_lsp(lsp.clone());

        let cancelled = Arc::new(AtomicBool::new(false));

        let session = Self {
            conversation_id: new_conversation_id,
            client,
            tool_router,
            messages,
            submission_rx,
            event_tx,
            turn_id: user_msg_count as u64,
            total_usage: TokenUsage::default(),
            token_counter: Arc::new(TokenCounter),
            running: true,
            current_agent_profile: {
                if let Some(agent_name) = &config.current_agent {
                    let profiles = crate::agent::AgentProfile::load_all().unwrap_or_default();
                    profiles.get(agent_name).cloned()
                } else {
                    None
                }
            },
            config,
            recorder: Some(recorder),
            pending_approvals: std::collections::HashMap::new(),
            cancelled: cancelled.clone(),
            snapshot_manager,
            undo_history: crate::tasks::undo::UndoHistory::new(50),
            redo_history: crate::tasks::undo::RedoHistory::new(50),
            current_undo_actions: Vec::new(),
            share_service: crate::share_service::ShareService::new(),
            lsp,
        };

        let handle = SessionHandle {
            submission_tx,
            event_rx,
            conversation_id: session.conversation_id,
            cancelled,
        };

        Ok((session, handle))
    }
}

/// List available sessions.
pub fn list_sessions(cortex_home: &PathBuf) -> Result<Vec<SessionInfo>> {
    let sessions_dir = cortex_home.join(SESSIONS_SUBDIR);

    if !sessions_dir.exists() {
        return Ok(Vec::new());
    }

    let mut sessions = Vec::new();

    for entry in std::fs::read_dir(&sessions_dir)? {
        let entry = entry?;
        let path = entry.path();

        if path.extension().is_some_and(|e| e == "jsonl")
            && let Ok(entries) = read_rollout(&path)
            && let Some(meta) = get_session_meta(&entries)
        {
            let cwd = PathBuf::from(&meta.cwd);
            let git_branch = get_git_branch_for_dir(&cwd);
            sessions.push(SessionInfo {
                id: meta.id.clone(),
                timestamp: meta.timestamp.clone(),
                model: meta.model.clone(),
                cwd,
                message_count: entries
                    .iter()
                    .filter(|e| matches!(e.item, RolloutItem::EventMsg(EventMsg::UserMessage(_))))
                    .count(),
                git_branch,
            });
        }
    }

    // Sort by timestamp (newest first)
    sessions.sort_by(|a, b| b.timestamp.cmp(&a.timestamp));

    Ok(sessions)
}

/// Session info for listing.
#[derive(Debug, Clone)]
pub struct SessionInfo {
    pub id: String,
    pub timestamp: String,
    pub model: Option<String>,
    pub cwd: PathBuf,
    pub message_count: usize,
    pub git_branch: Option<String>,
}

/// Get git branch for a directory.
fn get_git_branch_for_dir(dir: &PathBuf) -> Option<String> {
    if !dir.exists() {
        return None;
    }
    std::process::Command::new("git")
        .args(["rev-parse", "--abbrev-ref", "HEAD"])
        .current_dir(dir)
        .output()
        .ok()
        .and_then(|o| {
            if o.status.success() {
                String::from_utf8(o.stdout)
                    .ok()
                    .map(|s| s.trim().to_string())
            } else {
                None
            }
        })
}

/// System prompt for the Cortex Agent - loaded from cortex_prompt.txt
const SYSTEM_PROMPT: &str = include_str!("../../cortex_prompt.txt");

/// Build the system prompt for the agent.
fn build_system_prompt(config: &Config) -> String {
    build_system_prompt_with_skills(config, None)
}

/// Build the system prompt with optional skills section.
fn build_system_prompt_with_skills(config: &Config, skills_section: Option<&str>) -> String {
    let cwd = config.cwd.display().to_string();
    let user_instructions = config.user_instructions.as_deref().unwrap_or("");

    // Get system info
    let system_info = get_system_info();
    let current_date = chrono::Utc::now().format("%Y-%m-%d").to_string();

    // Build environment context
    let env_context = "# The commands below were executed at the start of all sessions to gather context about the environment.\n\
         # You do not need to repeat them, unless you think the environment has changed.\n\
         # Remember: They are not necessarily related to the current conversation, but may be useful for context.".to_string();

    // Replace template variables
    let mut prompt = if let Some(agent_name) = &config.current_agent {
        // Try to load the agent to get its custom prompt
        let mut p = format!("You are the {} agent. ", agent_name) + SYSTEM_PROMPT;

        // Try project-level agent first
        let project_agent_path = config
            .cwd
            .join(".cortex")
            .join("agents")
            .join(format!("{}.md", agent_name));
        let user_agent_path = config
            .cortex_home
            .join("agents")
            .join(format!("{}.md", agent_name));

        let path_to_try = if project_agent_path.exists() {
            Some(project_agent_path)
        } else if user_agent_path.exists() {
            Some(user_agent_path)
        } else {
            None
        };

        if let Some(path) = path_to_try
            && let Ok(content) = std::fs::read_to_string(path)
        {
            // If it starts with frontmatter, try to parse it
            if content.starts_with("---") {
                if let Ok((_meta, agent_prompt)) = crate::agents::parse_agent_md(&content) {
                    p = agent_prompt;
                }
            } else {
                p = content;
            }
        }
        p
    } else {
        SYSTEM_PROMPT.to_string()
    };

    prompt = prompt.replace("{{SYSTEM_INFO}}", &system_info);
    prompt = prompt.replace("{{MODEL_NAME}}", &config.model);
    prompt = prompt.replace("{{CURRENT_DATE}}", &current_date);
    prompt = prompt.replace("{{CWD}}", &cwd);
    prompt = prompt.replace("{{ENVIRONMENT_CONTEXT}}", &env_context);

    // Load AGENTS.md instructions
    let agents_instructions = load_agents_md(config);

    // Additional context (user instructions + AGENTS.md)
    let mut additional = String::new();

    if !agents_instructions.is_empty() {
        additional.push_str("## Project Instructions (from AGENTS.md)\n");
        additional.push_str(&agents_instructions);
        additional.push_str("\n\n");
    }

    if !user_instructions.is_empty() {
        additional.push_str("## User Instructions\n");
        additional.push_str(user_instructions);
        additional.push('\n');
    }

    // Add skills section if provided
    if let Some(skills) = skills_section
        && !skills.is_empty()
    {
        additional.push_str("\n## Available Skills\n");
        additional.push_str(skills);
        additional.push('\n');
    }

    prompt = prompt.replace("{{ADDITIONAL_CONTEXT}}", &additional);

    prompt
}

/// Load and merge AGENTS.md files.
/// Order: ~/.cortex/AGENTS.md -> repo root -> directories down to CWD
/// AGENTS.override.md replaces instead of merging.
fn load_agents_md(config: &Config) -> String {
    let mut instructions = Vec::new();

    // 1. Global AGENTS.md from ~/.cortex/
    let global_path = config.cortex_home.join("AGENTS.md");
    if let Ok(content) = std::fs::read_to_string(&global_path) {
        instructions.push(content);
    }

    // 2. Find git root or use cwd
    let repo_root = find_git_root(&config.cwd).unwrap_or_else(|| config.cwd.clone());

    // 3. Walk from repo root to cwd, collecting AGENTS.md files
    let _current = repo_root.clone();
    let cwd = &config.cwd;

    // Collect all directories from root to cwd
    let mut dirs_to_check = vec![repo_root.clone()];
    if let Ok(relative) = cwd.strip_prefix(&repo_root) {
        let mut path = repo_root.clone();
        for component in relative.components() {
            path = path.join(component);
            dirs_to_check.push(path.clone());
        }
    }

    for dir in dirs_to_check {
        // Check for AGENTS.override.md first (replaces all previous)
        let override_path = dir.join("AGENTS.override.md");
        if let Ok(content) = std::fs::read_to_string(&override_path) {
            instructions.clear();
            instructions.push(content);
            continue;
        }

        // Regular AGENTS.md (merges)
        let agents_path = dir.join("AGENTS.md");
        if let Ok(content) = std::fs::read_to_string(&agents_path) {
            instructions.push(content);
        }
    }

    instructions.join("\n\n---\n\n")
}

/// Find git repository root.
fn find_git_root(start: &PathBuf) -> Option<PathBuf> {
    let mut current = start.clone();
    loop {
        if current.join(".git").exists() {
            return Some(current);
        }
        if !current.pop() {
            return None;
        }
    }
}

/// Get system information string.
fn get_system_info() -> String {
    let os = std::env::consts::OS;
    let arch = std::env::consts::ARCH;

    #[cfg(target_os = "linux")]
    let kernel = std::process::Command::new("uname")
        .arg("-r")
        .output()
        .ok()
        .and_then(|o| String::from_utf8(o.stdout).ok())
        .map(|s| s.trim().to_string())
        .unwrap_or_default();

    #[cfg(not(target_os = "linux"))]
    let kernel = String::new();

    if kernel.is_empty() {
        format!("{os} {arch}")
    } else {
        format!("{os} {arch} ({kernel})")
    }
}

/// Load skills section for system prompt.
/// Returns None if SkillsManager is not initialized or no skills are available.
fn load_skills_section(cwd: &PathBuf) -> Option<String> {
    // Try to get the global skills manager
    let manager = try_get_skills_manager()?;

    // Load skills for current working directory (synchronously)
    let outcome = manager.skills_for_cwd_sync(cwd, false);

    if outcome.skills.is_empty() {
        return None;
    }

    // Render skills section for system prompt
    render_skills_section(&outcome.skills)
}