feat: 实现聊天消息的串行锁，确保同一聊天的消息处理串行执行

src/gateway/execution.rs

@@ -200,6 +200,17 @@ impl AgentExecutionService {
         &self,
         request: MessageExecutionRequest<'_>,
     ) -> Result<Vec<OutboundMessage>, AgentError> {
+        // 获取该 chat 的串行锁（通过短暂获取 session 锁）
+        // 同一 chat 的消息处理必须串行执行，防止并发 loop 操作同一历史的不同快照
+        let serial_lock = {
+            let mut session_guard = request.session.lock().await;
+            session_guard.chat_serial_lock(request.chat_id)
+        };
+
+        // 等待该 chat 的前一条消息处理完成（含压缩）
+        // await 串行锁时不持有 session 锁，其他 chat 的消息可以正常处理
+        let _serial_guard = serial_lock.lock().await;
+
         let (history, agent, user_message, user_message_count, original_topic_id) = {
             let mut session_guard = request.session.lock().await;
 
@@ -280,6 +291,15 @@ impl AgentExecutionService {
         &self,
         request: ScheduledExecutionRequest<'_>,
     ) -> Result<Vec<OutboundMessage>, AgentError> {
+        // 获取该 chat 的串行锁（与普通消息路径共享，保证串行执行）
+        let serial_lock = {
+            let mut session_guard = request.session.lock().await;
+            session_guard.chat_serial_lock(request.chat_id)
+        };
+
+        // 等待该 chat 的前一条消息处理完成（含压缩）
+        let _serial_guard = serial_lock.lock().await;
+
         let (history, mut agent, user_message, user_message_count, original_topic_id, store, session_id) = {
             let mut session_guard = request.session.lock().await;
 
@@ -479,4 +499,62 @@ mod tests {
         assert!(!should_display_message_to_user(false, &message));
         assert!(should_display_message_to_user(true, &message));
     }
+
+    /// 对抗性测试：同一 chat 的串行锁被持有时，第二次获取应阻塞
+    #[tokio::test]
+    async fn test_chat_serial_lock_blocks_concurrent_access() {
+        let lock = std::sync::Arc::new(tokio::sync::Mutex::new(()));
+        let _guard1 = lock.lock().await;
+
+        // 第二次获取应阻塞，1ms 超时验证
+        let result = tokio::time::timeout(
+            std::time::Duration::from_millis(1),
+            lock.lock(),
+        )
+        .await;
+
+        assert!(result.is_err(), "第二次获取同一锁应阻塞");
+    }
+
+    /// 对抗性测试：不同 chat 的串行锁互不影响，可同时获取
+    #[tokio::test]
+    async fn test_different_chat_locks_independent() {
+        let lock_a = std::sync::Arc::new(tokio::sync::Mutex::new(()));
+        let lock_b = std::sync::Arc::new(tokio::sync::Mutex::new(()));
+
+        let _guard_a = lock_a.lock().await;
+
+        // 不同锁应立即可获取
+        let result = tokio::time::timeout(
+            std::time::Duration::from_millis(100),
+            lock_b.lock(),
+        )
+        .await;
+
+        assert!(result.is_ok(), "不同 chat 的锁应互不影响");
+    }
+
+    /// 对抗性测试：错误返回路径锁被正确释放（RAII 保证）
+    #[tokio::test]
+    async fn test_serial_lock_released_on_error() {
+        let lock = std::sync::Arc::new(tokio::sync::Mutex::new(()));
+
+        // 模拟 prepare_and_execute_message 的错误路径：
+        // 获取锁 → 返回错误 → 锁应通过 RAII 释放
+        {
+            let _serial_guard = lock.lock().await;
+            // 模拟错误返回（`?` 或 `Err` 分支）
+            let _result: Result<(), AgentError> = Err(AgentError::Other("simulated".to_string()));
+            // _serial_guard 在此块结束时 Drop，释放锁
+        }
+
+        // 锁应已释放，可再次获取
+        let result = tokio::time::timeout(
+            std::time::Duration::from_millis(100),
+            lock.lock(),
+        )
+        .await;
+
+        assert!(result.is_ok(), "错误返回后锁应已释放");
+    }
 }

src/gateway/session.rs

@@ -533,6 +533,12 @@ impl Session {
         &self.compressor
     }
 
+    /// 获取该 chat 的串行化锁。
+    /// 同一 chat 的消息处理（agent loop + 压缩）共享此锁，保证串行执行。
+    pub(crate) fn chat_serial_lock(&mut self, chat_id: &str) -> Arc<tokio::sync::Mutex<()>> {
+        self.history.chat_serial_lock(chat_id)
+    }
+
     pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> {
         // 如果当前有 topic，加载该 topic 的消息（按 session_id 过滤，排除子智能体消息）
         if let Some(topic_id) = self.history.chat_topic(chat_id) {

src/gateway/session_history.rs

@@ -21,6 +21,10 @@ pub(crate) struct SessionHistory {
     chat_topic_ids: HashMap<String, String>, // 每个 chat 的当前 topic
     history_topic_ids: HashMap<String, String>, // 每个 chat 的历史所对应的话题
     compression_in_flight: HashSet<String>,
+    /// 按 chat_id 的串行化锁。
+    /// 同一 chat 的消息处理（agent loop + 压缩）必须串行执行，
+    /// 防止并发 loop 操作同一历史的不同快照产生交错序列。
+    chat_serial_locks: HashMap<String, Arc<tokio::sync::Mutex<()>>>,
     conversations: Arc<dyn ConversationRepository>,
     skill_events: Arc<dyn SkillEventRepository>,
 }
@@ -37,11 +41,21 @@ impl SessionHistory {
             chat_topic_ids: HashMap::new(),
             history_topic_ids: HashMap::new(),
             compression_in_flight: HashSet::new(),
+            chat_serial_locks: HashMap::new(),
             conversations,
             skill_events,
         }
     }
 
+    /// 获取或创建该 chat 的串行化锁。
+    /// 同一 chat 的所有消息处理共享同一个锁，保证串行执行。
+    pub(crate) fn chat_serial_lock(&mut self, chat_id: &str) -> Arc<tokio::sync::Mutex<()>> {
+        self.chat_serial_locks
+            .entry(chat_id.to_string())
+            .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
+            .clone()
+    }
+
     pub(crate) fn persistent_session_id(&self, chat_id: &str) -> String {
         persistent_session_id(&self.channel_name, chat_id)
     }

src/providers/openai.rs

@@ -394,6 +394,17 @@ impl OpenAIProvider {
         let status = resp.status();
         if !status.is_success() {
             let text = resp.text().await.unwrap_or_default();
+            let sequence = format_message_sequence(&body);
+            tracing::error!(
+                provider = %self.name,
+                model = %self.model_id,
+                url = %url,
+                status = %status,
+                response_len = text.len(),
+                response_body = %text,
+                sequence = ?sequence,
+                "OpenAI-compatible streaming API request failed"
+            );
             return Err(format!("API error {}: {}", status, text).into());
         }
 
@@ -871,52 +882,42 @@ impl OpenAIProvider {
             body["tools"] = json!(tools);
         }
 
-        // Diagnostic: log the final message sequence when tool_calls are involved.
-        // This captures the exact sequence sent to the API, making 400 errors
-        // like "insufficient tool messages following tool_calls message" easy to
-        // diagnose.
-        let has_tool_calls = body["messages"].as_array()
-            .map(|msgs| msgs.iter().any(|m| m.get("tool_calls").is_some()))
-            .unwrap_or(false);
-        if has_tool_calls {
-            let sequence: Vec<String> = body["messages"].as_array()
-                .map(|msgs| msgs.iter().enumerate().map(|(i, m)| {
-                    let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("?");
-                    match role {
-                        "assistant" => {
-                            let tc_count = m.get("tool_calls")
-                                .and_then(|t| t.as_array())
-                                .map(|a| a.len())
-                                .unwrap_or(0);
-                            if tc_count > 0 {
-                                format!("[{}] assistant(tool_calls={})", i, tc_count)
-                            } else {
-                                format!("[{}] assistant", i)
-                            }
-                        }
-                        "tool" => {
-                            let tcid = m.get("tool_call_id")
-                                .and_then(|t| t.as_str())
-                                .unwrap_or("??");
-                            format!("[{}] tool(id={})", i, tcid)
-                        }
-                        _ => format!("[{}] {}", i, role),
-                    }
-                }).collect())
-                .unwrap_or_default();
-            tracing::info!(
-                provider = %self.name,
-                model = %self.model_id,
-                message_count = sequence.len(),
-                sequence = ?sequence,
-                "build_request_body: final message sequence with tool_calls"
-            );
-        }
-
         body
     }
 }
 
+/// Builds a compact, human-readable summary of the message sequence in `body`
+/// for diagnostic logging. Only emitted on API errors (e.g. 400 responses) to
+/// avoid flooding logs on every request — see callers in `chat` and
+/// `chat_streaming_internal`.
+fn format_message_sequence(body: &Value) -> Vec<String> {
+    body["messages"].as_array()
+        .map(|msgs| msgs.iter().enumerate().map(|(i, m)| {
+            let role = m.get("role").and_then(|r| r.as_str()).unwrap_or("?");
+            match role {
+                "assistant" => {
+                    let tc_count = m.get("tool_calls")
+                        .and_then(|t| t.as_array())
+                        .map(|a| a.len())
+                        .unwrap_or(0);
+                    if tc_count > 0 {
+                        format!("[{}] assistant(tool_calls={})", i, tc_count)
+                    } else {
+                        format!("[{}] assistant", i)
+                    }
+                }
+                "tool" => {
+                    let tcid = m.get("tool_call_id")
+                        .and_then(|t| t.as_str())
+                        .unwrap_or("??");
+                    format!("[{}] tool(id={})", i, tcid)
+                }
+                _ => format!("[{}] {}", i, role),
+            }
+        }).collect())
+        .unwrap_or_default()
+}
+
 #[derive(Deserialize)]
 struct OpenAIResponse {
     id: String,
@@ -1063,6 +1064,7 @@ impl LLMProvider for OpenAIProvider {
 
         // Debug: Log LLM response (only in debug builds)
         if !status.is_success() {
+            let sequence = format_message_sequence(&body);
             tracing::error!(
                 provider = %self.name,
                 model = %self.model_id,
@@ -1070,6 +1072,7 @@ impl LLMProvider for OpenAIProvider {
                 status = %status,
                 response_len = text.len(),
                 response_body = %text,
+                sequence = ?sequence,
                 "OpenAI-compatible API request failed"
             );
             return Err(format!("API error {}: {}", status, text).into());