fix: 修复消息持久化缺失 topic 关联和 assistant 文本丢失

- PersistingEmittedMessageHandler 新增 topic_id 参数,使用 append_message_with_topic 替代 append_message
- agent_loop 的所有退出路径中为最终 assistant 文本添加 emit_live_tool_call_message
- 更新 finalize_result filter,live_emitter 存在时抑制所有消息的 post-loop 广播

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
oudecheng 2026-05-29 18:09:00 +08:00
parent 2bda66a042
commit 06756a4816
4 changed files with 13 additions and 7 deletions

View File

@ -663,6 +663,7 @@ pub struct PersistingEmittedMessageHandler<H: EmittedMessageHandler> {
inner: H, inner: H,
conversation_repository: Arc<dyn ConversationRepository>, conversation_repository: Arc<dyn ConversationRepository>,
session_id: String, session_id: String,
topic_id: Option<String>,
} }
impl<H: EmittedMessageHandler> PersistingEmittedMessageHandler<H> { impl<H: EmittedMessageHandler> PersistingEmittedMessageHandler<H> {
@ -670,8 +671,9 @@ impl<H: EmittedMessageHandler> PersistingEmittedMessageHandler<H> {
inner: H, inner: H,
conversation_repository: Arc<dyn ConversationRepository>, conversation_repository: Arc<dyn ConversationRepository>,
session_id: impl Into<String>, session_id: impl Into<String>,
topic_id: Option<String>,
) -> Self { ) -> Self {
Self { inner, conversation_repository, session_id: session_id.into() } Self { inner, conversation_repository, session_id: session_id.into(), topic_id }
} }
} }
@ -679,7 +681,7 @@ impl<H: EmittedMessageHandler> PersistingEmittedMessageHandler<H> {
impl<H: EmittedMessageHandler> EmittedMessageHandler for PersistingEmittedMessageHandler<H> { impl<H: EmittedMessageHandler> EmittedMessageHandler for PersistingEmittedMessageHandler<H> {
async fn handle(&self, message: ChatMessage) { async fn handle(&self, message: ChatMessage) {
if let Err(e) = self.conversation_repository if let Err(e) = self.conversation_repository
.append_message(&self.session_id, &message) .append_message_with_topic(&self.session_id, self.topic_id.as_deref(), &message)
{ {
tracing::error!(error = %e, session_id = %self.session_id, tracing::error!(error = %e, session_id = %self.session_id,
"Failed to persist emitted message"); "Failed to persist emitted message");
@ -915,6 +917,7 @@ impl AgentLoop {
let assistant_message = let assistant_message =
ChatMessage::assistant(recoverable_llm_message(&e.to_string())); ChatMessage::assistant(recoverable_llm_message(&e.to_string()));
emitted_messages.push(assistant_message.clone()); emitted_messages.push(assistant_message.clone());
self.emit_live_tool_call_message(assistant_message.clone()).await;
return Ok(AgentProcessResult { return Ok(AgentProcessResult {
final_response: assistant_message, final_response: assistant_message,
emitted_messages, emitted_messages,
@ -939,6 +942,7 @@ impl AgentLoop {
ChatMessage::assistant(response.content) ChatMessage::assistant(response.content)
}; };
emitted_messages.push(assistant_message.clone()); emitted_messages.push(assistant_message.clone());
self.emit_live_tool_call_message(assistant_message.clone()).await;
return Ok(AgentProcessResult { return Ok(AgentProcessResult {
final_response: assistant_message, final_response: assistant_message,
emitted_messages, emitted_messages,
@ -1044,6 +1048,7 @@ impl AgentLoop {
tool_call.name, tool_call.name,
)); ));
emitted_messages.push(assistant_message.clone()); emitted_messages.push(assistant_message.clone());
self.emit_live_tool_call_message(assistant_message.clone()).await;
return Ok(AgentProcessResult { return Ok(AgentProcessResult {
final_response: assistant_message, final_response: assistant_message,
emitted_messages, emitted_messages,
@ -1120,6 +1125,7 @@ impl AgentLoop {
ChatMessage::assistant(response.content) ChatMessage::assistant(response.content)
}; };
emitted_messages.push(assistant_message.clone()); emitted_messages.push(assistant_message.clone());
self.emit_live_tool_call_message(assistant_message.clone()).await;
Ok(AgentProcessResult { Ok(AgentProcessResult {
final_response: assistant_message, final_response: assistant_message,
emitted_messages, emitted_messages,
@ -1135,6 +1141,7 @@ impl AgentLoop {
); );
let final_message = ChatMessage::assistant(recoverable_llm_message(&e.to_string())); let final_message = ChatMessage::assistant(recoverable_llm_message(&e.to_string()));
emitted_messages.push(final_message.clone()); emitted_messages.push(final_message.clone());
self.emit_live_tool_call_message(final_message.clone()).await;
Ok(AgentProcessResult { Ok(AgentProcessResult {
final_response: final_message, final_response: final_message,
emitted_messages, emitted_messages,

View File

@ -157,11 +157,8 @@ impl AgentExecutionService {
.emitted_messages .emitted_messages
.iter() .iter()
.filter(|message| { .filter(|message| {
// 当存在 live_emitter 时,工具调用和工具结果已在 loop 中实时广播 // 当存在 live_emitter 时,所有消息已在 loop 中实时广播,不需要 post-loop 发送
// 只保留最终 assistant 文本通过 post-loop 路径发送 !request.suppress_live_tool_calls
let already_emitted = request.suppress_live_tool_calls
&& (message.is_assistant_tool_call_message() || message.role == "tool");
!already_emitted
&& should_display_message_to_user(self.show_tool_results, message) && should_display_message_to_user(self.show_tool_results, message)
}) })
.flat_map(|message| { .flat_map(|message| {

View File

@ -227,6 +227,7 @@ impl InboundProcessor {
), ),
self.session_manager.store(), self.session_manager.store(),
&session_id, &session_id,
current_topic.clone(),
)); ));
match self match self

View File

@ -236,6 +236,7 @@ impl DefaultSubAgentRuntime {
}, },
self.conversation_repository.clone(), self.conversation_repository.clone(),
session.session_id.clone(), session.session_id.clone(),
session.parent_topic_id.clone(),
)); ));
return agent.with_emitted_message_handler(emitter); return agent.with_emitted_message_handler(emitter);