From 141ffda1ee8b3191c1148a5c2ddc8747a65c1d6e Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Wed, 15 Jul 2026 17:47:52 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E8=81=8A=E5=A4=A9?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E7=9A=84=E4=B8=B2=E8=A1=8C=E9=94=81=EF=BC=8C?= =?UTF-8?q?=E7=A1=AE=E4=BF=9D=E5=90=8C=E4=B8=80=E8=81=8A=E5=A4=A9=E7=9A=84?= =?UTF-8?q?=E6=B6=88=E6=81=AF=E5=A4=84=E7=90=86=E4=B8=B2=E8=A1=8C=E6=89=A7?= =?UTF-8?q?=E8=A1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/gateway/execution.rs | 78 ++++++++++++++++++++++++++++++ src/gateway/session.rs | 6 +++ src/gateway/session_history.rs | 14 ++++++ src/providers/openai.rs | 87 ++++++++++++++++++---------------- 4 files changed, 143 insertions(+), 42 deletions(-) diff --git a/src/gateway/execution.rs b/src/gateway/execution.rs index 73bf432..5713018 100644 --- a/src/gateway/execution.rs +++ b/src/gateway/execution.rs @@ -200,6 +200,17 @@ impl AgentExecutionService { &self, request: MessageExecutionRequest<'_>, ) -> Result, 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, 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(), "错误返回后锁应已释放"); + } } diff --git a/src/gateway/session.rs b/src/gateway/session.rs index 73aad11..fee3ed3 100644 --- a/src/gateway/session.rs +++ b/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> { + 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) { diff --git a/src/gateway/session_history.rs b/src/gateway/session_history.rs index c1824c8..685e698 100644 --- a/src/gateway/session_history.rs +++ b/src/gateway/session_history.rs @@ -21,6 +21,10 @@ pub(crate) struct SessionHistory { chat_topic_ids: HashMap, // 每个 chat 的当前 topic history_topic_ids: HashMap, // 每个 chat 的历史所对应的话题 compression_in_flight: HashSet, + /// 按 chat_id 的串行化锁。 + /// 同一 chat 的消息处理（agent loop + 压缩）必须串行执行， + /// 防止并发 loop 操作同一历史的不同快照产生交错序列。 + chat_serial_locks: HashMap>>, conversations: Arc, skill_events: Arc, } @@ -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> { + 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) } diff --git a/src/providers/openai.rs b/src/providers/openai.rs index aac4282..a066d8b 100644 --- a/src/providers/openai.rs +++ b/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 = 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 { + 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());