use std::collections::HashMap;
use std::sync::Arc;

use tokio::sync::{Mutex, mpsc, oneshot};

use super::persistence::{
    append_persisted_messages, append_persisted_messages_with_meta, finalize_turn_after_persistence,
};
use super::turn::{TurnBlock, TurnController, TurnSnapshot};
use super::turn_input::prepare_turn_input;
use crate::bus::{
    ChannelContext, ChatMessage, CompletionStatus, InboundMessage, MediaItem, MediaRef,
    MessageSource, OutboundMessage, SourceKind,
};
use crate::mcp::get_mcp_status;
use crate::storage::{Storage, StorageError};
use std::sync::Arc as StdArc;

pub(super) type MessagePersistSnapshot = (
    StdArc<Storage>,
    String,
    crate::storage::message::MessageMeta,
    crate::storage::session::SessionMeta,
);

const SESSION_QUEUE_CAPACITY: usize = 32;

fn outbound_session_metadata(session_id: &str) -> HashMap<String, String> {
    HashMap::from([("_session_id".to_string(), session_id.to_string())])
}

fn outbound_turn_metadata(
    session_id: &str,
    private_context: &HashMap<String, String>,
) -> HashMap<String, String> {
    let mut metadata = private_context.clone();
    metadata.insert("_session_id".to_string(), session_id.to_string());
    metadata
}

fn committed_turn_delta(
    session_id: &str,
    messages: Vec<crate::storage::message::MessageMeta>,
) -> crate::bus::CommittedTurnDelta {
    let history_revision = messages.last().map_or(0, |message| message.seq);
    let messages = messages
        .into_iter()
        .map(|message| crate::bus::CommittedMessage {
            id: message.id,
            seq: message.seq,
            role: message.role,
            content: message.content,
            reasoning_content: message.reasoning_content,
            completion_status: message.completion_status,
            media_refs: message
                .media_refs
                .and_then(|refs| serde_json::from_str(&refs).ok())
                .unwrap_or_default(),
            created_at: message.created_at,
            tool_call_id: message.tool_call_id,
            tool_name: message.tool_name,
            tool_calls: message
                .tool_calls
                .and_then(|calls| serde_json::from_str(&calls).ok()),
        })
        .collect();
    crate::bus::CommittedTurnDelta {
        session_id: session_id.to_string(),
        history_revision,
        messages,
    }
}

tokio::task_local! {
    pub(super) static CURRENT_SOURCE_SESSION: Option<String>;
    pub(super) static CURRENT_TURN_ID: Option<String>;
    pub(super) static CURRENT_TURN_DELIVERIES: Option<Arc<std::sync::Mutex<Vec<PendingTurnDelivery>>>>;
}

pub(super) struct PendingTurnDelivery {
    pub content: String,
    pub media: Vec<MediaItem>,
}

fn take_pending_turn_deliveries(
    deliveries: &Arc<std::sync::Mutex<Vec<PendingTurnDelivery>>>,
) -> Vec<PendingTurnDelivery> {
    std::mem::take(
        &mut *deliveries
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()),
    )
}

fn take_current_turn_deliveries() -> Vec<PendingTurnDelivery> {
    CURRENT_TURN_DELIVERIES
        .try_with(|deliveries| {
            deliveries
                .as_ref()
                .map(take_pending_turn_deliveries)
                .unwrap_or_default()
        })
        .unwrap_or_default()
}

fn collapse_pending_turn_deliveries(pending: Vec<PendingTurnDelivery>) -> (String, Vec<MediaRef>) {
    let fallback_content = pending
        .iter()
        .map(|delivery| delivery.content.trim())
        .filter(|content| !content.is_empty())
        .collect::<Vec<_>>()
        .join("\n\n");
    let mut media_refs = Vec::new();
    for media_ref in pending
        .into_iter()
        .flat_map(|delivery| delivery.media.into_iter().map(|media| media.to_media_ref()))
    {
        if !media_refs.iter().any(|existing: &MediaRef| {
            existing.path == media_ref.path && existing.media_type == media_ref.media_type
        }) {
            media_refs.push(media_ref);
        }
    }
    (fallback_content, media_refs)
}

fn attach_pending_to_message(
    message: &mut ChatMessage,
    fallback_content: &str,
    media_refs: &[MediaRef],
) {
    if message.content.trim().is_empty() && !fallback_content.is_empty() {
        message.content = fallback_content.to_string();
    }
    for media_ref in media_refs {
        if !message.media_refs.iter().any(|existing| {
            existing.path == media_ref.path && existing.media_type == media_ref.media_type
        }) {
            message.media_refs.push(media_ref.clone());
        }
    }
}

fn attach_pending_turn_deliveries(
    result: &mut crate::agent::AgentProcessResult,
    pending: Vec<PendingTurnDelivery>,
) {
    let (fallback_content, media_refs) = collapse_pending_turn_deliveries(pending);
    if media_refs.is_empty() && fallback_content.is_empty() {
        return;
    }

    attach_pending_to_message(&mut result.final_response, &fallback_content, &media_refs);

    if let Some(final_message) = result
        .emitted_messages
        .iter_mut()
        .rev()
        .find(|message| message.id == result.final_response.id)
    {
        final_message
            .content
            .clone_from(&result.final_response.content);
        final_message
            .media_refs
            .clone_from(&result.final_response.media_refs);
    }
}

fn partial_assistant_with_pending_deliveries(
    snapshot: &TurnSnapshot,
    completion_status: CompletionStatus,
    pending: Vec<PendingTurnDelivery>,
) -> Option<ChatMessage> {
    let (fallback_content, media_refs) = collapse_pending_turn_deliveries(pending);
    let mut message = partial_assistant_message(snapshot, completion_status).or_else(|| {
        (!media_refs.is_empty()).then(|| {
            let mut message = ChatMessage::assistant("");
            message.id = snapshot.message_id.clone();
            message.turn_id = Some(snapshot.id.0.clone());
            message.completion_status = completion_status;
            message
        })
    })?;
    attach_pending_to_message(&mut message, &fallback_content, &media_refs);
    Some(message)
}

/// Result of handling a message - either an AI response or a command output
pub enum HandleResult {
    /// AI response to be sent as AssistantResponse
    AgentResponse(String),
    /// Command output to be sent as CommandExecuted
    CommandOutput(String),
    /// Agent processing spawned in background; response will be sent via bus
    AgentProcessing,
}
use crate::agent::context_compressor::ContextCompressionConfig;
use crate::agent::system_prompt::build_system_prompt;
use crate::agent::{AgentError, AgentLoop, AgentTurnContext, ContextCompressor, TurnEmitter};
use crate::channels::slash_command::parse_slash_command;
use crate::config::BrowserConfig;
use crate::config::LLMProviderConfig;
use crate::delivery::{TurnDeliveryHandle, TurnDeliveryService};

/// Check if an LLM error message indicates a context window overflow.
fn is_context_overflow_error(msg: &str) -> bool {
    let lower = msg.to_lowercase();
    lower.contains("context length")
        || lower.contains("context window")
        || lower.contains("maximum context")
        || lower.contains("too many tokens")
        || lower.contains("token limit exceeded")
        || lower.contains("prompt is too long")
        || lower.contains("input is too long")
}

fn partial_assistant_message(
    snapshot: &TurnSnapshot,
    completion_status: CompletionStatus,
) -> Option<ChatMessage> {
    let mut assistant_segments = Vec::new();
    let mut reasoning_segments = Vec::new();
    let mut last_iteration = None;
    for block in &snapshot.blocks {
        match block {
            TurnBlock::Assistant {
                iteration, text, ..
            } if !text.is_empty() => {
                assistant_segments.push(text.as_str());
                last_iteration = Some(*iteration);
            }
            TurnBlock::Reasoning { text, .. } if !text.is_empty() => {
                reasoning_segments.push(text.as_str());
            }
            _ => {}
        }
    }
    if assistant_segments.is_empty() {
        return None;
    }

    let mut message = ChatMessage::assistant(assistant_segments.join("\n\n"));
    message.id = snapshot.message_id.clone();
    message.turn_id = Some(snapshot.id.0.clone());
    message.iteration = last_iteration;
    message.completion_status = completion_status;
    message.reasoning_content =
        (!reasoning_segments.is_empty()).then(|| reasoning_segments.join("\n\n"));
    Some(message)
}

fn terminal_fallback_content(snapshot: &TurnSnapshot) -> Option<String> {
    partial_assistant_message(snapshot, CompletionStatus::Interrupted)
        .map(|message| message.content)
        .or_else(|| {
            (snapshot.status == super::turn::TurnStatus::Failed)
                .then(|| "The response could not be delivered. Please try again.".to_string())
        })
}

async fn deliver_terminal_fallback(
    handle: TurnDeliveryHandle,
    bus: &MessageBus,
    target: &crate::channels::TurnTarget,
    controller: &TurnController,
) {
    let Err(error) = handle.wait().await else {
        return;
    };
    tracing::error!(channel = %target.channel, chat_id = %target.chat_id, error = %error, "Turn sink terminal delivery failed; using ordinary outbound fallback");
    let snapshot = controller.snapshot();
    let Some(content) = terminal_fallback_content(&snapshot) else {
        return;
    };
    let outbound = OutboundMessage {
        channel: target.channel.clone(),
        chat_id: target.chat_id.clone(),
        content,
        reply_to: target.reply_to.clone(),
        media: vec![],
        metadata: target.metadata.clone(),
        delivery: None,
    };
    if let Err(fallback_error) = bus.deliver_outbound(outbound).await {
        tracing::error!(channel = %target.channel, chat_id = %target.chat_id, error = %fallback_error, "Ordinary terminal fallback delivery failed");
    }
}

async fn fail_turn_with_partial(
    controller: &TurnController,
    session: &Arc<Mutex<Session>>,
    error: String,
) {
    let snapshot = controller.snapshot();
    let partial = partial_assistant_with_pending_deliveries(
        &snapshot,
        CompletionStatus::Interrupted,
        take_current_turn_deliveries(),
    );
    if let Some(partial) = partial {
        controller.begin_finalizing();
        if let Err(persistence_error) = append_persisted_messages(session, vec![partial]).await {
            controller.fail(format!(
                "{error}; failed to persist interrupted turn: {persistence_error}"
            ));
            return;
        }
    }
    controller.fail(error);
}

#[cfg(test)]
mod cancelled_partial_tests {
    use super::*;
    use crate::agent::TurnEvent;
    use crate::bus::{MessageBus, OutboundDispatcher};
    use crate::channels::{Channel, ChannelError, ChannelManager, CliChatChannel};
    use crate::delivery::{ConversationWriteLocks, DeliveryError};
    use crate::task_supervisor::TaskSupervisor;

    #[derive(Default)]
    struct RecordingChannel {
        messages: Mutex<Vec<String>>,
    }

    #[async_trait::async_trait]
    impl Channel for RecordingChannel {
        fn name(&self) -> &str {
            "recording"
        }

        fn is_running(&self) -> bool {
            true
        }

        async fn start(&self, _bus: Arc<MessageBus>) -> Result<(), ChannelError> {
            Ok(())
        }

        async fn stop(&self) -> Result<(), ChannelError> {
            Ok(())
        }

        async fn send(&self, message: OutboundMessage) -> Result<(), ChannelError> {
            self.messages.lock().await.push(message.content);
            Ok(())
        }
    }

    #[test]
    fn turn_metadata_preserves_channel_cleanup_fields() {
        let forwarded = HashMap::from([
            ("feishu.message_id".to_string(), "message-1".to_string()),
            ("feishu.reaction_id".to_string(), "reaction-1".to_string()),
            ("_session_id".to_string(), "stale".to_string()),
        ]);

        let metadata = outbound_turn_metadata("session-1", &forwarded);

        assert_eq!(
            metadata.get("_session_id").map(String::as_str),
            Some("session-1")
        );
        assert_eq!(
            metadata.get("feishu.message_id").map(String::as_str),
            Some("message-1")
        );
        assert_eq!(
            metadata.get("feishu.reaction_id").map(String::as_str),
            Some("reaction-1")
        );
    }

    #[test]
    fn visible_partial_text_becomes_cancelled_persisted_message() {
        let (controller, emitter, _) = TurnController::start("session", "message-id");
        emitter
            .emit(TurnEvent::ReasoningDelta {
                iteration: 0,
                delta: "reason".into(),
            })
            .unwrap();
        emitter
            .emit(TurnEvent::TextDelta {
                iteration: 0,
                delta: "first".into(),
            })
            .unwrap();
        emitter
            .emit(TurnEvent::TextSegmentFinished { iteration: 0 })
            .unwrap();
        emitter
            .emit(TurnEvent::TextDelta {
                iteration: 1,
                delta: "second".into(),
            })
            .unwrap();

        let message =
            partial_assistant_message(&controller.snapshot(), CompletionStatus::Cancelled).unwrap();
        assert_eq!(message.id, "message-id");
        assert_eq!(message.content, "first\n\nsecond");
        assert_eq!(message.reasoning_content.as_deref(), Some("reason"));
        assert_eq!(message.iteration, Some(1));
        assert_eq!(message.completion_status, CompletionStatus::Cancelled);
    }

    #[test]
    fn failed_terminal_without_visible_text_has_safe_fallback() {
        let (controller, _emitter, _) = TurnController::start("session", "message-id");
        controller.fail("provider response contained a secret");

        assert_eq!(
            terminal_fallback_content(&controller.snapshot()).as_deref(),
            Some("The response could not be delivered. Please try again.")
        );
    }

    #[tokio::test]
    async fn asynchronous_terminal_failure_uses_one_ordinary_fallback() {
        let bus = MessageBus::new(8);
        let cli = Arc::new(CliChatChannel::new());
        let channels = ChannelManager::with_bus(cli, bus.clone());
        let channel = Arc::new(RecordingChannel::default());
        channels
            .register_channel("recording", channel.clone())
            .await;
        let supervisor = TaskSupervisor::new();
        let dispatcher = OutboundDispatcher::new(
            bus.clone(),
            channels,
            supervisor.clone(),
            ConversationWriteLocks::default(),
            Arc::new(std::sync::atomic::AtomicUsize::new(0)),
        );
        let dispatcher_task = tokio::spawn(async move { dispatcher.run().await });

        let (controller, emitter, _) = TurnController::start("session", "message-id");
        emitter
            .emit(TurnEvent::TextDelta {
                iteration: 0,
                delta: "completed response".into(),
            })
            .unwrap();
        controller.complete(None);
        let (sender, completion) = oneshot::channel();
        sender.send(Err(DeliveryError::FinalTimedOut)).unwrap();

        let target = crate::channels::TurnTarget {
            channel: "recording".to_string(),
            chat_id: "chat".to_string(),
            session_id: "session".to_string(),
            reply_to: None,
            metadata: HashMap::new(),
        };
        deliver_terminal_fallback(
            TurnDeliveryHandle { completion },
            &bus,
            &target,
            &controller,
        )
        .await;

        assert_eq!(
            channel.messages.lock().await.as_slice(),
            &["completed response"]
        );
        dispatcher_task.abort();
        let _ = dispatcher_task.await;
        supervisor.shutdown(std::time::Duration::from_secs(1)).await;
    }

    #[test]
    fn reasoning_only_cancel_does_not_create_assistant_history() {
        let (controller, emitter, _) = TurnController::start("session", "message-id");
        emitter
            .emit(TurnEvent::ReasoningDelta {
                iteration: 0,
                delta: "private".into(),
            })
            .unwrap();
        assert!(
            partial_assistant_message(&controller.snapshot(), CompletionStatus::Cancelled,)
                .is_none()
        );
    }

    #[test]
    fn pending_same_turn_media_is_attached_to_the_final_response() {
        let final_message = ChatMessage::assistant("截图已经准备好了");
        let final_id = final_message.id.clone();
        let mut result = crate::agent::AgentProcessResult {
            final_response: final_message.clone(),
            emitted_messages: vec![
                ChatMessage::tool("call-1", "send_message", "附件已加入当前回复"),
                final_message,
            ],
            total_tokens: None,
            usage: None,
        };

        attach_pending_turn_deliveries(
            &mut result,
            vec![PendingTurnDelivery {
                content: "这是百度首页截图".to_string(),
                media: vec![MediaItem::new("/tmp/baidu.png", "image")],
            }],
        );

        assert_eq!(result.final_response.content, "截图已经准备好了");
        assert_eq!(result.final_response.media_refs.len(), 1);
        assert_eq!(result.final_response.media_refs[0].path, "/tmp/baidu.png");
        let committed_final = result
            .emitted_messages
            .iter()
            .find(|message| message.id == final_id)
            .unwrap();
        assert_eq!(committed_final.media_refs.len(), 1);
        assert_eq!(committed_final.media_refs[0].path, "/tmp/baidu.png");
    }

    #[test]
    fn pending_media_survives_a_turn_without_partial_text() {
        let (controller, _emitter, _) = TurnController::start("session", "message-id");

        let message = partial_assistant_with_pending_deliveries(
            &controller.snapshot(),
            CompletionStatus::Interrupted,
            vec![PendingTurnDelivery {
                content: "这是已生成的截图".to_string(),
                media: vec![MediaItem::new("/tmp/baidu.png", "image")],
            }],
        )
        .unwrap();

        assert_eq!(message.id, "message-id");
        assert_eq!(message.content, "这是已生成的截图");
        assert_eq!(message.completion_status, CompletionStatus::Interrupted);
        assert_eq!(message.media_refs.len(), 1);
    }
}
use crate::bus::MessageBus;
use crate::providers::{LLMProvider, create_provider};
use crate::session::events::DialogInfo;
use crate::session::session_id::UnifiedSessionId;
use crate::skills::SkillsLoader;
use crate::tools::OutboundMessenger;
use crate::tools::SendMessageTool;
use crate::tools::{ToolRegistry, create_default_tools};

/// Session = 一个 dialog
/// 每个 Session 对应一个 UnifiedSessionId，有独立的 messages history
pub struct Session {
    pub id: UnifiedSessionId,
    pub title: String,
    pub created_at: i64,
    pub last_active_at: i64,
    pub message_count: i64,
    pub total_message_count: i64,

    messages: Vec<ChatMessage>,
    seq_counter: i64,

    provider_config: LLMProviderConfig,
    provider: Arc<dyn LLMProvider>,
    tools: Arc<ToolRegistry>,
    compressor: ContextCompressor,

    storage: Option<StdArc<Storage>>,
    routing_info: String,
    archived_at: Option<i64>,
    /// Timestamp (Unix ms) of the last consolidation.
    /// Messages before this time have been compressed into memory.
    pub last_consolidated_at: Option<i64>,
    pub last_compressed_message_at: Option<i64>,
    memory_manager: Arc<crate::memory::MemoryManager>,

    /// Task queue for per-session serial agent processing
    agent_tx: Option<mpsc::Sender<AgentTask>>,
    /// Cancel signal for the currently executing agent task
    current_cancel: Option<oneshot::Sender<()>>,
    active_turn_emitter: Option<ActiveTurnEmitter>,
    /// Monotonic counter to detect stale workers
    worker_generation: u64,
    /// Prevents duplicate background title requests while the title is still default.
    title_generation_in_flight: bool,
    /// Monotonic counter for in-memory session mutations.
    ///
    /// Slow work such as memory recall, compression, and title generation runs
    /// outside the session lock. Workers capture this version before starting
    /// that work and verify it before committing results, so stale snapshots do
    /// not overwrite a session that was changed by a command such as /clear or
    /// /delete while the slow work was in flight.
    state_version: u64,
    /// Serializes durable mutations while allowing the session state mutex to
    /// be released during SQLite I/O.
    pub(super) persistence_lock: Arc<Mutex<()>>,
}

struct ActiveTurnEmitter {
    turn_id: String,
    emitter: TurnEmitter,
}

/// A task to be processed by the per-session agent worker
struct AgentTask {
    channel: String,
    sender_id: String,
    chat_id: String,
    content: String,
    received_at: i64,
    media: Vec<MediaItem>,
    channel_context: ChannelContext,
}

#[derive(Clone)]
struct AgentWorkerDeps {
    bus: Arc<MessageBus>,
    memory_manager: Arc<crate::memory::MemoryManager>,
    work_manager: Arc<crate::work::WorkManager>,
    skills_loader: Arc<SkillsLoader>,
    task_supervisor: crate::task_supervisor::TaskSupervisor,
    turn_delivery: TurnDeliveryService,
}

impl Session {
    pub async fn new(
        id: UnifiedSessionId,
        provider_config: LLMProviderConfig,
        tools: Arc<ToolRegistry>,
        storage: Option<StdArc<Storage>>,
        routing_info: String,
        title: String,
        memory_manager: Arc<crate::memory::MemoryManager>,
    ) -> Result<Self, AgentError> {
        let mut provider_box = create_provider(provider_config.clone())
            .map_err(|e| AgentError::Other(format!("provider creation error: {}", e)))?;
        if let Some(ref s) = storage {
            provider_box.set_storage(s.clone());
        }
        let provider: Arc<dyn LLMProvider> = Arc::from(provider_box);

        let compressor_config = ContextCompressionConfig {
            protect_first_n: 2,
            ..Default::default()
        };

        let mut compressor = ContextCompressor::with_config(
            provider.clone(),
            provider_config.token_limit,
            compressor_config,
            memory_manager.clone(),
        );
        compressor.set_session_id(Some(id.to_string()));

        let now = chrono::Utc::now().timestamp_millis();

        Ok(Self {
            id: id.clone(),
            title,
            created_at: now,
            last_active_at: now,
            message_count: 0,
            total_message_count: 0,
            messages: Vec::new(),
            seq_counter: 1,
            provider_config: provider_config.clone(),
            provider: provider.clone(),
            tools,
            compressor,
            storage,
            routing_info,
            archived_at: None,
            last_consolidated_at: None,
            last_compressed_message_at: None,
            memory_manager,
            agent_tx: None,
            current_cancel: None,
            active_turn_emitter: None,
            worker_generation: 0,
            title_generation_in_flight: false,
            state_version: 0,
            persistence_lock: Arc::new(Mutex::new(())),
        })
    }

    /// 从 Storage 恢复 Session
    pub async fn from_storage(
        id: UnifiedSessionId,
        provider_config: LLMProviderConfig,
        tools: Arc<ToolRegistry>,
        storage: StdArc<Storage>,
        memory_manager: Arc<crate::memory::MemoryManager>,
    ) -> Result<Self, AgentError> {
        let session_meta = storage.get_session(&id.to_string()).await.map_err(|e| {
            AgentError::Other(format!("failed to load session from storage: {}", e))
        })?;

        let mut provider_box = create_provider(provider_config.clone())
            .map_err(|e| AgentError::Other(format!("provider creation error: {}", e)))?;
        provider_box.set_storage(storage.clone());
        let provider: Arc<dyn LLMProvider> = Arc::from(provider_box);

        let compressor_config = ContextCompressionConfig {
            protect_first_n: 2,
            ..Default::default()
        };

        let mut compressor = ContextCompressor::with_config(
            provider.clone(),
            provider_config.token_limit,
            compressor_config,
            memory_manager.clone(),
        );
        compressor.set_session_id(Some(id.to_string()));

        let mut chat_messages: Vec<ChatMessage> = Vec::new();
        let mut restored_compressed_at = session_meta.last_compressed_message_at;

        if let Some(after_ts) = session_meta.last_compressed_message_at {
            // Load last 4 timelines to detect if there are more than 3
            let timelines = storage
                .load_session_timelines(&id.to_string(), 4)
                .await
                .map_err(|e| {
                    AgentError::Other(format!("failed to load session timelines: {}", e))
                })?;

            let has_more_timelines = timelines.len() > 3;

            if has_more_timelines {
                chat_messages.push(ChatMessage::user(
                    "[Earlier conversation summaries exist. \
                     Use `timeline_recall` to search if needed.]",
                ));
            }

            // Insert latest 3 timelines as context (reversed: oldest first)
            for tl in timelines.iter().take(3).rev() {
                chat_messages.push(ChatMessage::user(format!(
                    "[Previous Context]\n{}",
                    tl.content
                )));
            }

            // Load raw messages after compressed timestamp
            let tail = storage
                .load_messages_after_timestamp(&id.to_string(), after_ts)
                .await
                .map_err(|e| {
                    AgentError::Other(format!("failed to load messages after timestamp: {}", e))
                })?;

            let mut tail_msgs: Vec<ChatMessage> = tail
                .into_iter()
                .map(|m| ChatMessage {
                    id: m.id,
                    role: m.role,
                    content: m.content,
                    reasoning_content: m.reasoning_content,
                    provider_state: m.provider_state.and_then(|state| {
                        crate::bus::ProviderReasoningState::from_json_lossy(&state)
                    }),
                    turn_id: m.turn_id,
                    iteration: m.iteration.and_then(|value| u32::try_from(value).ok()),
                    completion_status: m.completion_status,
                    media_refs: m
                        .media_refs
                        .map(|refs| serde_json::from_str(&refs).unwrap_or_default())
                        .unwrap_or_default(),
                    timestamp: m.created_at,
                    tool_call_id: m.tool_call_id,
                    tool_name: m.tool_name,
                    tool_calls: m
                        .tool_calls
                        .and_then(|tc| {
                            serde_json::from_str::<Vec<crate::providers::ToolCall>>(&tc).ok()
                        })
                        .filter(|v| !v.is_empty()),
                    source: m.source.and_then(|s| serde_json::from_str(&s).ok()),
                })
                .collect();

            repair_tool_call_chains(&mut tail_msgs);
            chat_messages.extend(tail_msgs);
        } else {
            // No prior compression — load all messages
            let messages = storage
                .load_messages(&id.to_string(), 0)
                .await
                .map_err(|e| {
                    AgentError::Other(format!("failed to load messages from storage: {}", e))
                })?;

            chat_messages = messages
                .into_iter()
                .map(|m| ChatMessage {
                    id: m.id,
                    role: m.role,
                    content: m.content,
                    reasoning_content: m.reasoning_content,
                    provider_state: m.provider_state.and_then(|state| {
                        crate::bus::ProviderReasoningState::from_json_lossy(&state)
                    }),
                    turn_id: m.turn_id,
                    iteration: m.iteration.and_then(|value| u32::try_from(value).ok()),
                    completion_status: m.completion_status,
                    media_refs: m
                        .media_refs
                        .map(|refs| serde_json::from_str(&refs).unwrap_or_default())
                        .unwrap_or_default(),
                    timestamp: m.created_at,
                    tool_call_id: m.tool_call_id,
                    tool_name: m.tool_name,
                    tool_calls: m
                        .tool_calls
                        .and_then(|tc| {
                            serde_json::from_str::<Vec<crate::providers::ToolCall>>(&tc).ok()
                        })
                        .filter(|v| !v.is_empty()),
                    source: m.source.and_then(|s| serde_json::from_str(&s).ok()),
                })
                .collect();

            repair_tool_call_chains(&mut chat_messages);
        }

        // Compress loaded history if it exceeds budget
        if !chat_messages.is_empty() {
            let result = compressor
                .compress_if_needed(chat_messages)
                .await
                .map_err(|e| AgentError::Other(format!("compression during restore: {}", e)))?;
            if result.created_timelines {
                restored_compressed_at = Some(chrono::Utc::now().timestamp_millis());
            }
            chat_messages = result.history;
        }

        // seq_counter from actual DB max
        let max_seq = storage
            .get_max_message_seq(&id.to_string())
            .await
            .map_err(|e| AgentError::Other(format!("failed to load message sequence: {}", e)))?;
        let seq_counter = max_seq + 1;
        let total_message_count = session_meta.message_count;

        Ok(Self {
            id: id.clone(),
            title: session_meta.title,
            created_at: session_meta.created_at,
            last_active_at: session_meta.last_active_at,
            message_count: session_meta.message_count,
            total_message_count,
            messages: chat_messages,
            seq_counter,
            provider_config: provider_config.clone(),
            provider: provider.clone(),
            tools,
            compressor,
            storage: Some(storage),
            routing_info: session_meta.routing_info.unwrap_or_default(),
            archived_at: session_meta.archived_at,
            last_consolidated_at: session_meta.last_consolidated_at,
            last_compressed_message_at: restored_compressed_at,
            memory_manager,
            agent_tx: None,
            current_cancel: None,
            active_turn_emitter: None,
            worker_generation: 0,
            title_generation_in_flight: false,
            state_version: 0,
            persistence_lock: Arc::new(Mutex::new(())),
        })
    }

    /// 获取 session ID
    pub fn session_id(&self) -> String {
        self.id.to_string()
    }

    pub(super) fn add_message_in_memory_with_version(
        &mut self,
        message: ChatMessage,
        persist: bool,
        advance_state_version: bool,
    ) -> Option<MessagePersistSnapshot> {
        let is_user = message.role == "user";
        let now = chrono::Utc::now().timestamp_millis();

        // Assign seq
        let seq = self.seq_counter;
        self.seq_counter += 1;

        let persist_snapshot = if persist {
            self.storage.clone().map(|storage| {
                let msg_meta = crate::storage::message::MessageMeta {
                    id: message.id.clone(),
                    session_id: self.id.to_string(),
                    seq,
                    role: message.role.clone(),
                    content: message.content.clone(),
                    reasoning_content: message.reasoning_content.clone(),
                    provider_state: message
                        .provider_state
                        .as_ref()
                        .and_then(|state| serde_json::to_string(state).ok()),
                    turn_id: message.turn_id.clone(),
                    iteration: message.iteration.map(i64::from),
                    completion_status: message.completion_status,
                    media_refs: if message.media_refs.is_empty() {
                        None
                    } else {
                        Some(serde_json::to_string(&message.media_refs).unwrap_or_default())
                    },
                    tool_call_id: message.tool_call_id.clone(),
                    tool_name: message.tool_name.clone(),
                    tool_calls: message
                        .tool_calls
                        .as_ref()
                        .and_then(|tc| serde_json::to_string(tc).ok()),
                    source: message
                        .source
                        .as_ref()
                        .map(|s| serde_json::to_string(s).unwrap_or_default()),
                    created_at: now,
                };
                (storage, self.id.to_string(), msg_meta)
            })
        } else {
            None
        };

        // Update in-memory state
        self.messages.push(message);
        self.total_message_count += 1;
        if is_user {
            self.message_count += 1;
        }
        self.last_active_at = now;
        if advance_state_version {
            self.state_version = self.state_version.wrapping_add(1);
        }

        persist_snapshot.map(|(storage, session_id, msg_meta)| {
            let session_meta = crate::storage::session::SessionMeta {
                id: session_id.clone(),
                channel: self.id.channel.clone(),
                chat_id: self.id.chat_id.clone(),
                dialog_id: self.id.dialog_id.clone(),
                title: self.title.clone(),
                created_at: self.created_at,
                last_active_at: self.last_active_at,
                message_count: self.message_count,
                routing_info: if self.routing_info.is_empty() {
                    None
                } else {
                    Some(self.routing_info.clone())
                },
                archived_at: self.archived_at,
                deleted_at: None,
                last_consolidated_at: self.last_consolidated_at,
                last_compressed_message_at: self.last_compressed_message_at,
            };
            (storage, session_id, msg_meta, session_meta)
        })
    }

    /// Roll back messages that were appended in memory but whose atomic
    /// persistence failed. This is only called while holding the session lock,
    /// so the suffix check also protects against removing unrelated messages.
    pub(super) fn rollback_message_suffix_with_version(
        &mut self,
        message_ids: &[String],
        advance_state_version: bool,
    ) {
        if message_ids.is_empty() || self.messages.len() < message_ids.len() {
            return;
        }
        let start = self.messages.len() - message_ids.len();
        if self.messages[start..]
            .iter()
            .zip(message_ids)
            .any(|(message, id)| &message.id != id)
        {
            tracing::error!(session_id = %self.id, "Refusing to roll back a non-matching message suffix");
            return;
        }

        let removed_user_messages = self.messages[start..]
            .iter()
            .filter(|message| message.role == "user")
            .count() as i64;
        self.messages.truncate(start);
        self.seq_counter -= message_ids.len() as i64;
        self.total_message_count -= message_ids.len() as i64;
        self.message_count -= removed_user_messages;
        if advance_state_version {
            self.state_version = self.state_version.wrapping_add(1);
        }
    }

    pub(super) fn owns_active_turn(&self, turn_id: &str) -> bool {
        self.active_turn_emitter
            .as_ref()
            .is_some_and(|active| active.turn_id == turn_id)
    }

    #[cfg(test)]
    pub(super) fn state_version_for_test(&self) -> u64 {
        self.state_version
    }

    #[cfg(test)]
    pub(super) fn set_active_turn_for_test(&mut self, turn_id: &str) {
        let (_controller, emitter, _) = TurnController::start(self.id.to_string(), "test-message");
        self.active_turn_emitter = Some(ActiveTurnEmitter {
            turn_id: turn_id.to_string(),
            emitter,
        });
    }

    /// 获取消息历史
    pub fn get_history(&self) -> &[ChatMessage] {
        &self.messages
    }

    pub fn create_user_message(&self, content: &str, media_refs: Vec<MediaRef>) -> ChatMessage {
        if media_refs.is_empty() {
            ChatMessage::user(content)
        } else {
            ChatMessage::user_with_media(content, media_refs)
        }
    }

    pub fn create_user_message_with_source(
        &self,
        content: &str,
        media_refs: Vec<MediaRef>,
        source: MessageSource,
    ) -> ChatMessage {
        let mut message = ChatMessage::user_with_source(content, source);
        message.media_refs = media_refs;
        message
    }

    fn session_meta_snapshot(
        &self,
    ) -> Option<(StdArc<Storage>, crate::storage::session::SessionMeta)> {
        let storage = self.storage.clone()?;
        let meta = crate::storage::session::SessionMeta {
            id: self.id.to_string(),
            channel: self.id.channel.clone(),
            chat_id: self.id.chat_id.clone(),
            dialog_id: self.id.dialog_id.clone(),
            title: self.title.clone(),
            created_at: self.created_at,
            last_active_at: self.last_active_at,
            message_count: self.message_count,
            routing_info: if self.routing_info.is_empty() {
                None
            } else {
                Some(self.routing_info.clone())
            },
            archived_at: self.archived_at,
            deleted_at: None,
            last_consolidated_at: self.last_consolidated_at,
            last_compressed_message_at: self.last_compressed_message_at,
        };
        Some((storage, meta))
    }

    /// 检查是否需要自动生成 title（5 条用户消息后）
    pub fn should_generate_title(&self) -> bool {
        self.title == "新对话" && self.message_count >= 5
    }

    fn title_prompt_snapshot(&self) -> Option<String> {
        if !self.should_generate_title() {
            return None;
        }

        Some(format!(
            r#"给定以下对话历史，生成一个简短的会话标题（5-15 个中文字符），概括这个对话的核心内容或用户的主要需求。只返回一个标题，不要解释。

历史：
{}"#,
            self.messages
                .iter()
                .filter(|m| m.role == "user" || m.role == "assistant")
                .take(20)
                .map(|m| format!("[{}]: {}", m.role, m.content))
                .collect::<Vec<_>>()
                .join("\n")
        ))
    }

    fn apply_generated_title(&mut self, title: String) -> bool {
        if title.is_empty() || !self.should_generate_title() {
            return false;
        }

        self.title = title;
        self.state_version = self.state_version.wrapping_add(1);
        true
    }

    fn fresh_context_compressor(&self) -> ContextCompressor {
        let compressor_config = ContextCompressionConfig {
            protect_first_n: 2,
            ..Default::default()
        };
        let mut compressor = ContextCompressor::with_config(
            self.provider.clone(),
            self.provider_config.token_limit,
            compressor_config,
            self.memory_manager.clone(),
        );
        compressor.set_session_id(Some(self.id.to_string()));
        compressor
    }

    fn replace_history_in_memory(&mut self, messages: Vec<ChatMessage>) {
        self.messages = messages;
        self.seq_counter = self.messages.len() as i64 + 1;
        self.total_message_count = self.messages.len() as i64;
        self.message_count = self.messages.iter().filter(|m| m.role == "user").count() as i64;
        self.last_active_at = chrono::Utc::now().timestamp_millis();
        self.state_version = self.state_version.wrapping_add(1);
    }

    /// 获取 provider_config 引用
    pub fn provider_config(&self) -> &LLMProviderConfig {
        &self.provider_config
    }

    /// 获取 compressor 引用
    pub fn compressor(&self) -> &ContextCompressor {
        &self.compressor
    }

    /// Get the compressor's current threshold for diagnostics/fallback.
    pub fn compressor_threshold(&self) -> usize {
        self.compressor.threshold()
    }

    /// 创建一个临时的 AgentLoop 实例来处理消息
    pub fn create_agent(&self) -> Result<AgentLoop, AgentError> {
        Ok(AgentLoop::with_provider_and_tools(
            self.provider.clone(),
            self.tools.clone(),
            self.provider_config.max_tool_iterations,
            self.provider_config.model_id.clone(),
            self.provider_config.workspace_dir.clone(),
            self.provider_config.input_types.clone(),
        )
        .with_context_window(self.provider_config.token_limit))
    }

    /// 构建系统提示词（包含 AgentLoop 的基础提示词 + skills + memory）
    pub fn build_system_prompt(&self, skills_prompt: &str) -> String {
        let base_prompt = build_system_prompt(
            &self.provider_config.workspace_dir,
            &self.provider_config.model_id,
            &self.tools,
        );

        if skills_prompt.trim().is_empty() {
            base_prompt
        } else {
            format!("{}\n\n{}", base_prompt, skills_prompt)
        }
    }

    /// 将当前 session 导出为 markdown 文档并保存到文件
    pub fn dump_to_file(&self, system_prompt: &str) -> std::io::Result<String> {
        use chrono::Local;
        use std::fs;
        use std::io::Write;

        let md = self.dump_as_markdown_with_system_prompt(system_prompt);

        // Create dumps directory under workspace
        let dumps_dir = self.provider_config.workspace_dir.join("dumps");
        fs::create_dir_all(&dumps_dir)?;

        // Generate filename based on session info
        let timestamp = Local::now().format("%Y%m%d_%H%M%S");
        let filename = format!("{}_{}_{}.md", self.id.channel, self.id.chat_id, timestamp);
        let filepath = dumps_dir.join(&filename);

        // Write to file
        let mut file = fs::File::create(&filepath)?;
        file.write_all(md.as_bytes())?;

        Ok(filepath.to_string_lossy().to_string())
    }

    /// 将当前 session 导出为 markdown 文档（纯内存版本）
    pub fn dump_as_markdown(&self) -> String {
        use chrono::{DateTime, Local};

        let now = Local::now().format("%Y-%m-%d %H:%M:%S");

        let mut md = String::new();
        md.push_str("# Session Dump\n\n");
        md.push_str(&format!("- **Session ID**: `{}`\n", self.id));
        md.push_str(&format!("- **Channel**: `{}`\n", self.id.channel));
        md.push_str(&format!("- **Chat ID**: `{}`\n", self.id.chat_id));
        md.push_str(&format!("- **Dialog ID**: `{}`\n", self.id.dialog_id));
        md.push_str(&format!("- **Message Count**: {}\n", self.messages.len()));
        md.push_str(&format!(
            "- **Model**: `{}`\n",
            self.provider_config.model_id
        ));
        md.push_str(&format!("- **Exported At**: {}\n", now));
        md.push_str("\n---\n\n");

        md.push_str("## Conversation History\n\n");

        for (i, msg) in self.messages.iter().enumerate() {
            let role = match msg.role.as_str() {
                "system" => "System",
                "user" => "User",
                "assistant" => "Assistant",
                "tool" => "Tool",
                r => r,
            };

            let timestamp = if msg.timestamp > 0 {
                DateTime::from_timestamp_millis(msg.timestamp)
                    .map(|dt| {
                        dt.with_timezone(&Local)
                            .format("%Y-%m-%d %H:%M:%S")
                            .to_string()
                    })
                    .unwrap_or_default()
            } else {
                String::new()
            };

            md.push_str(&format!("### [{:03}] {} {}\n\n", i + 1, role, timestamp));
            md.push_str("```\n");

            if let Some(ref tool_calls) = msg.tool_calls {
                md.push_str("[Tool Calls]\n");
                for tc in tool_calls {
                    md.push_str(&format!("- {}: {:?}\n", tc.name, tc.arguments));
                }
            }

            if let Some(ref tool_name) = msg.tool_name {
                md.push_str(&format!("[Tool: {}]\n", tool_name));
            }

            if let Some(ref tool_call_id) = msg.tool_call_id {
                md.push_str(&format!("[Tool Call ID: {}]\n", tool_call_id));
            }

            md.push_str(&msg.content);
            md.push_str("\n```\n\n");

            if !msg.media_refs.is_empty() {
                md.push_str(&format!("**Media**: {:?}\n\n", msg.media_refs));
            }
        }

        md
    }

    /// 将当前 session 导出为 markdown 文档（包含系统提示词）
    pub fn dump_as_markdown_with_system_prompt(&self, system_prompt: &str) -> String {
        use chrono::{DateTime, Local};

        let now = Local::now().format("%Y-%m-%d %H:%M:%S");

        let mut md = String::new();
        md.push_str("# Session Dump\n\n");
        md.push_str(&format!("- **Session ID**: `{}`\n", self.id));
        md.push_str(&format!("- **Channel**: `{}`\n", self.id.channel));
        md.push_str(&format!("- **Chat ID**: `{}`\n", self.id.chat_id));
        md.push_str(&format!("- **Dialog ID**: `{}`\n", self.id.dialog_id));
        md.push_str(&format!("- **Message Count**: {}\n", self.messages.len()));
        md.push_str(&format!(
            "- **Model**: `{}`\n",
            self.provider_config.model_id
        ));
        md.push_str(&format!("- **Exported At**: {}\n", now));
        md.push_str("\n---\n\n");

        // System Prompt Section
        md.push_str("## System Prompt (Injected to Model)\n\n");
        md.push_str("```\n");
        md.push_str(system_prompt);
        md.push_str("\n```\n\n");
        md.push_str("---\n\n");

        md.push_str("## Conversation History\n\n");

        for (i, msg) in self.messages.iter().enumerate() {
            let role = match msg.role.as_str() {
                "system" => "System",
                "user" => "User",
                "assistant" => "Assistant",
                "tool" => "Tool",
                r => r,
            };

            let timestamp = if msg.timestamp > 0 {
                DateTime::from_timestamp_millis(msg.timestamp)
                    .map(|dt| {
                        dt.with_timezone(&Local)
                            .format("%Y-%m-%d %H:%M:%S")
                            .to_string()
                    })
                    .unwrap_or_default()
            } else {
                String::new()
            };

            md.push_str(&format!("### [{:03}] {} {}\n\n", i + 1, role, timestamp));
            md.push_str("```\n");

            if let Some(ref tool_calls) = msg.tool_calls {
                md.push_str("[Tool Calls]\n");
                for tc in tool_calls {
                    md.push_str(&format!("- {}: {:?}\n", tc.name, tc.arguments));
                }
            }

            if let Some(ref tool_name) = msg.tool_name {
                md.push_str(&format!("[Tool: {}]\n", tool_name));
            }

            if let Some(ref tool_call_id) = msg.tool_call_id {
                md.push_str(&format!("[Tool Call ID: {}]\n", tool_call_id));
            }

            md.push_str(&msg.content);
            md.push_str("\n```\n\n");

            if !msg.media_refs.is_empty() {
                md.push_str(&format!("**Media**: {:?}\n\n", msg.media_refs));
            }
        }

        md
    }
}

/// Repair damaged tool call chains after restoring from storage.
/// Handles cases where the gateway crashed mid-loop, leaving assistant
/// tool_calls without corresponding tool result messages.
fn repair_tool_call_chains(messages: &mut [ChatMessage]) {
    let mut i = 0;
    while i < messages.len() {
        let calls = match &messages[i].tool_calls {
            Some(calls) if !calls.is_empty() => calls.clone(),
            _ => {
                i += 1;
                continue;
            }
        };

        if messages[i].role != "assistant" {
            i += 1;
            continue;
        }

        // Collect expected tool call IDs
        let expected_ids: std::collections::HashSet<&str> =
            calls.iter().map(|c| c.id.as_str()).collect();
        let expected_count = expected_ids.len();

        // Check following messages for matching tool results (same tool_call_id)
        let mut found = 0;
        let mut j = i + 1;
        while j < messages.len() && found < expected_count {
            if messages[j].role == "tool" {
                if let Some(ref tc_id) = messages[j].tool_call_id
                    && expected_ids.contains(tc_id.as_str())
                {
                    found += 1;
                }
            } else if messages[j].role == "user" || messages[j].role == "assistant" {
                // Next user/assistant message — stop scanning, chain is broken
                break;
            }
            j += 1;
        }

        if found < expected_count {
            // Incomplete chain: remove tool_calls and add interruption note
            tracing::warn!(
                found,
                expected = expected_count,
                "Repairing incomplete tool call chain — gateway restart likely interrupted execution"
            );
            let old_content = std::mem::take(&mut messages[i].content);
            messages[i].content = format!(
                "{}\n\n[Tool calls ({}): {} — execution interrupted by gateway restart]",
                old_content,
                expected_count,
                calls
                    .iter()
                    .map(|c| c.name.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            );
            messages[i].tool_calls = None;
        }

        i += 1;
    }
}

/// SessionManager 管理所有 Session，按 channel_name 路由
#[derive(Clone)]
pub struct SessionManager {
    inner: Arc<Mutex<SessionManagerInner>>,
    provider_config: LLMProviderConfig,
    tools: Arc<ToolRegistry>,
    skills_loader: Arc<SkillsLoader>,
    storage: Arc<Storage>,
    pub(super) bus: Arc<MessageBus>,
    memory_manager: Arc<crate::memory::MemoryManager>,
    work_manager: Arc<crate::work::WorkManager>,
    sub_agent_manager: Arc<crate::agent::SubAgentManager>,
    task_supervisor: crate::task_supervisor::TaskSupervisor,
    turn_delivery: TurnDeliveryService,
    reload: crate::gateway::reload::ReloadHandle,
}

/// Gateway-owned runtime services shared by all Session workers.
pub struct SessionManagerServices {
    bus: Arc<MessageBus>,
    memory_manager: Arc<crate::memory::MemoryManager>,
    task_supervisor: crate::task_supervisor::TaskSupervisor,
    turn_delivery: TurnDeliveryService,
    reload: crate::gateway::reload::ReloadHandle,
    admission: crate::gateway::reload::RuntimeAdmission,
}

impl SessionManagerServices {
    pub fn new(
        bus: Arc<MessageBus>,
        memory_manager: Arc<crate::memory::MemoryManager>,
        task_supervisor: crate::task_supervisor::TaskSupervisor,
        turn_delivery: TurnDeliveryService,
        reload: crate::gateway::reload::ReloadHandle,
    ) -> Self {
        Self {
            bus,
            memory_manager,
            task_supervisor,
            turn_delivery,
            reload,
            admission: crate::gateway::reload::RuntimeAdmission::open(),
        }
    }

    pub(crate) fn with_admission(
        mut self,
        admission: crate::gateway::reload::RuntimeAdmission,
    ) -> Self {
        self.admission = admission;
        self
    }
}

struct SessionManagerInner {
    /// Sessions keyed by UnifiedSessionId.to_string()
    sessions: HashMap<String, Arc<Mutex<Session>>>,
    /// Current active session per channel:chat_id
    current_sessions: HashMap<String, String>,
}

/// 斜杠命令定义
#[derive(Debug, Clone)]
pub struct SlashCommand {
    /// 命令名称
    pub name: &'static str,
    /// 命令描述
    pub description: &'static str,
    /// 命令别名（触发词）
    pub aliases: &'static [&'static str],
}

impl SlashCommand {
    /// 检查给定内容是否匹配此命令
    pub fn matches(&self, content: &str) -> bool {
        let trimmed = content.trim();
        self.aliases
            .iter()
            .any(|&alias| trimmed == alias || trimmed.starts_with(&format!("{} ", alias)))
    }
}

/// Session 支持的斜杠命令列表
pub static SLASH_COMMANDS: &[SlashCommand] = &[
    SlashCommand {
        name: "new",
        description: "创建新对话",
        aliases: &["/new"],
    },
    SlashCommand {
        name: "sessions",
        description: "列出最近对话",
        aliases: &["/sessions"],
    },
    SlashCommand {
        name: "switch",
        description: "切换到指定对话",
        aliases: &["/switch"],
    },
    SlashCommand {
        name: "rename",
        description: "重命名当前对话",
        aliases: &["/rename"],
    },
    SlashCommand {
        name: "delete",
        description: "删除当前对话",
        aliases: &["/delete"],
    },
    SlashCommand {
        name: "compact",
        description: "手动触发上下文压缩",
        aliases: &["/compact"],
    },
    SlashCommand {
        name: "info",
        description: "显示当前对话信息",
        aliases: &["/info"],
    },
    SlashCommand {
        name: "dump",
        description: "保存当前对话为 markdown 文档",
        aliases: &["/dump"],
    },
    SlashCommand {
        name: "?",
        description: "显示帮助",
        aliases: &["/?", "/help"],
    },
    SlashCommand {
        name: "mcp",
        description: "显示 MCP 服务状态和工具列表",
        aliases: &["/mcp"],
    },
    SlashCommand {
        name: "stop",
        description: "停止当前正在执行的任务并清空消息队列",
        aliases: &["/stop"],
    },
    SlashCommand {
        name: "todo",
        description: "查看、完成或取消当前任务计划",
        aliases: &["/todo"],
    },
    SlashCommand {
        name: "reload",
        description: "重新加载配置",
        aliases: &["/reload"],
    },
];

fn resolve_slash_command(command: &str) -> Option<&'static SlashCommand> {
    let command = command.strip_prefix('/').unwrap_or(command);
    SLASH_COMMANDS.iter().find(|candidate| {
        candidate.name == command
            || candidate
                .aliases
                .iter()
                .any(|alias| alias.strip_prefix('/') == Some(command))
    })
}

impl SessionManager {
    fn worker_deps(&self) -> AgentWorkerDeps {
        AgentWorkerDeps {
            bus: self.bus.clone(),
            memory_manager: self.memory_manager.clone(),
            work_manager: self.work_manager.clone(),
            skills_loader: self.skills_loader.clone(),
            task_supervisor: self.task_supervisor.clone(),
            turn_delivery: self.turn_delivery.clone(),
        }
    }

    pub fn new(
        provider_config: LLMProviderConfig,
        storage: Arc<Storage>,
        services: SessionManagerServices,
        browser_config: Option<BrowserConfig>,
        max_concurrent_background_tasks: usize,
    ) -> Result<Self, AgentError> {
        let SessionManagerServices {
            bus,
            memory_manager,
            task_supervisor,
            turn_delivery,
            reload,
            admission,
        } = services;
        let mut skills_loader = SkillsLoader::new();
        skills_loader.load_skills();
        skills_loader.set_workspace_skills_dir(provider_config.workspace_dir.clone());
        let skills_loader = Arc::new(skills_loader);

        let work_manager = Arc::new(crate::work::WorkManager::new(storage.clone()));
        let tools = Arc::new(create_default_tools(
            skills_loader.clone(),
            memory_manager.clone(),
            work_manager.clone(),
            None, // SubAgentManager created below
            browser_config.as_ref(),
        ));

        // Create SubAgentManager and register DelegateTool
        let (notify_tx, mut notify_rx) = tokio::sync::mpsc::unbounded_channel();
        let sub_agent_manager = Arc::new(
            crate::agent::SubAgentManager::new(
                provider_config.clone(),
                tools.clone(),
                Some(storage.clone()),
                notify_tx,
                max_concurrent_background_tasks,
                Some(skills_loader.clone()),
                task_supervisor.clone(),
            )
            .with_admission(admission)
            .with_work_manager(work_manager.clone()),
        );
        tools.register(crate::tools::DelegateTool::new(sub_agent_manager.clone()));
        tools.register(crate::tools::ReloadConfigTool::new(reload.clone()));

        // Start background task notification consumer
        let sm_bus = bus.clone();
        task_supervisor.spawn("background-task-notifications", async move {
            while let Some(notif) = notify_rx.recv().await {
                let content =
                    format_task_notification(&notif.task_id, &notif.status, &notif.result_summary);
                let metadata = HashMap::from([
                    ("_type".to_string(), "notification".to_string()),
                    ("_session_id".to_string(), notif.session_id),
                ]);
                let outbound = OutboundMessage {
                    channel: notif.channel,
                    chat_id: notif.chat_id,
                    content,
                    reply_to: None,
                    media: vec![],
                    metadata,
                    delivery: None,
                };
                let _ = sm_bus.publish_outbound(outbound).await;
            }
        });

        // Start periodic background task cleanup (every hour, TTL 24h)
        let cleanup_storage = storage.clone();
        task_supervisor.spawn("background-task-cleanup", async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(3600));
            interval.tick().await; // skip immediate first tick
            loop {
                interval.tick().await;
                match cleanup_storage.cleanup_old_tasks(86_400_000).await {
                    Ok(count) if count > 0 => {
                        tracing::info!(count, "Cleaned up old background tasks");
                    }
                    Err(e) => {
                        tracing::warn!(error = %e, "Failed to clean up old background tasks");
                    }
                    _ => {}
                }
            }
        });

        Ok(Self {
            inner: Arc::new(Mutex::new(SessionManagerInner {
                sessions: HashMap::new(),
                current_sessions: HashMap::new(),
            })),
            provider_config,
            tools,
            skills_loader,
            storage,
            bus,
            memory_manager,
            work_manager,
            sub_agent_manager,
            task_supervisor,
            turn_delivery,
            reload,
        })
    }

    /// Register the send_message tool (requires self in Arc)
    pub fn register_outbound_tool(self: &Arc<Self>, available_channels: Vec<String>) {
        let messenger: Arc<dyn OutboundMessenger> = self.clone();
        self.tools
            .register(SendMessageTool::new(messenger, available_channels));
    }

    pub fn tools(&self) -> Arc<ToolRegistry> {
        self.tools.clone()
    }

    pub fn work_manager(&self) -> Arc<crate::work::WorkManager> {
        self.work_manager.clone()
    }

    /// 为定时任务创建一个无 session 绑定的 AgentLoop
    pub fn create_cron_agent(&self) -> Result<AgentLoop, AgentError> {
        let tools = self.tools.without(&["reload_config"]);
        let provider = create_provider(self.provider_config.clone())
            .map_err(|e| AgentError::Other(format!("failed to create cron provider: {}", e)))?;
        Ok(AgentLoop::with_provider_and_tools(
            Arc::from(provider),
            tools,
            self.provider_config.max_tool_iterations,
            self.provider_config.model_id.clone(),
            self.provider_config.workspace_dir.clone(),
            self.provider_config.input_types.clone(),
        )
        .with_context_window(self.provider_config.token_limit))
    }

    fn create_managed_scheduled_agent(&self) -> Result<(AgentLoop, Arc<ToolRegistry>), AgentError> {
        let tools = self.tools.without(&[
            "send_message",
            "cron_add",
            "cron_update",
            "cron_remove",
            "cron_enable",
            "cron_disable",
            "reload_config",
        ]);
        let provider = create_provider(self.provider_config.clone())
            .map_err(|e| AgentError::Other(format!("failed to create scheduled provider: {e}")))?;
        let agent = AgentLoop::with_provider_and_tools(
            Arc::from(provider),
            tools.clone(),
            self.provider_config.max_tool_iterations,
            self.provider_config.model_id.clone(),
            self.provider_config.workspace_dir.clone(),
            self.provider_config.input_types.clone(),
        )
        .with_context_window(self.provider_config.token_limit);
        Ok((agent, tools))
    }

    /// 获取所有可用的斜杠命令
    pub fn get_slash_commands(&self) -> &[SlashCommand] {
        SLASH_COMMANDS
    }

    /// 执行斜杠命令
    /// 返回 (新session_id, 响应消息)
    pub async fn execute_slash_command(
        &self,
        command: &str,
        args: Option<&str>,
        channel: &str,
        chat_id: &str,
        current_session_id: Option<&UnifiedSessionId>,
    ) -> Result<(Option<UnifiedSessionId>, String), AgentError> {
        let cmd = resolve_slash_command(command)
            .ok_or_else(|| AgentError::Other(format!("Unknown command: {}", command)))?;

        tracing::info!(cmd = %cmd.name, args = ?args, "Executing slash command");

        match cmd.name {
            "new" => {
                let title = args.map(|s| s.to_string());
                let (new_id, title) = self
                    .create_session(channel, chat_id, title.as_deref(), String::new())
                    .await?;
                Ok((Some(new_id), format!("新对话 '{}' 已创建。", title)))
            }
            "delete" => {
                if let Some(sid) = current_session_id {
                    self.delete_dialog(sid).await?;
                }
                let (new_id, _title) = self
                    .create_session(channel, chat_id, None, String::new())
                    .await?;
                Ok((Some(new_id), "对话已删除。新对话已创建。".to_string()))
            }
            "compact" => {
                if let Some(sid) = current_session_id {
                    let session = self.get_or_create_session(sid).await?;
                    let (original_count, history, mut compressor, base_version) = {
                        let session_guard = session.lock().await;
                        (
                            session_guard.get_history().len(),
                            session_guard.get_history().to_vec(),
                            session_guard.fresh_context_compressor(),
                            session_guard.state_version,
                        )
                    };

                    let result = compressor.compress_if_needed(history).await?;
                    let compressed_count = result.history.len();
                    let meta_snapshot = {
                        let mut session_guard = session.lock().await;
                        if session_guard.state_version != base_version {
                            return Ok((
                                None,
                                "Context changed while compacting; please run /compact again."
                                    .to_string(),
                            ));
                        }
                        if result.created_timelines {
                            session_guard.last_compressed_message_at =
                                Some(chrono::Utc::now().timestamp_millis());
                        }
                        session_guard.replace_history_in_memory(result.history);
                        session_guard.session_meta_snapshot()
                    };

                    if let Some((storage, meta)) = meta_snapshot
                        && let Err(e) = storage.upsert_session(&meta).await
                    {
                        tracing::warn!(error = %e, "Failed to persist compression marker after /compact");
                    }
                    Ok((
                        None,
                        format!(
                            "Context compressed: {} → {} messages.",
                            original_count, compressed_count
                        ),
                    ))
                } else {
                    Ok((None, "No active conversation to compress.".to_string()))
                }
            }
            "info" => {
                if let Some(sid) = current_session_id {
                    let session = self.get_or_create_session(sid).await?;
                    let session_guard = session.lock().await;
                    let history = session_guard.get_history();
                    let message_count = history.len();
                    let session_id_str = session_guard.session_id();
                    let title = &session_guard.title;
                    let model_name = &session_guard.provider_config.name;
                    let created_at =
                        chrono::DateTime::from_timestamp_millis(session_guard.created_at)
                            .map(|dt| {
                                dt.with_timezone(&chrono::Local)
                                    .format("%Y-%m-%d %H:%M:%S")
                                    .to_string()
                            })
                            .unwrap_or_default();
                    let last_active_at =
                        chrono::DateTime::from_timestamp_millis(session_guard.last_active_at)
                            .map(|dt| {
                                dt.with_timezone(&chrono::Local)
                                    .format("%Y-%m-%d %H:%M:%S")
                                    .to_string()
                            })
                            .unwrap_or_default();
                    let token_info = session_guard.compressor.token_info(history);
                    let cache_info = if token_info.cache_active {
                        format!(
                            "API精确: {} tokens",
                            token_info.last_api_tokens.unwrap_or(0)
                        )
                    } else {
                        "无API精确缓存".to_string()
                    };
                    let threshold_pct = if token_info.context_window > 0 {
                        (token_info.threshold as f64 / token_info.context_window as f64 * 100.0)
                            as usize
                    } else {
                        0
                    };
                    let usage_pct = if token_info.context_window > 0 {
                        (token_info.estimated_tokens as f64 / token_info.context_window as f64
                            * 100.0)
                            .min(100.0) as usize
                    } else {
                        0
                    };
                    let usage_bar = if token_info.context_window > 0 {
                        format!(
                            "{}/{} tokens ({}%)",
                            token_info.estimated_tokens, token_info.context_window, usage_pct
                        )
                    } else {
                        "未设置".to_string()
                    };
                    let compression_status = if token_info.estimated_tokens > token_info.threshold {
                        "[即将压缩]"
                    } else {
                        "[正常]"
                    };
                    let ctx_info = format!(
                        "[窗口] {} [阈值] {}/{} ({}) [状态] {} {}",
                        usage_bar,
                        token_info.threshold,
                        token_info.context_window,
                        threshold_pct,
                        compression_status,
                        cache_info,
                    );
                    Ok((
                        None,
                        format!(
                            "对话标题: {}\nSession ID: {}\n模型: {}\n用户消息: {} / 总消息: {}\n创建时间: {}\n最后活跃: {}\n\n上下文: {}",
                            title,
                            session_id_str,
                            model_name,
                            session_guard.message_count,
                            message_count,
                            created_at,
                            last_active_at,
                            ctx_info,
                        ),
                    ))
                } else {
                    Ok((None, "No active session.".to_string()))
                }
            }
            "dump" => {
                if let Some(sid) = current_session_id {
                    let session = self.get_or_create_session(sid).await?;
                    let session_guard = session.lock().await;

                    // Build the same system prompt that would be injected to the model
                    let skills_prompt = self.skills_loader.build_skills_prompt();
                    let system_prompt = session_guard.build_system_prompt(&skills_prompt);

                    let filepath = session_guard
                        .dump_to_file(&system_prompt)
                        .map_err(|e| AgentError::Other(format!("Failed to save dump: {}", e)))?;
                    Ok((None, format!("Session dump saved to: {}", filepath)))
                } else {
                    Ok((None, "No active session.".to_string()))
                }
            }
            "sessions" => {
                let (dialogs, _current) = self.list_dialogs(channel, chat_id, false).await?;
                if dialogs.is_empty() {
                    Ok((None, "暂无对话记录。".to_string()))
                } else {
                    let lines: Vec<String> = dialogs
                        .iter()
                        .map(|d| {
                            let current = if current_session_id
                                .map(|s| s.dialog_id == d.session_id.dialog_id)
                                .unwrap_or(false)
                            {
                                " [当前]"
                            } else {
                                ""
                            };
                            format!(
                                "- {} ({}){} — {}",
                                d.session_id.dialog_id,
                                d.title,
                                current,
                                chrono::DateTime::from_timestamp_millis(d.last_active_at)
                                    .map(|dt| dt
                                        .with_timezone(&chrono::Local)
                                        .format("%m-%d %H:%M")
                                        .to_string())
                                    .unwrap_or_default()
                            )
                        })
                        .collect();
                    Ok((None, format!("最近对话：\n{}", lines.join("\n"))))
                }
            }
            "switch" => {
                let dialog_id = args
                    .ok_or_else(|| AgentError::Other("Usage: /switch <dialog_id>".to_string()))?;
                let new_id = self.switch_dialog(channel, chat_id, dialog_id).await?;
                Ok((None, format!("已切换到对话：{}", new_id.dialog_id)))
            }
            "rename" => {
                let title =
                    args.ok_or_else(|| AgentError::Other("Usage: /rename <新标题>".to_string()))?;
                if let Some(sid) = current_session_id {
                    self.rename_dialog(sid, title).await?;
                    Ok((None, format!("对话已重命名为：{}", title)))
                } else {
                    Ok((None, "No active session.".to_string()))
                }
            }
            "?" => {
                let lines: Vec<String> = SLASH_COMMANDS
                    .iter()
                    .map(|c| format!("  {}  - {}", c.aliases.join(", "), c.description))
                    .collect();
                Ok((None, format!("可用命令：\n{}", lines.join("\n"))))
            }
            "mcp" => {
                let servers = get_mcp_status();
                if servers.is_empty() {
                    return Ok((None, "未配置 MCP 服务。".to_string()));
                }
                let lines: Vec<String> = servers
                    .iter()
                    .map(|s| {
                        let status = if s.connected {
                            format!("✅ 已连接 ({})", s.transport)
                        } else {
                            format!("❌ 连接失败: {}", s.error.as_deref().unwrap_or("未知错误"))
                        };
                        let tool_lines: Vec<String> = s
                            .tools
                            .iter()
                            .map(|t| {
                                let desc = if t.description.is_empty() {
                                    "无描述".to_string()
                                } else {
                                    t.description.chars().take(60).collect::<String>()
                                };
                                format!("    - {}: {}", t.name, desc)
                            })
                            .collect();
                        let tools_section = if tool_lines.is_empty() {
                            String::new()
                        } else {
                            format!("\n{}", tool_lines.join("\n"))
                        };
                        format!("{} {}{}", s.name, status, tools_section)
                    })
                    .collect();
                Ok((None, format!("MCP 服务：\n\n{}", lines.join("\n\n"))))
            }
            "stop" => {
                let sid = current_session_id
                    .ok_or_else(|| AgentError::Other("no active session".to_string()))?;
                let session = self.get_or_create_session(sid).await?;
                let msgs = {
                    let mut guard = session.lock().await;
                    let mut msgs: Vec<String> = Vec::new();
                    if guard.current_cancel.take().is_some() {
                        msgs.push("当前任务已发送停止信号。".to_string());
                    }
                    if let Some(active_turn) = guard.active_turn_emitter.take() {
                        active_turn.emitter.deactivate();
                    }
                    if guard.agent_tx.take().is_some() {
                        msgs.push("消息队列已清空。".to_string());
                    }
                    guard.worker_generation = guard.worker_generation.wrapping_add(1);
                    guard.state_version = guard.state_version.wrapping_add(1);
                    msgs
                };

                // Cancel all running background sub-agent tasks for this session
                // after releasing the session lock.
                self.sub_agent_manager
                    .cancel_by_session(&sid.to_string())
                    .await;
                let resp = if msgs.is_empty() {
                    "没有正在执行的任务或队列。".to_string()
                } else {
                    msgs.join(" ")
                };
                Ok((None, resp))
            }
            "todo" => {
                let sid = current_session_id
                    .ok_or_else(|| AgentError::Other("no active session".to_string()))?;
                let action = args.map(str::trim).filter(|value| !value.is_empty());
                match action {
                    Some("cancel") => self
                        .work_manager
                        .close_plan(&sid.to_string(), "cancelled", None)
                        .await
                        .map(|plan| (None, format!("任务计划已取消：{}", plan.objective)))
                        .map_err(|error| AgentError::Other(error.to_string())),
                    Some("done") => self
                        .work_manager
                        .close_plan(&sid.to_string(), "completed", None)
                        .await
                        .map(|plan| (None, format!("任务计划已完成：{}", plan.objective)))
                        .map_err(|error| AgentError::Other(error.to_string())),
                    Some(_) => Err(AgentError::Other("Usage: /todo [done|cancel]".to_string())),
                    None => match self.work_manager.active_plan(&sid.to_string()).await {
                        Ok(Some(plan)) => {
                            let mut lines = vec![format!(
                                "任务计划：{}（version {}）",
                                plan.objective, plan.version
                            )];
                            for item in plan.items {
                                let icon = match item.status.as_str() {
                                    "completed" => "✓",
                                    "in_progress" => "●",
                                    "blocked" => "!",
                                    _ => "○",
                                };
                                lines.push(format!(
                                    "{icon} {} [{}] {}",
                                    item.id, item.status, item.title
                                ));
                            }
                            Ok((None, lines.join("\n")))
                        }
                        Ok(None) => Ok((None, "当前 session 没有 active plan。".to_string())),
                        Err(error) => Err(AgentError::Other(error.to_string())),
                    },
                }
            }
            "reload" => self
                .reload
                .request()
                .await
                .map(|accepted| (None, accepted.message))
                .map_err(|error| AgentError::Other(error.to_string())),
            _ => Err(AgentError::Other(format!(
                "未知命令：/{}。输入 /? 获取帮助。",
                cmd.name
            ))),
        }
    }

    /// Wait until all interactive session Turns have reached a terminal state.
    /// A reload uses this to avoid cancelling the Turn that requested it.
    pub async fn wait_until_idle(&self, timeout: std::time::Duration) -> bool {
        let deadline = tokio::time::Instant::now() + timeout;
        let mut idle_since = None;
        loop {
            let sessions: Vec<_> = {
                let inner = self.inner.lock().await;
                inner.sessions.values().cloned().collect()
            };
            let mut busy = false;
            for session in sessions {
                let session = session.lock().await;
                let queued = session
                    .agent_tx
                    .as_ref()
                    .is_some_and(|sender| sender.capacity() < sender.max_capacity());
                if session.current_cancel.is_some() || queued {
                    busy = true;
                    break;
                }
            }
            if busy {
                idle_since = None;
            } else {
                let since = idle_since.get_or_insert_with(tokio::time::Instant::now);
                if since.elapsed() >= std::time::Duration::from_millis(100) {
                    return true;
                }
            }
            if tokio::time::Instant::now() >= deadline {
                return false;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    }

    /// Number of live sessions currently tracked by the manager.
    pub async fn session_count(&self) -> usize {
        self.inner.lock().await.sessions.len()
    }

    /// Number of sessions with an actively executing Turn.
    pub async fn active_turn_count(&self) -> usize {
        let sessions: Vec<_> = {
            let inner = self.inner.lock().await;
            inner.sessions.values().cloned().collect()
        };
        let mut count = 0;
        for session in sessions {
            let session = session.lock().await;
            if session.current_cancel.is_some() {
                count += 1;
            }
        }
        count
    }

    pub async fn create_session(
        &self,
        channel: &str,
        chat_id: &str,
        title: Option<&str>,
        routing_info: String,
    ) -> Result<(UnifiedSessionId, String), AgentError> {
        let dialog_id = crate::util::short_id();
        let unified_id = UnifiedSessionId::new(channel, chat_id, &dialog_id);
        let session_id_str = unified_id.to_string();

        let title = title
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(ToOwned::to_owned)
            .unwrap_or_else(|| "新对话".to_string());

        // Write to Storage first
        let now = chrono::Utc::now().timestamp_millis();
        let meta = crate::storage::session::SessionMeta {
            id: session_id_str.clone(),
            channel: channel.to_string(),
            chat_id: chat_id.to_string(),
            dialog_id: dialog_id.clone(),
            title: title.clone(),
            created_at: now,
            last_active_at: now,
            message_count: 0,
            routing_info: if routing_info.is_empty() {
                None
            } else {
                Some(routing_info.clone())
            },
            archived_at: None,
            deleted_at: None,
            last_consolidated_at: None,
            last_compressed_message_at: None,
        };
        self.storage.upsert_session(&meta).await.map_err(|e| {
            AgentError::Other(format!("failed to create session in storage: {}", e))
        })?;

        let session = Session::new(
            unified_id.clone(),
            self.provider_config.clone(),
            self.tools.clone(),
            Some(self.storage.clone()),
            routing_info,
            title.clone(),
            self.memory_manager.clone(),
        )
        .await?;

        let arc = Arc::new(Mutex::new(session));
        let inner = &mut *self.inner.lock().await;
        inner.sessions.insert(session_id_str.clone(), arc.clone());
        // Set as current session for this channel:chat_id
        let chat_scope = format!("{}:{}", channel, chat_id);
        inner.current_sessions.insert(chat_scope, session_id_str);

        Ok((unified_id, title))
    }

    pub async fn get_or_create_session(
        &self,
        unified_id: &UnifiedSessionId,
    ) -> Result<Arc<Mutex<Session>>, AgentError> {
        let session_id_str = unified_id.to_string();
        if let Some(session) = self
            .inner
            .lock()
            .await
            .sessions
            .get(&session_id_str)
            .cloned()
        {
            return Ok(session);
        }

        // Perform storage/provider I/O without holding the global registry lock.
        let session = match self.storage.get_session(&session_id_str).await {
            Ok(meta) => {
                tracing::debug!(session_id = %session_id_str, last_active_at = %meta.last_active_at, message_count = %meta.message_count, "Restoring session from Storage");
                Session::from_storage(
                    unified_id.clone(),
                    self.provider_config.clone(),
                    self.tools.clone(),
                    self.storage.clone(),
                    self.memory_manager.clone(),
                )
                .await?
            }
            Err(StorageError::NotFound(_)) => {
                let now = chrono::Utc::now().timestamp_millis();
                let meta = crate::storage::session::SessionMeta {
                    id: session_id_str.clone(),
                    channel: unified_id.channel.clone(),
                    chat_id: unified_id.chat_id.clone(),
                    dialog_id: unified_id.dialog_id.clone(),
                    title: "新对话".to_string(),
                    created_at: now,
                    last_active_at: now,
                    message_count: 0,
                    routing_info: None,
                    archived_at: None,
                    deleted_at: None,
                    last_consolidated_at: None,
                    last_compressed_message_at: None,
                };
                self.storage.upsert_session(&meta).await.map_err(|e| {
                    AgentError::Other(format!("failed to create session in storage: {}", e))
                })?;
                Session::new(
                    unified_id.clone(),
                    self.provider_config.clone(),
                    self.tools.clone(),
                    Some(self.storage.clone()),
                    String::new(),
                    "新对话".to_string(),
                    self.memory_manager.clone(),
                )
                .await?
            }
            Err(e) => {
                return Err(AgentError::Other(format!(
                    "failed to look up session in storage: {}",
                    e
                )));
            }
        };

        let arc = Arc::new(Mutex::new(session));
        // Another caller may have completed the same load while I/O was in
        // progress. Keep the already-published instance as the single source
        // of truth.
        let inner = &mut *self.inner.lock().await;
        if let Some(existing) = inner.sessions.get(&session_id_str) {
            return Ok(existing.clone());
        }
        inner.sessions.insert(session_id_str.clone(), arc.clone());
        let chat_scope = format!("{}:{}", unified_id.channel, unified_id.chat_id);
        inner.current_sessions.insert(chat_scope, session_id_str);
        Ok(arc)
    }

    pub async fn create_dialog(
        &self,
        channel: &str,
        chat_id: &str,
        title: Option<&str>,
    ) -> Result<(UnifiedSessionId, String), AgentError> {
        self.create_session(channel, chat_id, title, String::new())
            .await
    }

    pub async fn get_current_dialog(
        &self,
        channel: &str,
        chat_id: &str,
    ) -> Result<Option<UnifiedSessionId>, AgentError> {
        let chat_scope = format!("{}:{}", channel, chat_id);
        let current = {
            self.inner
                .lock()
                .await
                .current_sessions
                .get(&chat_scope)
                .cloned()
        };

        let Some(current) = current else {
            return Ok(None);
        };

        match self.storage.get_session(&current).await {
            Ok(_) => Ok(UnifiedSessionId::parse(&current)),
            Err(StorageError::NotFound(_)) => Ok(None),
            Err(e) => Err(AgentError::Other(format!("storage error: {}", e))),
        }
    }

    pub async fn switch_dialog(
        &self,
        channel: &str,
        chat_id: &str,
        dialog_id: &str,
    ) -> Result<UnifiedSessionId, AgentError> {
        let unified_id = UnifiedSessionId::new(channel, chat_id, dialog_id);
        // Ensure session is loaded into memory
        self.get_or_create_session(&unified_id).await?;
        // Update current session tracking
        let mut inner = self.inner.lock().await;
        let chat_scope = format!("{}:{}", channel, chat_id);
        inner
            .current_sessions
            .insert(chat_scope, unified_id.to_string());
        Ok(unified_id)
    }

    pub async fn get_dialog_history(
        &self,
        session_id: &UnifiedSessionId,
        limit: u32,
    ) -> Result<Vec<crate::storage::message::MessageMeta>, AgentError> {
        let session_id = session_id.to_string();
        self.storage
            .get_session(&session_id)
            .await
            .map_err(|e| AgentError::Other(format!("failed to load dialog: {e}")))?;
        self.storage
            .load_recent_session_messages(&session_id, limit)
            .await
            .map_err(|e| AgentError::Other(format!("failed to load dialog history: {e}")))
    }

    pub async fn get_task_plan(
        &self,
        session_id: &UnifiedSessionId,
    ) -> Result<Option<crate::work::TaskPlan>, AgentError> {
        let session_id = session_id.to_string();
        self.storage
            .get_session(&session_id)
            .await
            .map_err(|error| AgentError::Other(format!("failed to load dialog: {error}")))?;
        self.work_manager
            .plan_for_session(&session_id)
            .await
            .map_err(|error| AgentError::Other(format!("failed to load task plan: {error}")))
    }

    pub async fn list_dialogs(
        &self,
        channel: &str,
        chat_id: &str,
        include_archived: bool,
    ) -> Result<(Vec<DialogInfo>, Option<String>), AgentError> {
        let metas = self
            .storage
            .list_sessions(channel, chat_id, 100, include_archived)
            .await
            .map_err(|e| AgentError::Other(format!("failed to list dialogs: {}", e)))?;
        let current_dialog_id = self
            .get_current_dialog(channel, chat_id)
            .await?
            .map(|sid| sid.dialog_id);

        let dialogs: Vec<DialogInfo> = metas
            .into_iter()
            .map(|meta| DialogInfo {
                session_id: UnifiedSessionId::new(channel, chat_id, &meta.dialog_id),
                title: meta.title,
                created_at: meta.created_at,
                last_active_at: meta.last_active_at,
                message_count: meta.message_count,
                archived_at: meta.archived_at,
            })
            .collect();

        Ok((dialogs, current_dialog_id))
    }

    pub async fn rename_dialog(
        &self,
        session_id: &UnifiedSessionId,
        title: &str,
    ) -> Result<(), AgentError> {
        // Update in-memory session
        let session = self.get_or_create_session(session_id).await?;
        let persistence_lock = { session.lock().await.persistence_lock.clone() };
        let _persistence_guard = persistence_lock.lock().await;
        let meta_snapshot = {
            let mut session_guard = session.lock().await;
            session_guard.title = title.to_string();
            session_guard.state_version = session_guard.state_version.wrapping_add(1);
            session_guard.session_meta_snapshot()
        };
        if let Some((storage, meta)) = meta_snapshot {
            storage
                .upsert_session(&meta)
                .await
                .map_err(|e| AgentError::Other(format!("failed to rename dialog: {}", e)))?;
        }
        Ok(())
    }

    pub async fn delete_dialog(&self, session_id: &UnifiedSessionId) -> Result<(), AgentError> {
        let session_id_str = session_id.to_string();
        let session = self.get_or_create_session(session_id).await?;
        let persistence_lock = { session.lock().await.persistence_lock.clone() };
        let _persistence_guard = persistence_lock.lock().await;

        // Soft delete from Storage
        self.storage
            .soft_delete_session(&session_id_str)
            .await
            .map_err(|e| AgentError::Other(format!("failed to delete dialog: {}", e)))?;

        // Remove from memory and current sessions
        let mut inner = self.inner.lock().await;
        inner.sessions.remove(&session_id_str);
        let chat_scope = format!("{}:{}", session_id.channel, session_id.chat_id);
        inner.current_sessions.remove(&chat_scope);

        Ok(())
    }

    pub async fn archive_dialog(&self, session_id: &UnifiedSessionId) -> Result<(), AgentError> {
        let session_id_str = session_id.to_string();
        let session = self.get_or_create_session(session_id).await?;
        let persistence_lock = { session.lock().await.persistence_lock.clone() };
        let _persistence_guard = persistence_lock.lock().await;
        self.storage
            .archive_session(&session_id_str)
            .await
            .map_err(|e| AgentError::Other(format!("failed to archive dialog: {}", e)))?;

        let mut inner = self.inner.lock().await;
        inner.sessions.remove(&session_id_str);
        let chat_scope = format!("{}:{}", session_id.channel, session_id.chat_id);
        if inner
            .current_sessions
            .get(&chat_scope)
            .is_some_and(|id| id == &session_id_str)
        {
            inner.current_sessions.remove(&chat_scope);
        }

        Ok(())
    }

    pub async fn clear_dialog_history(
        &self,
        session_id: &UnifiedSessionId,
    ) -> Result<(), AgentError> {
        self.clear_session_history(session_id).await
    }

    /// Get or activate a specific session by its full UnifiedSessionId.
    /// Returns an error if the session does not exist in storage.
    /// If the session was expired from memory but still in storage,
    /// it will be restored (reactivated).
    pub async fn get_or_activate_session(
        &self,
        unified_id: &UnifiedSessionId,
    ) -> Result<Arc<Mutex<Session>>, AgentError> {
        let session_id_str = unified_id.to_string();
        match self.storage.get_session(&session_id_str).await {
            Ok(_) => self.get_or_create_session(unified_id).await,
            Err(StorageError::NotFound(_)) => Err(AgentError::Other(format!(
                "session not found: {}",
                unified_id
            ))),
            Err(e) => Err(AgentError::Other(format!("storage error: {}", e))),
        }
    }

    pub(super) async fn resolve_dialog_id(
        &self,
        channel: &str,
        chat_id: &str,
    ) -> Result<UnifiedSessionId, AgentError> {
        let chat_scope = format!("{}:{}", channel, chat_id);
        let current_id = {
            self.inner
                .lock()
                .await
                .current_sessions
                .get(&chat_scope)
                .cloned()
        };

        if let Some(ref current_id) = current_id {
            match self.storage.get_session(current_id).await {
                Ok(_) => {
                    if let Some(parsed) = UnifiedSessionId::parse(current_id) {
                        return Ok(parsed);
                    }
                }
                Err(StorageError::NotFound(_)) => {}
                Err(e) => {
                    return Err(AgentError::Other(format!(
                        "failed to resolve current session: {}",
                        e
                    )));
                }
            }
        }

        match self
            .storage
            .find_most_recent_session(channel, chat_id)
            .await
        {
            Ok(Some(meta)) => Ok(UnifiedSessionId::new(channel, chat_id, &meta.dialog_id)),
            Ok(None) => {
                let (new_id, _) = self
                    .create_session(channel, chat_id, None, String::new())
                    .await?;
                Ok(new_id)
            }
            Err(e) => Err(AgentError::Other(format!(
                "failed to find recent session: {}",
                e
            ))),
        }
    }

    pub(super) async fn restore_origin_dialog(&self, origin_id: &str, target: &UnifiedSessionId) {
        let Some(origin) = UnifiedSessionId::parse(origin_id) else {
            tracing::warn!(origin_id, "Ignoring malformed source session id");
            return;
        };
        if origin.channel == target.channel
            && origin.chat_id == target.chat_id
            && origin.dialog_id != target.dialog_id
        {
            self.inner
                .lock()
                .await
                .current_sessions
                .insert(target.chat_scope(), origin_id.to_string());
        }
    }

    /// Send a system notification (no LLM triggered).
    ///
    /// Flow:
    /// 1. Resolve target session (resolve_dialog_id)
    /// 2. Write assistant message with source tag to history
    /// 3. Publish OutboundMessage via bus to target channel
    pub async fn send_notification(
        &self,
        channel: &str,
        chat_id: &str,
        content: &str,
        system_name: &str,
        task_id: Option<&str>,
    ) -> Result<(), AgentError> {
        let unified_id = self.resolve_dialog_id(channel, chat_id).await?;
        let session = self.get_or_create_session(&unified_id).await?;
        let source = MessageSource {
            kind: SourceKind::SystemNotification,
            from_channel: None,
            from_session: None,
            from_user_id: None,
            system_name: Some(system_name.to_string()),
            task_id: task_id.map(|s| s.to_string()),
        };
        let msg = ChatMessage::assistant_with_source(content, source);
        append_persisted_messages(&session, vec![msg])
            .await
            .map_err(|e| AgentError::Other(format!("persist error: {}", e)))?;

        let outbound = OutboundMessage {
            channel: channel.to_string(),
            chat_id: chat_id.to_string(),
            content: content.to_string(),
            reply_to: None,
            media: vec![],
            metadata: outbound_session_metadata(&unified_id.to_string()),
            delivery: None,
        };
        self.bus
            .deliver_outbound(outbound)
            .await
            .map_err(|e| AgentError::Other(format!("bus publish error: {}", e)))?;

        Ok(())
    }

    pub async fn handle_message(
        &self,
        inbound: &InboundMessage,
    ) -> Result<HandleResult, AgentError> {
        let channel = inbound.channel.as_str();
        let sender_id = inbound.sender_id.as_str();
        let chat_id = inbound.chat_id.as_str();
        let content = inbound.content.as_str();
        let unified_id = self.resolve_dialog_id(channel, chat_id).await?;
        tracing::debug!(unified_id = %unified_id, "handle_message resolved unified_id");
        let session = self.get_or_create_session(&unified_id).await?;

        CURRENT_SOURCE_SESSION
            .scope(Some(unified_id.to_string()), async {
                // Check for slash command
                if let Some((cmd_name, cmd_args)) = parse_slash_command(content) {
                    let result = self
                        .execute_slash_command(
                            cmd_name,
                            if cmd_args.is_empty() {
                                None
                            } else {
                                Some(cmd_args)
                            },
                            channel,
                            chat_id,
                            Some(&unified_id),
                        )
                        .await;

                    return match result {
                        Ok((_new_session_id, response)) => {
                            Ok(HandleResult::CommandOutput(response))
                        }
                        Err(e) => Ok(HandleResult::CommandOutput(e.to_string())),
                    };
                }

                // Normal message: enqueue to per-session worker for serial processing.
                let task = AgentTask {
                    channel: channel.to_string(),
                    sender_id: sender_id.to_string(),
                    chat_id: chat_id.to_string(),
                    content: content.to_string(),
                    received_at: inbound.received_at,
                    media: inbound.media.clone(),
                    channel_context: inbound.channel_context.clone(),
                };
                let session_clone = session.clone();
                let unified_str = unified_id.to_string();
                {
                    let mut guard = session_clone.lock().await;
                    let needs_spawn = guard.agent_tx.is_none()
                        || guard.agent_tx.as_ref().is_some_and(|tx| tx.is_closed());
                    if needs_spawn {
                        guard.agent_tx = None;
                        guard.current_cancel = None;
                        guard.worker_generation = guard.worker_generation.wrapping_add(1);
                        let generation = guard.worker_generation;
                        let (tx, rx) = mpsc::channel(SESSION_QUEUE_CAPACITY);
                        guard.agent_tx = Some(tx);
                        spawn_agent_worker(
                            rx,
                            session_clone.clone(),
                            self.worker_deps(),
                            generation,
                            unified_str.clone(),
                        );
                    }
                    let Some(agent_tx) = guard.agent_tx.as_ref() else {
                        return Err(AgentError::Other(
                            "agent worker queue was not initialized".to_string(),
                        ));
                    };
                    if let Err(e) = agent_tx.try_send(task) {
                        if matches!(e, mpsc::error::TrySendError::Full(_)) {
                            tracing::warn!(session_id = %unified_str, capacity = SESSION_QUEUE_CAPACITY, "Session queue is full");
                            return Ok(HandleResult::CommandOutput(
                                "当前对话消息队列已满，请稍后重试。".to_string(),
                            ));
                        }
                        // Worker died after we just spawned it — respawn with the recovered task
                        let task = e.into_inner();
                        guard.agent_tx = None;
                        guard.current_cancel = None;
                        guard.worker_generation = guard.worker_generation.wrapping_add(1);
                        let generation = guard.worker_generation;
                        let (tx, rx) = mpsc::channel(SESSION_QUEUE_CAPACITY);
                        guard.agent_tx = Some(tx);
                        spawn_agent_worker(
                            rx,
                            session_clone.clone(),
                            self.worker_deps(),
                            generation,
                            unified_str.clone(),
                        );
                        guard
                            .agent_tx
                            .as_ref()
                            .ok_or_else(|| {
                                AgentError::Other(
                                    "agent worker queue was not initialized".to_string(),
                                )
                            })?
                            .try_send(task)
                            .map_err(|_| {
                                AgentError::Other(
                                    "agent worker spawn+send failed irrecoverably".to_string(),
                                )
                            })?;
                    }
                }
                Ok(HandleResult::AgentProcessing)
            })
            .await
    }
}

async fn generate_title(
    session: Arc<Mutex<Session>>,
    provider: Arc<dyn LLMProvider>,
    prompt: String,
) -> Result<(), AgentError> {
    use crate::providers::{ChatCompletionRequest, ChatCompletionResponse, Message};

    let request = ChatCompletionRequest {
        messages: vec![Message::user(prompt)],
        temperature: Some(0.3),
        max_tokens: Some(20),
        tools: None,
    };

    let response: ChatCompletionResponse = provider
        .chat(request)
        .await
        .map_err(|e| AgentError::Other(format!("LLM call failed: {}", e)))?;
    let title = response.content.trim().to_string();

    let persistence_lock = { session.lock().await.persistence_lock.clone() };
    let _persistence_guard = persistence_lock.lock().await;
    let meta_snapshot = {
        let mut guard = session.lock().await;
        if guard.apply_generated_title(title) {
            guard.session_meta_snapshot()
        } else {
            None
        }
    };

    if let Some((storage, meta)) = meta_snapshot {
        storage
            .upsert_session(&meta)
            .await
            .map_err(|e| AgentError::Other(format!("failed to persist title: {}", e)))?;
    }

    Ok(())
}

async fn schedule_title_generation(
    session: Arc<Mutex<Session>>,
    supervisor: crate::task_supervisor::TaskSupervisor,
    session_id: &str,
) {
    let title_job = {
        let mut guard = session.lock().await;
        if guard.title_generation_in_flight {
            None
        } else {
            guard.title_prompt_snapshot().map(|prompt| {
                guard.title_generation_in_flight = true;
                (guard.provider.clone(), prompt)
            })
        }
    };
    let Some((provider, prompt)) = title_job else {
        return;
    };

    let title_session = session.clone();
    let task_session = session.clone();
    let spawned = supervisor.spawn(format!("session-title:{session_id}"), async move {
        if let Err(error) = generate_title(title_session, provider, prompt).await {
            tracing::warn!(error = %error, "Failed to generate session title");
        }
        task_session.lock().await.title_generation_in_flight = false;
    });
    if !spawned {
        session.lock().await.title_generation_in_flight = false;
    }
}

fn spawn_agent_worker(
    mut task_rx: mpsc::Receiver<AgentTask>,
    session: Arc<Mutex<Session>>,
    deps: AgentWorkerDeps,
    worker_gen: u64,
    unified_str: String,
) {
    let AgentWorkerDeps {
        bus,
        memory_manager,
        work_manager,
        skills_loader,
        task_supervisor,
        turn_delivery,
    } = deps;
    let worker_supervisor = task_supervisor.clone();
    task_supervisor.spawn(format!("session-worker:{unified_str}"), async move {
        let unified_for_source = unified_str.clone();
        let _scope = CURRENT_SOURCE_SESSION.scope(Some(unified_for_source), async {
            'tasks: while let Some(task) = task_rx.recv().await {
                let task_chan = task.channel.clone();
                let task_cid = task.chat_id.clone();
                let task_metadata = task.channel_context.private.clone();
                let task_reply_to = task.channel_context.reply_to.clone();
                // Phase 1: capture a stable session snapshot under lock.
                // Memory recall and compression happen outside this block so
                // /stop and other commands are not blocked behind slow I/O or
                // LLM-backed compaction.
                let skills_prompt = skills_loader.build_skills_prompt();
                let user_message = {
                    let guard = session.lock().await;
                    if guard.worker_generation != worker_gen {
                        return;
                    }
                    let media_refs: Vec<MediaRef> =
                        task.media.iter().map(MediaItem::to_media_ref).collect();
                    let source = MessageSource {
                        kind: SourceKind::UserInput,
                        from_channel: Some(task.channel.clone()),
                        from_session: None,
                        from_user_id: Some(task.sender_id.clone()),
                        system_name: None,
                        task_id: None,
                    };
                    let mut message =
                        guard.create_user_message_with_source(&task.content, media_refs, source);
                    message.timestamp = task.received_at;
                    message
                };
                if let Err(e) = append_persisted_messages(&session, vec![user_message]).await {
                    tracing::error!(error = %e, "Failed to persist user message");
                    let err_outbound = OutboundMessage {
                        channel: task_chan.clone(),
                        chat_id: task_cid.clone(),
                        content: "Failed to save your message, please try again.".to_string(),
                        reply_to: task_reply_to.clone(),
                        media: vec![],
                        metadata: outbound_turn_metadata(&unified_str, &task_metadata),
                        delivery: None,
                    };
                    let _ = bus.publish_outbound(err_outbound).await;
                    continue 'tasks;
                }

                let (
                    agent,
                    history_raw,
                    mut compressor,
                    system_prompt_out,
                    base_version,
                    cancel_rx,
                ) = {
                    let mut guard = session.lock().await;

                    if guard.worker_generation != worker_gen {
                        return; // stale worker
                    }

                    let history_raw = guard.get_history().to_vec();

                    let agent = match guard.create_agent() {
                        Ok(a) => a,
                        Err(e) => {
                            tracing::error!(error = %e, "Failed to create agent");
                            let err_outbound = OutboundMessage {
                                channel: task_chan.clone(),
                                chat_id: task_cid.clone(),
                                content: "Agent creation failed, please try again."
                                    .to_string(),
                                reply_to: task_reply_to.clone(),
                                media: vec![],
                                metadata: outbound_turn_metadata(&unified_str, &task_metadata),
                                delivery: None,
                            };
                            let _ = bus.publish_outbound(err_outbound).await;
                            continue 'tasks;
                        }
                    };

                    let (cancel_tx, cancel_rx) = oneshot::channel();

                    if guard.worker_generation != worker_gen {
                        return; // /stop replaced us
                    }
                    guard.current_cancel = Some(cancel_tx);

                    (
                        agent,
                        history_raw,
                        guard.fresh_context_compressor(),
                        guard.build_system_prompt(&skills_prompt),
                        guard.state_version,
                        cancel_rx,
                    )
                }; // lock released

                let prepared_input = prepare_turn_input(
                    memory_manager.clone(),
                    work_manager.clone(),
                    &unified_str,
                    &task.content,
                    system_prompt_out,
                    &mut compressor,
                    history_raw,
                )
                .await;
                let meta_snapshot = {
                    let mut guard = session.lock().await;
                    if guard.worker_generation != worker_gen {
                        return;
                    }
                    if guard.state_version != base_version {
                        tracing::warn!(
                            session_id = %guard.id,
                            "Session changed while preparing agent history; dropping stale task"
                        );
                        guard.current_cancel = None;
                        continue 'tasks;
                    }
                    if prepared_input.created_timelines {
                        guard.last_compressed_message_at =
                            Some(chrono::Utc::now().timestamp_millis());
                    }
                    guard.last_consolidated_at = Some(chrono::Utc::now().timestamp_millis());
                    guard.session_meta_snapshot()
                };
                if let Some((storage, meta)) = meta_snapshot
                    && let Err(e) = storage.upsert_session(&meta).await
                {
                    tracing::warn!(error = %e, "Failed to persist session meta after compression");
                }
                let history_out = prepared_input.messages;
                let runtime_context = prepared_input.runtime;

                let (turn_controller, turn_emitter, turn_receiver) = TurnController::start(
                    unified_str.clone(),
                    uuid::Uuid::new_v4().to_string(),
                );
                let initial_turn = turn_controller.snapshot();
                let active_turn_id = initial_turn.id.0.clone();
                let turn_target = crate::channels::TurnTarget {
                    channel: task_chan.clone(),
                    chat_id: task_cid.clone(),
                    session_id: unified_str.clone(),
                    reply_to: task_reply_to.clone(),
                    metadata: outbound_turn_metadata(&unified_str, &task_metadata),
                };
                let delivery_handle = match turn_delivery
                    .start(turn_target.clone(), turn_receiver)
                    .await
                {
                    Ok(handle) => Some(handle),
                    Err(error) => {
                        tracing::debug!(
                            channel = %task_chan,
                            error = %error,
                            "Live turn delivery unavailable; using ordinary final delivery"
                        );
                        None
                    }
                };
                let live_delivery_started = delivery_handle.is_some();
                {
                    let mut guard = session.lock().await;
                    if guard.worker_generation != worker_gen || guard.state_version != base_version {
                        turn_emitter.deactivate();
                        turn_controller.cancel(Some(
                            "session changed before model execution".to_string(),
                        ));
                        guard.current_cancel = None;
                        continue 'tasks;
                    }
                    guard.active_turn_emitter = Some(ActiveTurnEmitter {
                        turn_id: initial_turn.id.0.clone(),
                        emitter: turn_emitter.clone(),
                    });
                }
                let agent_turn = AgentTurnContext::new(
                    initial_turn.id.0.clone(),
                    initial_turn.message_id.clone(),
                    turn_emitter,
                );

                // Phase 2 + 3: LLM call with cancellation
                let session2 = session.clone();
                let bus2 = bus.clone();
                let chan2 = task_chan.clone();
                let cid2 = task_cid.clone();
                let unified_str2 = unified_str.clone();
                let task_metadata2 = task_metadata.clone();
                let task_reply_to2 = task_reply_to.clone();
                let title_supervisor = worker_supervisor.clone();
                let commit_delivery = turn_delivery.clone();
                let commit_target = turn_target.clone();
                let turn_lifecycle = &turn_controller;
                let pending_turn_deliveries = Arc::new(std::sync::Mutex::new(Vec::new()));
                let scoped_turn_deliveries = pending_turn_deliveries.clone();
                let process_future = async move {
                    let response_session_id = unified_str2.clone();
                    let process_result = crate::agent::sub_agent::DELEGATE_CONTEXT.scope(
                        crate::agent::DelegateContext {
                            session_id: unified_str2,
                            channel: chan2.clone(),
                            chat_id: cid2.clone(),
                        },
                        agent.process_streaming(history_out.clone(), agent_turn.clone()),
                    ).await;
                    let mut result = match process_result {
                        Ok(r) => r,
                        Err(AgentError::LlmError(ref msg))
                            if is_context_overflow_error(msg) =>
                        {
                            let (raw, mut retry_compressor, retry_base_version, new_window) = {
                                let guard = session2.lock().await;
                                let new_window =
                                    crate::agent::ContextCompressor::parse_context_limit_from_error(msg)
                                        .unwrap_or(guard.compressor_threshold());
                                tracing::warn!(
                                    new_window,
                                    error = %msg,
                                    "Context overflow in worker — retrying"
                                );
                                (
                                    guard.get_history().to_vec(),
                                    guard.fresh_context_compressor(),
                                    guard.state_version,
                                    new_window,
                                )
                            };
                            retry_compressor.set_context_window(new_window);
                            let retry_result =
                                match retry_compressor.compress_if_needed(raw).await {
                                    Ok(r) => r,
                                    Err(e) => {
                                        tracing::error!(error = %e, "Retry compression failed");
                                        fail_turn_with_partial(
                                            turn_lifecycle,
                                            &session2,
                                            format!("context overflow handling failed: {e}"),
                                        )
                                        .await;
                                        let err_outbound = OutboundMessage {
                                            channel: chan2,
                                            chat_id: cid2,
                                            content: "Context overflow handling failed."
                                                .to_string(),
                                            reply_to: task_reply_to2.clone(),
                                            media: vec![],
                                            metadata: outbound_turn_metadata(
                                                &response_session_id,
                                                &task_metadata2,
                                            ),
                                            delivery: None,
                                        };
                                        if !live_delivery_started {
                                            let _ = bus2.publish_outbound(err_outbound).await;
                                        }
                                        return;
                                    }
                                };

                            let meta_snapshot = {
                                let mut guard = session2.lock().await;
                                if guard.state_version != retry_base_version {
                                    tracing::warn!(
                                        session_id = %guard.id,
                                        "Session changed while retry-compressing after context overflow"
                                    );
                                    turn_lifecycle.cancel(Some(
                                        "session changed during context overflow recovery"
                                            .to_string(),
                                    ));
                                    return;
                                }
                                guard.compressor.set_context_window(new_window);
                                if retry_result.created_timelines {
                                    guard.last_compressed_message_at =
                                        Some(chrono::Utc::now().timestamp_millis());
                                }
                                guard.session_meta_snapshot()
                            };
                            if let Some((storage, meta)) = meta_snapshot
                                && let Err(e) = storage.upsert_session(&meta).await
                            {
                                tracing::warn!(error = %e, "Failed to persist session meta after retry compression");
                            }

                            let retry_history = runtime_context.assemble(retry_result.history);

                            match agent
                                .process_streaming(retry_history, agent_turn.clone())
                                .await
                            {
                                Ok(r) => r,
                                Err(e) => {
                                    tracing::error!(
                                        error = %e,
                                        "Agent retry after overflow failed"
                                    );
                                    fail_turn_with_partial(
                                        turn_lifecycle,
                                        &session2,
                                        e.to_string(),
                                    )
                                    .await;
                                    let err_outbound = OutboundMessage {
                                        channel: chan2,
                                        chat_id: cid2,
                                        content: format!("Processing error: {}", e),
                                        reply_to: task_reply_to2.clone(),
                                        media: vec![],
                                        metadata: outbound_turn_metadata(
                                            &response_session_id,
                                            &task_metadata2,
                                        ),
                                        delivery: None,
                                    };
                                    if !live_delivery_started {
                                        let _ = bus2.publish_outbound(err_outbound).await;
                                    }
                                    return;
                                }
                            }
                        }
                        Err(e) => {
                            tracing::error!(error = %e, "Agent processing error");
                            fail_turn_with_partial(
                                turn_lifecycle,
                                &session2,
                                e.to_string(),
                            )
                            .await;
                            let err_outbound = OutboundMessage {
                                channel: chan2,
                                chat_id: cid2,
                                content: format!("Processing error: {}", e),
                                reply_to: task_reply_to2.clone(),
                                media: vec![],
                                metadata: outbound_turn_metadata(
                                    &response_session_id,
                                    &task_metadata2,
                                ),
                                delivery: None,
                            };
                            if !live_delivery_started {
                                let _ = bus2.publish_outbound(err_outbound).await;
                            }
                            return;
                        }
                    };

                    let pending = take_current_turn_deliveries();
                    attach_pending_turn_deliveries(&mut result, pending);
                    let response_content = result.final_response.content;
                    let total_tokens = result.total_tokens;
                    let usage = result.usage;
                    {
                        let guard = session2.lock().await;
                        if guard.worker_generation != worker_gen
                            || guard.state_version != base_version
                        {
                            turn_lifecycle.cancel(Some(
                                "session changed before turn commit".to_string(),
                            ));
                            return;
                        }
                    }
                    let response = match finalize_turn_after_persistence(
                        turn_lifecycle,
                        usage,
                        append_persisted_messages_with_meta(
                            &session2,
                            result.emitted_messages,
                        ),
                    )
                    .await
                    {
                        Ok(committed_messages) => {
                            let mut guard = session2.lock().await;
                            let sent_count = guard.messages.len();
                            guard.compressor.set_last_api_info(sent_count, total_tokens);
                            Some((response_content, committed_messages))
                        }
                        Err(e) => {
                            tracing::error!(error = %e, "Failed to atomically persist agent turn");
                            None
                        }
                    };

                    let Some((response, committed_messages)) = response else {
                        let err_outbound = OutboundMessage {
                            channel: chan2,
                            chat_id: cid2,
                            content: "Failed to save the agent response, please try again."
                                .to_string(),
                            reply_to: task_reply_to2.clone(),
                            media: vec![],
                            metadata: outbound_turn_metadata(
                                &response_session_id,
                                &task_metadata2,
                            ),
                            delivery: None,
                        };
                        if !live_delivery_started {
                            let _ = bus2.publish_outbound(err_outbound).await;
                        }
                        return;
                    };

                    let delta = committed_turn_delta(&response_session_id, committed_messages);
                    if let Err(error) = commit_delivery.commit(&commit_target, delta).await {
                        tracing::warn!(error = %error, "Failed to publish committed turn delta");
                    }

                    schedule_title_generation(
                        session2.clone(),
                        title_supervisor,
                        &response_session_id,
                    )
                    .await;

                    if !live_delivery_started {
                        let outbound = OutboundMessage {
                            channel: chan2,
                            chat_id: cid2,
                            content: response,
                            reply_to: task_reply_to2.clone(),
                            media: vec![],
                            metadata: outbound_turn_metadata(
                                &response_session_id,
                                &task_metadata2,
                            ),
                            delivery: None,
                        };
                        let _ = bus2.publish_outbound(outbound).await;
                    }
                };
                let process_future =
                    CURRENT_TURN_ID.scope(Some(active_turn_id.clone()), process_future);
                let process_future = CURRENT_TURN_DELIVERIES
                    .scope(Some(scoped_turn_deliveries), process_future);

                tokio::select! {
                    () = process_future => {}
                    _ = cancel_rx => {
                        // cancelled — current_cancel already taken by /stop
                        let snapshot = turn_controller.snapshot();
                        if let Some(partial) = partial_assistant_with_pending_deliveries(
                            &snapshot,
                            CompletionStatus::Cancelled,
                            take_pending_turn_deliveries(&pending_turn_deliveries),
                        ) {
                            turn_controller.begin_finalizing();
                            match append_persisted_messages(&session, vec![partial]).await {
                                Ok(()) => {
                                    turn_controller.cancel(Some("stopped by user".to_string()));
                                }
                                Err(error) => {
                                    tracing::error!(error = %error, "Failed to persist cancelled partial turn");
                                    turn_controller.fail(format!(
                                        "failed to persist cancelled turn: {error}"
                                    ));
                                }
                            }
                        } else {
                            turn_controller.cancel(Some("stopped by user".to_string()));
                        }
                    }
                }

                if let Some(handle) = delivery_handle {
                    deliver_terminal_fallback(
                        handle,
                        &bus,
                        &turn_target,
                        &turn_controller,
                    )
                    .await;
                }

                // Clean up
                let mut guard = session.lock().await;
                if guard
                    .active_turn_emitter
                    .as_ref()
                    .is_some_and(|active| active.turn_id == active_turn_id)
                    && let Some(active) = guard.active_turn_emitter.take()
                {
                    active.emitter.deactivate();
                }
                if guard.worker_generation == worker_gen {
                    guard.current_cancel = None;
                }
            }
        }).await;
    });
}

impl SessionManager {
    ///
    /// Runs in a stateless manner: no session creation, no history persistence.
    /// The cron system prompt instructs the LLM to deliver results via the
    /// `send_message` tool, which handles both delivery and history writing
    /// on the target session.
    pub async fn handle_cron_message(
        &self,
        channel: &str,
        chat_id: &str,
        prompt: &str,
        job_id: &str,
        job_name: &str,
    ) -> Result<HandleResult, AgentError> {
        let skills_prompt = self.skills_loader.build_skills_prompt();

        let base_prompt = build_system_prompt(
            &self.provider_config.workspace_dir,
            &self.provider_config.model_id,
            &self.tools,
        );
        let cron_context = format!(
            "## 定时任务执行\n\n\
             你正在执行定时任务「{job_name}」({job_id})。\n\
             目标渠道: {channel}:{chat_id}\n\n\
             规则:\n\
             - 这不是聊天对话，没有用户会直接看到你的输出\n\
             - 你必须使用 send_message 工具将最终结果发送到目标渠道\n\
             - send_message 格式: target_chat_id=\"{channel}:{chat_id}\", content=\"消息内容\"\n\
             - 可以调用其他工具收集信息、处理任务，但最终消息必须通过 send_message 发送\n\
             - 只输出最终消息内容，不要输出中间思考过程或分析！"
        );
        let full_system_prompt =
            format!("{}\n\n{}\n\n{}", base_prompt, skills_prompt, cron_context);

        let history = vec![
            ChatMessage::system(full_system_prompt),
            ChatMessage::user(prompt),
        ];

        let agent = self.create_cron_agent()?;
        let source_session = format!("cron:{}", job_name);
        let result = CURRENT_SOURCE_SESSION
            .scope(Some(source_session), async { agent.process(history).await })
            .await
            .inspect_err(|e| {
                tracing::error!(error = %e, job_id = %job_id, "Cron agent processing error");
            })?;

        Ok(HandleResult::AgentResponse(result.final_response.content))
    }

    /// Execute a scheduler-managed task. The agent returns a result but cannot
    /// deliver it itself; Scheduler applies the configured delivery policy.
    pub async fn handle_managed_scheduled_message(
        &self,
        prompt: &str,
        job_id: &str,
        job_name: &str,
        monitor: bool,
    ) -> Result<String, AgentError> {
        let (agent, tools) = self.create_managed_scheduled_agent()?;
        let base_prompt = build_system_prompt(
            &self.provider_config.workspace_dir,
            &self.provider_config.model_id,
            &tools,
        );
        let skills_prompt = self.skills_loader.build_skills_prompt();
        let result_contract = if monitor {
            "这是无人值守巡检。完成必要检查后：一切正常且无需用户关注时，只返回 NO_REPLY[INFO]: <简短原因>；发现问题时返回简洁、可操作的告警；无法完成时返回 NO_REPLY[FAIL]: <原因>；因安全或权限拒绝时返回 NO_REPLY[REFUSE]: <原因>。不要调用 send_message，不要把不确定当作正常。"
        } else {
            "这是 Scheduler 托管投递的定时任务。完成任务后只返回应交付给用户的最终内容，不要调用 send_message。"
        };
        let system = format!(
            "{base_prompt}\n\n{skills_prompt}\n\n## 定时任务执行\n任务「{job_name}」({job_id})。\n{result_contract}"
        );
        let history = vec![ChatMessage::system(system), ChatMessage::user(prompt)];
        let source_session = format!("cron:{job_id}");
        let result = CURRENT_SOURCE_SESSION
            .scope(Some(source_session), async { agent.process(history).await })
            .await?;
        Ok(result.final_response.content)
    }

    pub async fn clear_session_history(
        &self,
        unified_id: &UnifiedSessionId,
    ) -> Result<(), AgentError> {
        let session = self.get_or_create_session(unified_id).await?;
        let persistence_lock = { session.lock().await.persistence_lock.clone() };
        let _persistence_guard = persistence_lock.lock().await;
        let (storage, session_id, meta_snapshot) = {
            let mut session_guard = session.lock().await;
            // Clear in-memory
            session_guard.messages.clear();
            session_guard.seq_counter = 1;
            session_guard.total_message_count = 0;
            session_guard.message_count = 0;
            session_guard.last_consolidated_at = None;
            session_guard.last_compressed_message_at = None;
            session_guard.state_version = session_guard.state_version.wrapping_add(1);
            (
                session_guard.storage.clone(),
                session_guard.id.to_string(),
                session_guard.session_meta_snapshot(),
            )
        };
        // Clear Storage outside the session lock.
        if let Some(storage) = storage {
            storage
                .clear_messages(&session_id)
                .await
                .map_err(|e| AgentError::Other(format!("failed to clear messages: {}", e)))?;
        }
        if let Some((storage, meta)) = meta_snapshot {
            storage.upsert_session(&meta).await.map_err(|e| {
                AgentError::Other(format!("failed to persist cleared session: {}", e))
            })?;
        }
        Ok(())
    }
}

fn format_task_notification(
    task_id: &str,
    status: &crate::agent::TaskStatus,
    summary: &str,
) -> String {
    match status {
        crate::agent::TaskStatus::Completed => format!(
            "📋 后台任务完成\n\n任务 ID: {}\n\n结果:\n{}",
            task_id, summary
        ),
        crate::agent::TaskStatus::Failed(err) => {
            format!("📋 后台任务失败\n\n任务 ID: {}\n错误: {}", task_id, err)
        }
        crate::agent::TaskStatus::Cancelled => format!("📋 后台任务已取消\n\n任务 ID: {}", task_id),
        crate::agent::TaskStatus::TimedOut => format!("📋 后台任务超时\n\n任务 ID: {}", task_id),
    }
}

#[cfg(test)]
mod slash_command_tests {
    use super::resolve_slash_command;

    #[test]
    fn aliases_resolve_to_their_canonical_command() {
        assert_eq!(
            resolve_slash_command("?").map(|command| command.name),
            Some("?")
        );
        assert_eq!(
            resolve_slash_command("help").map(|command| command.name),
            Some("?")
        );
        assert_eq!(
            resolve_slash_command("/help").map(|command| command.name),
            Some("?")
        );
        assert_eq!(
            resolve_slash_command("reload").map(|command| command.name),
            Some("reload")
        );
        assert!(resolve_slash_command("unknown").is_none());
    }
}