From d7ff969560d2a429b3f3bc2666251003eb36ec04 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Mon, 6 Jul 2026 09:04:32 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=BC=BA=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E5=A4=84=E7=90=86=EF=BC=8C=E6=B8=85=E7=90=86=E4=B8=8D=E5=AE=8C?= =?UTF-8?q?=E6=95=B4=E7=9A=84=20tool=5Fcall=20=E5=BA=8F=E5=88=97=E4=BB=A5?= =?UTF-8?q?=E9=98=B2=E6=AD=A2=E6=95=B0=E6=8D=AE=E6=B1=A1=E6=9F=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/agent/agent_loop.rs | 45 +++++++++++++++++++++++++++++++++- src/gateway/session_history.rs | 26 +++++++++++++++++++- 2 files changed, 69 insertions(+), 2 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 86f97c6..7bd961b 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -899,7 +899,32 @@ impl AgentLoop { // Sanitize: remove any trailing incomplete tool call sequences // that may have been persisted before a process interruption. - crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut messages); + { + let tool_call_ids: Vec<_> = messages.iter() + .filter(|m| m.role == "assistant") + .filter_map(|m| m.tool_calls.as_ref()) + .flatten() + .map(|tc| tc.id.clone()) + .collect(); + let tool_result_ids: Vec<_> = messages.iter() + .filter(|m| m.role == "tool") + .filter_map(|m| m.tool_call_id.clone()) + .collect(); + tracing::debug!( + total_messages = messages.len(), + tool_call_ids = ?tool_call_ids, + tool_result_ids = ?tool_result_ids, + "Pre-process message state before sanitize" + ); + } + let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut messages); + if removed > 0 { + tracing::warn!( + removed_count = removed, + after_len = messages.len(), + "Pre-process sanitize removed incomplete tool_call sequences" + ); + } // Track tool calls for loop detection let mut loop_detector = LoopDetector::new(LoopDetectorConfig::default()); @@ -1152,6 +1177,18 @@ impl AgentLoop { let tool_results = if self.cancel_token.is_some() { tokio::select! { _ = self.cancel_signal() => { + // 为所有 tool_calls 补充取消结果，避免孤立 assistant(tool_calls) + for tool_call in &response.tool_calls { + let cancel_result = ChatMessage::tool_with_state( + tool_call.id.clone(), + tool_call.name.clone(), + "[Tool execution cancelled by user]".to_string(), + ToolMessageState::Completed, + ); + messages.push(cancel_result.clone()); + emitted_messages.push(cancel_result.clone()); + self.emit_tool_result(cancel_result, Some(0)).await; + } let cancel = Self::build_cancel_result(iteration, emitted_messages); self.emit_live_tool_call_message(cancel.final_response.clone()).await; return Ok(cancel); @@ -1230,6 +1267,12 @@ impl AgentLoop { // Max iterations reached - ask LLM for a summary based on completed work tracing::warn!("Max iterations reached, requesting final summary from LLM"); + // Defense: sanitize before final summary request + let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut messages); + if removed > 0 { + tracing::warn!(removed_count = removed, "Sanitized before max-iterations summary"); + } + // Add a message asking for summary let summary_request = ChatMessage::user( "You have reached the maximum number of tool call iterations. \ diff --git a/src/gateway/session_history.rs b/src/gateway/session_history.rs index bebda25..4c5b1ef 100644 --- a/src/gateway/session_history.rs +++ b/src/gateway/session_history.rs @@ -65,7 +65,7 @@ impl SessionHistory { } // 如果提供了 topic_id，按 topic 加载；否则按 session 加载 - let history = if let Some(tid) = topic_id { + let mut history = if let Some(tid) = topic_id { let sid = self.persistent_session_id(chat_id); self.conversations .load_messages_for_topic(tid, Some(&sid)) @@ -75,6 +75,17 @@ impl SessionHistory { .load_messages(&self.persistent_session_id(chat_id)) .map_err(|err| AgentError::Other(format!("session history load error: {}", err)))? }; + + // 清理 DB 加载的历史中可能存在的不完整 tool_call 序列 + let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut history); + if removed > 0 { + tracing::warn!( + chat_id = %chat_id, + removed_count = removed, + "Sanitized incomplete tool_call sequences on history load" + ); + } + self.chat_histories.insert(chat_id.to_string(), history); Ok(()) } @@ -159,6 +170,19 @@ impl SessionHistory { return Ok(()); } + // 在追加新消息前，先清理内存历史中的不完整 tool_call 序列 + // 这防止脏数据（如取消时产生的孤立 assistant(tool_calls)）在内存历史中累积 + if let Some(history) = self.chat_histories.get_mut(chat_id) { + let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(history); + if removed > 0 { + tracing::warn!( + chat_id = %chat_id, + removed_count = removed, + "Sanitized in-memory history before appending persisted messages" + ); + } + } + for message in messages { self.add_message(chat_id, message); }