feat: McpClientManager 传递 cwd 参数至 connect_stdio

.gitignore

@@ -35,3 +35,4 @@ uv.lock
 node_modules
 logs
 dist
+.trae

Cargo.toml

@@ -13,6 +13,7 @@ serde_yaml = "0.9"
 async-trait = "0.1"
 thiserror = "2.0.18"
 tokio = { version = "1.0", features = ["full"] }
+tokio-util = { version = "0.7", features = ["rt"] }
 uuid = { version = "1.0", features = ["v4"] }
 axum = { version = "0.8", features = ["ws"] }
 tokio-tungstenite = { version = "0.29.0", features = ["rustls-tls-webpki-roots", "rustls"] }

src/channels/cli.rs

@@ -2,6 +2,7 @@ use async_trait::async_trait;
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::{RwLock, mpsc};
+use tokio_util::sync::CancellationToken;
 
 use crate::bus::{MessageBus, OutboundMessage};
 use crate::protocol::WsOutbound;
@@ -18,34 +19,36 @@ struct CliConnection {
 #[derive(Clone)]
 pub struct CliChannel {
     connections: Arc<RwLock<HashMap<String, CliConnection>>>,
+    /// 全局关闭信号：stop() 时 cancel，所有 handle_socket 同时收到信号
+    shutdown_token: CancellationToken,
 }
 
 impl CliChannel {
     pub fn new() -> Self {
         Self {
             connections: Arc::new(RwLock::new(HashMap::new())),
+            shutdown_token: CancellationToken::new(),
         }
     }
 
+    /// 注册连接，返回 CancellationToken 供 handle_socket 监听
+    ///
+    /// 注意：此方法不会向被替换的旧连接发送关闭信号。
+    /// 关闭信号仅由 stop() 统一触发，避免正常消息处理时误关连接。
     pub async fn register_connection(
         &self,
         session_id: impl Into<String>,
         connection_id: impl Into<String>,
         sender: mpsc::Sender<WsOutbound>,
-    ) {
-        let session_id = session_id.into();
-        let connection_id = connection_id.into();
-        let previous = self.connections.write().await.insert(
-            session_id.clone(),
+    ) -> CancellationToken {
+        self.connections.write().await.insert(
+            session_id.into(),
             CliConnection {
-                connection_id: connection_id.clone(),
+                connection_id: connection_id.into(),
                 sender,
             },
         );
-
-        if previous.is_some() {
-            tracing::info!(session_id = %session_id, connection_id = %connection_id, "CLI session sender replaced");
-        }
+        self.shutdown_token.clone()
     }
 
     pub async fn unregister_connection(&self, connection_id: &str) {
@@ -77,7 +80,11 @@ impl Channel for CliChannel {
     }
 
     async fn stop(&self) -> Result<(), ChannelError> {
+        // 先 cancel 所有 handle_socket，再清空 connections
+        self.shutdown_token.cancel();
+        let count = self.connections.read().await.len();
         self.connections.write().await.clear();
+        tracing::info!(connection_count = count, "CliChannel stopped, all connections signaled to close");
         Ok(())
     }
 
@@ -111,7 +118,7 @@ mod tests {
     async fn test_cli_channel_sends_to_registered_session() {
         let channel = CliChannel::new();
         let (sender, mut receiver) = mpsc::channel(4);
-        channel
+        let _token = channel
             .register_connection("session-1", "conn-1", sender)
             .await;
 
@@ -135,7 +142,7 @@ mod tests {
     async fn test_cli_channel_unregisters_connection_sessions() {
         let channel = CliChannel::new();
         let (sender, _receiver) = mpsc::channel(4);
-        channel
+        let _token = channel
             .register_connection("session-1", "conn-1", sender)
             .await;
         channel.unregister_connection("conn-1").await;
@@ -154,4 +161,36 @@ mod tests {
 
         assert!(error.to_string().contains("No active CLI connection"));
     }
+
+    #[tokio::test]
+    async fn test_stop_cancels_token() {
+        let channel = CliChannel::new();
+        let (sender, _receiver) = mpsc::channel(4);
+        let token = channel
+            .register_connection("session-1", "conn-1", sender)
+            .await;
+
+        assert!(!token.is_cancelled());
+        channel.stop().await.unwrap();
+        assert!(token.is_cancelled());
+    }
+
+    #[tokio::test]
+    async fn test_register_connection_does_not_cancel_old_token() {
+        // 验证：register_connection 替换旧连接时不会 cancel token
+        // 只有 stop() 才会 cancel
+        let channel = CliChannel::new();
+        let (sender1, _receiver1) = mpsc::channel(4);
+        let token1 = channel
+            .register_connection("session-1", "conn-1", sender1)
+            .await;
+
+        let (sender2, _receiver2) = mpsc::channel(4);
+        let _token2 = channel
+            .register_connection("session-1", "conn-2", sender2)
+            .await;
+
+        // 旧 token 不应被 cancel（只有 stop() 才 cancel）
+        assert!(!token1.is_cancelled());
+    }
 }

src/gateway/agent_factory.rs

@@ -15,6 +15,8 @@ pub(crate) struct AgentFactory {
     skills: Arc<SkillRuntime>,
     reinject_every: usize,
     prompt_repository: Arc<dyn PromptInjectionRepository>,
+    /// 实例创建时间戳（用于区分新旧 AgentFactory 实例）
+    instance_id: u64,
 }
 
 pub(crate) struct AgentBuildRequest<'a> {
@@ -37,17 +39,36 @@ impl AgentFactory {
         reinject_every: usize,
         prompt_repository: Arc<dyn PromptInjectionRepository>,
     ) -> Self {
+        // 使用 Arc 指针地址作为实例标识符，用于区分新旧 AgentFactory 实例
+        let instance_id = Arc::as_ptr(&tools) as u64;
+        tracing::info!(
+            instance_id = instance_id,
+            tool_count = tools.tool_names().len(),
+            "AgentFactory::new created"
+        );
         Self {
             tools,
             skills,
             reinject_every,
             prompt_repository,
+            instance_id,
         }
     }
 
     pub(crate) fn create(&self, request: AgentBuildRequest<'_>) -> Result<AgentLoop, AgentError> {
         let session_id = persistent_session_id(request.channel_name, request.session_chat_id);
 
+        // 诊断日志：记录 agent 实际使用的配置和实例 ID
+        tracing::info!(
+            instance_id = self.instance_id,
+            channel = %request.channel_name,
+            session_id = %session_id,
+            provider = %request.provider_config.name,
+            model_id = %request.provider_config.model_id,
+            tool_count = self.tools.tool_names().len(),
+            "AgentFactory: creating agent with config"
+        );
+
         // 创建组合的系统提示词提供者
         let system_prompt_provider = Arc::new(CompositeSystemPromptProvider::new(vec![
             Box::new(AgentPromptProvider::new(

src/gateway/agent_prompt_provider.rs

@@ -55,6 +55,15 @@ impl SystemPromptProvider for AgentPromptProvider {
             return None;
         }
 
+        // 诊断日志：记录系统提示词实际使用的模型配置（用于排查"配置不生效"问题）
+        tracing::info!(
+            session_id = ?context.session_id,
+            chat_id = %context.chat_id,
+            provider = %self.provider_config.name,
+            model_id = %self.provider_config.model_id,
+            "AgentPromptProvider: building system prompt with model config"
+        );
+
         // 加载 Agent 提示词（AGENT.md + builtin + MEMORY_SUMMARY.md）
         let agent_prompt = load_agent_prompt().ok().flatten()?;
 

src/gateway/execution.rs

@@ -6,7 +6,7 @@ use crate::agent::{AgentError, AgentProcessResult, EmittedMessageHandler, Persis
 use crate::bus::message::ToolMessageState;
 use crate::bus::{ChatMessage, MediaItem, OutboundMessage, SYSTEM_CONTEXT_SCHEDULED_PROMPT};
 use crate::config::LLMProviderConfig;
-use crate::storage::ConversationRepository;
+use crate::storage::{persistent_session_id, ConversationRepository};
 use tokio::sync::Mutex;
 
 use super::compaction::schedule_background_history_compaction;
@@ -252,7 +252,7 @@ impl AgentExecutionService {
 
         // 构建系统提示词上下文
         let system_prompt_context = SystemPromptContext {
-            session_id: Some(format!("{}:{}", request.channel_name, request.chat_id)),
+            session_id: Some(persistent_session_id(request.channel_name, request.chat_id)),
             chat_id: request.chat_id.to_string(),
             user_message_count,
         };
@@ -356,7 +356,7 @@ impl AgentExecutionService {
 
         // 构建系统提示词上下文
         let system_prompt_context = SystemPromptContext {
-            session_id: Some(format!("{}:{}", request.channel_name, request.chat_id)),
+            session_id: Some(persistent_session_id(request.channel_name, request.chat_id)),
             chat_id: request.chat_id.to_string(),
             user_message_count,
         };

src/gateway/mod.rs

@@ -104,6 +104,12 @@ impl GatewayState {
             Some(bus.clone()),
         )?;
 
+        // 诊断日志：记录新 GatewayState 的创建（用于排查重启后是否使用了新状态）
+        tracing::info!(
+            mcp_manager_present = mcp_manager.is_some(),
+            "GatewayState::from_config: new GatewayState created"
+        );
+
         let cancel_manager = CancelManager::new();
 
         Ok(Self {

src/gateway/runtime.rs

@@ -224,6 +224,15 @@ pub(crate) fn build_session_manager_with_sender(
     // Build base tools
     let tools = factory.build();
 
+    // 诊断日志：记录 MCP 初始化状态和基础工具数量
+    tracing::info!(
+        mcp_enabled = mcp_initializer.is_enabled(),
+        mcp_tools_collected = mcp_tools_for_subagents.len(),
+        base_tool_count = tools.tool_names().len(),
+        base_tools = ?tools.tool_names(),
+        "build_session_manager: base tools built (before MCP registration)"
+    );
+
     // Register MCP tools to main agent (async)
     // Note: MCP tools for subagents are already collected above
     if mcp_initializer.is_enabled() {
@@ -240,6 +249,13 @@ pub(crate) fn build_session_manager_with_sender(
 
     let tools = Arc::new(tools);
 
+    // 诊断日志：记录最终工具数量（含 MCP 工具，如果启用）
+    tracing::info!(
+        final_tool_count = tools.tool_names().len(),
+        final_tools = ?tools.tool_names(),
+        "build_session_manager: final tools registered to main agent"
+    );
+
     let prompt_repository: Arc<dyn PromptInjectionRepository> = store.clone();
     let agent_factory = AgentFactory::new(
         tools.clone(),

src/gateway/ws.rs

@@ -28,6 +28,7 @@ use crate::command::handlers::switch_topic::SwitchTopicCommandHandler;
 use crate::gateway::agent_prompt_provider::AgentPromptProvider;
 use crate::protocol::{WsInbound, WsOutbound, MediaSummary, parse_inbound, serialize_outbound};
 use crate::skills::SkillPromptProvider;
+use crate::storage::persistent_session_id;
 use crate::tools::task::repository::TaskRepository;
 use crate::tools::task::types::TaskSessionState;
 use axum::extract::State;
@@ -157,9 +158,10 @@ async fn handle_socket(ws: WebSocket, state: Arc<GatewayState>) {
     };
 
     let runtime_session_id = uuid::Uuid::new_v4().to_string();
-    let mut current_session_id = initial_record.id.clone();
+    // 清理数据库中可能已被污染的多重前缀 session_id（幂等处理）
+    let mut current_session_id = persistent_session_id(WS_CHANNEL_NAME, &initial_record.id);
     let mut current_topic_id: Option<String> = None;
-    state
+    let shutdown_token = state
         .channel_manager
         .websocket_channel()
         .register_connection(
@@ -232,50 +234,63 @@ async fn handle_socket(ws: WebSocket, state: Arc<GatewayState>) {
         }
     });
 
-    while let Some(msg) = ws_receiver.next().await {
-        match msg {
-            Ok(WsMessage::Text(text)) => {
-                let text = text.to_string();
-                match parse_inbound(&text) {
-                    Ok(inbound) => {
-                        if let Err(e) = handle_inbound(
-                            &state,
-                            &sender,
-                            &runtime_session_id,
-                            &mut current_session_id,
-                            &mut current_topic_id,
-                            inbound,
-                        )
-                        .await
-                        {
-                            tracing::warn!(error = %e, session_id = %current_session_id, "Failed to handle inbound message");
-                            let _ = sender
-                                .send(WsOutbound::Error {
-                                    timestamp: Some(crate::protocol::now_timestamp()),
-                                    code:"SESSION_ERROR".to_string(),
-                                    message: e.to_string(),
-                                })
-                                .await;
-                        }
-                    }
-                    Err(e) => {
-                        tracing::warn!(error = %e, "Failed to parse inbound message");
-                        let _ = sender
-                            .send(WsOutbound::Error {
-                                timestamp: Some(crate::protocol::now_timestamp()),
-                                code:"PARSE_ERROR".to_string(),
-                                message: e.to_string(),
-                            })
-                            .await;
-                    }
-                }
-            }
-            Ok(WsMessage::Close(_)) | Err(_) => {
-                #[cfg(debug_assertions)]
-                tracing::debug!(session_id = %runtime_session_id, "WebSocket closed");
+    loop {
+        tokio::select! {
+            // 监听 shutdown 信号（来自 CliChannel::stop()）
+            _ = shutdown_token.cancelled() => {
+                tracing::info!(session_id = %current_session_id, "WebSocket shutdown signal received, closing connection");
                 break;
             }
-            _ => {}
+            // 监听 WebSocket 消息
+            msg = ws_receiver.next() => {
+                let Some(msg) = msg else {
+                    break;
+                };
+                match msg {
+                    Ok(WsMessage::Text(text)) => {
+                        let text = text.to_string();
+                        match parse_inbound(&text) {
+                            Ok(inbound) => {
+                                if let Err(e) = handle_inbound(
+                                    &state,
+                                    &sender,
+                                    &runtime_session_id,
+                                    &mut current_session_id,
+                                    &mut current_topic_id,
+                                    inbound,
+                                )
+                                .await
+                                {
+                                    tracing::warn!(error = %e, session_id = %current_session_id, "Failed to handle inbound message");
+                                    let _ = sender
+                                        .send(WsOutbound::Error {
+                                            timestamp: Some(crate::protocol::now_timestamp()),
+                                            code:"SESSION_ERROR".to_string(),
+                                            message: e.to_string(),
+                                        })
+                                        .await;
+                                }
+                            }
+                            Err(e) => {
+                                tracing::warn!(error = %e, "Failed to parse inbound message");
+                                let _ = sender
+                                    .send(WsOutbound::Error {
+                                        timestamp: Some(crate::protocol::now_timestamp()),
+                                        code:"PARSE_ERROR".to_string(),
+                                        message: e.to_string(),
+                                    })
+                                    .await;
+                            }
+                        }
+                    }
+                    Ok(WsMessage::Close(_)) | Err(_) => {
+                        #[cfg(debug_assertions)]
+                        tracing::debug!(session_id = %runtime_session_id, "WebSocket closed");
+                        break;
+                    }
+                    _ => {}
+                }
+            }
         }
     }
 
@@ -307,7 +322,7 @@ async fn handle_inbound(
             let chat_id = chat_id.unwrap_or_else(|| current_session_id.clone());
             let sender_id = resolve_ws_sender_id(sender_id.as_deref(), runtime_session_id);
 
-            state
+            let _ = state
                 .channel_manager
                 .websocket_channel()
                 .register_connection(
@@ -479,7 +494,7 @@ async fn handle_inbound(
                         "Updating current_session_id"
                     );
                     *current_session_id = session_id.clone();
-                    state
+                    let _ = state
                         .channel_manager
                         .websocket_channel()
                         .register_connection(

src/mcp/client.rs

@@ -182,8 +182,8 @@ impl McpClientManager {
 
         let transport = config.transport().map_err(|e| anyhow::anyhow!("{}", e))?;
         let client = match transport {
-            McpTransportConfig::Stdio { command, args, env } => {
-                self.connect_stdio(&command, &args, &env).await?
+            McpTransportConfig::Stdio { command, args, env, cwd } => {
+                self.connect_stdio(&command, &args, &env, &cwd).await?
             }
             McpTransportConfig::Http { url, headers } => {
                 self.connect_http(&url, &headers).await?
@@ -222,6 +222,7 @@ impl McpClientManager {
         command: &str,
         args: &[String],
         env: &HashMap<String, String>,
+        _cwd: &Option<std::path::PathBuf>,
     ) -> anyhow::Result<McpClient> {
         let mut cmd = Command::new(command);
         cmd.args(args);

src/mcp/config.rs

@@ -63,6 +63,9 @@ pub struct McpServerConfig {
     /// Environment variables for stdio transport
     #[serde(default)]
     pub env: Option<HashMap<String, String>>,
+    /// Working directory for stdio transport (optional). If set, child process runs in this dir.
+    #[serde(default)]
+    pub cwd: Option<String>,
 
     // HTTP transport fields
     /// Base URL for HTTP transport (Claude Desktop compatible naming)
@@ -99,6 +102,7 @@ impl McpServerConfig {
                     command,
                     args: self.args.clone().unwrap_or_default(),
                     env: self.env.clone().unwrap_or_default(),
+                    cwd: self.cwd.as_ref().map(|s| std::path::PathBuf::from(s)),
                 })
             }
             "http" | "streamableHttp" => {
@@ -124,6 +128,7 @@ impl McpServerConfig {
             command: Some(command.into()),
             args: Some(args),
             env: Some(HashMap::new()),
+            cwd: None,
             base_url: None,
             headers: None,
             description: None,
@@ -139,6 +144,7 @@ impl McpServerConfig {
             command: None,
             args: None,
             env: None,
+            cwd: None,
             base_url: Some(url.into()),
             headers: Some(HashMap::new()),
             description: None,
@@ -154,6 +160,7 @@ pub enum McpTransportConfig {
         command: String,
         args: Vec<String>,
         env: HashMap<String, String>,
+        cwd: Option<std::path::PathBuf>,
     },
     /// HTTP transport: connect to a remote server (Streamable HTTP)
     Http {
@@ -300,6 +307,7 @@ mod tests {
             command: Some("npx".to_string()),
             args: Some(vec!["-y".to_string(), "server".to_string()]),
             env: None,
+            cwd: None,
             base_url: None,
             headers: None,
             description: None,
@@ -317,6 +325,7 @@ mod tests {
             command: Some("npx".to_string()),
             args: Some(vec!["-y".to_string(), "server".to_string()]),
             env: None,
+            cwd: None,
             base_url: None,
             headers: None,
             description: None,
@@ -335,6 +344,7 @@ mod tests {
             command: None,
             args: None,
             env: None,
+            cwd: None,
             base_url: None,
             headers: None,
             description: None,
@@ -349,6 +359,7 @@ mod tests {
             command: None,
             args: None,
             env: None,
+            cwd: None,
             base_url: None,
             headers: None,
             description: None,
@@ -363,6 +374,7 @@ mod tests {
             command: Some("cmd".to_string()),
             args: None,
             env: None,
+            cwd: None,
             base_url: None,
             headers: None,
             description: None,
@@ -385,4 +397,27 @@ mod tests {
         assert!(matches!(transport_http, McpTransportConfig::Http { .. }));
         assert!(matches!(transport_streamable, McpTransportConfig::Http { .. }));
     }
+
+    #[test]
+    fn test_stdio_cwd_field() {
+        let json = r#"{"mcpServers": {"test": {"type": "stdio", "command": "uv", "args": ["run", "server.py"], "cwd": "/some/path", "isActive": true}}}"#;
+        let config: McpConfig = serde_json::from_str(json).unwrap();
+        let server = config.mcp_servers.get("test").unwrap();
+        assert_eq!(server.cwd.as_deref(), Some("/some/path"));
+        let transport = server.transport().unwrap();
+        match transport {
+            McpTransportConfig::Stdio { cwd, .. } => {
+                assert_eq!(cwd, Some(std::path::PathBuf::from("/some/path")));
+            }
+            _ => panic!("Expected stdio transport"),
+        }
+    }
+
+    #[test]
+    fn test_stdio_cwd_defaults_to_none() {
+        let json = r#"{"mcpServers": {"test": {"type": "stdio", "command": "uv", "args": ["run", "server.py"]}}}"#;
+        let config: McpConfig = serde_json::from_str(json).unwrap();
+        let server = config.mcp_servers.get("test").unwrap();
+        assert!(server.cwd.is_none());
+    }
 }

src/storage/mod.rs

@@ -1601,6 +1601,13 @@ impl SessionStore {
 }
 
 pub fn persistent_session_id(channel_name: &str, chat_id: &str) -> String {
+    // 幂等：循环去除已存在的 "{channel_name}:" 前缀，防止前缀累积
+    let prefix = format!("{}:", channel_name);
+    let mut chat_id = chat_id;
+    while chat_id.starts_with(&prefix) {
+        chat_id = &chat_id[prefix.len()..];
+    }
+
     if channel_name == "cli" || channel_name == "websocket" {
         chat_id.to_string()
     } else {
@@ -2288,8 +2295,12 @@ mod tests {
     #[test]
     fn test_persistent_session_id_for_cli_and_channel() {
         assert_eq!(persistent_session_id("cli", "abc"), "abc");
-        assert_eq!(persistent_session_id("websocket", "websocket:abc"), "websocket:abc");
+        // 幂等：已带前缀的 chat_id 会被清理，不会累积前缀
+        assert_eq!(persistent_session_id("websocket", "websocket:abc"), "abc");
+        assert_eq!(persistent_session_id("websocket", "websocket:websocket:abc"), "abc");
         assert_eq!(persistent_session_id(TEST_CHANNEL, "abc"), "test-channel:abc");
+        // 其他通道也幂等
+        assert_eq!(persistent_session_id(TEST_CHANNEL, "test-channel:abc"), "test-channel:abc");
     }
 
     #[test]