From 869e0c47e694ccc17fc5722cb7cf48ade0fed6df Mon Sep 17 00:00:00 2001 From: matevip Date: Thu, 23 Apr 2026 08:09:48 +0800 Subject: [PATCH] refactor(memory): unify ConversationCompletedEvent publish --- .../mate/channel/ChannelMessageRouter.java | 21 +++--- .../vip/mate/channel/web/ChatController.java | 22 ++---- .../channel/web/TalkModeWebSocketHandler.java | 6 ++ .../vip/mate/cron/service/CronJobService.java | 13 +--- .../ConversationCompletionPublisher.java | 71 +++++++++++++++++++ 5 files changed, 92 insertions(+), 41 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/memory/event/ConversationCompletionPublisher.java diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java index d334833a..b1834df1 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java @@ -10,8 +10,7 @@ import vip.mate.channel.model.ChannelEntity; import vip.mate.channel.notification.ApprovalNotificationService; import vip.mate.channel.service.ChannelService; import vip.mate.channel.web.ChatStreamTracker; -import org.springframework.context.ApplicationEventPublisher; -import vip.mate.memory.event.ConversationCompletedEvent; +import vip.mate.memory.event.ConversationCompletionPublisher; import vip.mate.tts.TtsService; import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.model.MessageContentPart; @@ -50,7 +49,7 @@ public class ChannelMessageRouter { private final ChannelSessionStore channelSessionStore; private final ApprovalService approvalService; private final ApprovalNotificationService approvalNotificationService; - private final ApplicationEventPublisher eventPublisher; + private final ConversationCompletionPublisher completionPublisher; private final TtsService ttsService; private final ObjectMapper objectMapper; private final ChatStreamTracker streamTracker; @@ -95,7 +94,7 @@ public class ChannelMessageRouter { ChannelSessionStore channelSessionStore, ApprovalService approvalService, ApprovalNotificationService approvalNotificationService, - ApplicationEventPublisher eventPublisher, + ConversationCompletionPublisher completionPublisher, TtsService ttsService, ObjectMapper objectMapper, ChatStreamTracker streamTracker) { @@ -105,7 +104,7 @@ public class ChannelMessageRouter { this.channelSessionStore = channelSessionStore; this.approvalService = approvalService; this.approvalNotificationService = approvalNotificationService; - this.eventPublisher = eventPublisher; + this.completionPublisher = completionPublisher; this.ttsService = ttsService; this.objectMapper = objectMapper; this.streamTracker = streamTracker; @@ -612,17 +611,13 @@ public class ChannelMessageRouter { } /** - * 发布对话完成事件（触发异步记忆提取），失败不影响正常流程 + * Publish the conversation-completed event (triggers async memory extraction). + * Delegates to {@link ConversationCompletionPublisher} so the try/catch and + * messageCount lookup no longer live here. */ private void publishConversationCompletedEvent(Long agentId, String conversationId, String userMessage, String assistantReply) { - try { - int msgCount = conversationService.getMessageCount(conversationId); - eventPublisher.publishEvent(new ConversationCompletedEvent( - agentId, conversationId, userMessage, assistantReply, msgCount, "channel")); - } catch (Exception e) { - log.debug("[Memory] Failed to publish ConversationCompletedEvent: {}", e.getMessage()); - } + completionPublisher.publish(agentId, conversationId, userMessage, assistantReply, "channel"); } // ==================== 流式处理（Web 渠道专用，不走队列） ==================== diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java index aa06f9bb..682d2e95 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java @@ -14,13 +14,12 @@ import org.springframework.security.core.Authentication; import org.springframework.web.bind.annotation.*; import org.springframework.web.multipart.MultipartFile; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; -import org.springframework.context.ApplicationEventPublisher; import vip.mate.common.result.R; import vip.mate.agent.AgentService; import vip.mate.agent.model.AgentEntity; import vip.mate.approval.ApprovalService; import vip.mate.approval.PendingApproval; -import vip.mate.memory.event.ConversationCompletedEvent; +import vip.mate.memory.event.ConversationCompletionPublisher; import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.model.MessageContentPart; import vip.mate.workspace.conversation.model.MessageEntity; @@ -59,7 +58,7 @@ public class ChatController { private final ApprovalService approvalService; private final ChatStreamTracker streamTracker; private final ObjectMapper objectMapper; - private final ApplicationEventPublisher eventPublisher; + private final ConversationCompletionPublisher completionPublisher; private final Path uploadRoot = Paths.get("data", "chat-uploads"); // 使用虚拟线程池处理 SSE（Java 17+ 兼容，Java 21 可用 Executors.newVirtualThreadPerTaskExecutor()） @@ -464,13 +463,7 @@ public class ChatController { } // 发布对话完成事件（仅正常完成时，停止/中断不触发记忆提取） if (!wasStopped) { - try { - int msgCount = conversationService.getMessageCount(conversationId); - eventPublisher.publishEvent(new ConversationCompletedEvent( - agentId, conversationId, message, assistantText, msgCount, "web")); - } catch (Exception ex) { - log.debug("[Memory] Failed to publish ConversationCompletedEvent: {}", ex.getMessage()); - } + completionPublisher.publish(agentId, conversationId, message, assistantText, "web"); } if (isInterruptFollowup) { @@ -806,14 +799,7 @@ public class ChatController { String promptText = buildPromptText(request.getMessage(), request.getContentParts()); String response = agentService.chat(agentId, promptText, request.getConversationId()); conversationService.saveMessage(request.getConversationId(), "assistant", response); - // 发布对话完成事件 - try { - int msgCount = conversationService.getMessageCount(request.getConversationId()); - eventPublisher.publishEvent(new ConversationCompletedEvent( - agentId, request.getConversationId(), request.getMessage(), response, msgCount, "web")); - } catch (Exception ex) { - log.debug("[Memory] Failed to publish ConversationCompletedEvent: {}", ex.getMessage()); - } + completionPublisher.publish(agentId, request.getConversationId(), request.getMessage(), response, "web"); return R.ok(response); } diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/TalkModeWebSocketHandler.java b/mateclaw-server/src/main/java/vip/mate/channel/web/TalkModeWebSocketHandler.java index 1d529639..6f8cea55 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/web/TalkModeWebSocketHandler.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/web/TalkModeWebSocketHandler.java @@ -10,6 +10,7 @@ import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.AbstractWebSocketHandler; import vip.mate.agent.AgentService; +import vip.mate.memory.event.ConversationCompletionPublisher; import vip.mate.stt.SttService; import vip.mate.tts.TtsService; import vip.mate.workspace.conversation.ConversationService; @@ -48,6 +49,7 @@ public class TalkModeWebSocketHandler extends AbstractWebSocketHandler { private final TtsService ttsService; private final AgentService agentService; private final ConversationService conversationService; + private final ConversationCompletionPublisher completionPublisher; private final ObjectMapper objectMapper; private final ExecutorService executor = Executors.newCachedThreadPool(); @@ -148,6 +150,10 @@ public class TalkModeWebSocketHandler extends AbstractWebSocketHandler { // 6. 保存助手回复 conversationService.saveMessage(talkSession.conversationId, "assistant", reply, List.of()); + // Publish conversation-completed event so memory extraction runs for voice turns too. + completionPublisher.publish(talkSession.agentId, talkSession.conversationId, + transcript, reply, "talk"); + // 7. 推送文字回复 sendJson(session, Map.of("type", "reply", "text", reply)); diff --git a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java index f30274fc..2196ba43 100644 --- a/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java +++ b/mateclaw-server/src/main/java/vip/mate/cron/service/CronJobService.java @@ -18,9 +18,8 @@ import vip.mate.agent.repository.AgentMapper; import vip.mate.cron.model.CronJobDTO; import vip.mate.cron.model.CronJobEntity; import vip.mate.cron.repository.CronJobMapper; -import org.springframework.context.ApplicationEventPublisher; import vip.mate.exception.MateClawException; -import vip.mate.memory.event.ConversationCompletedEvent; +import vip.mate.memory.event.ConversationCompletionPublisher; import vip.mate.workspace.conversation.ConversationService; import java.time.LocalDateTime; @@ -48,7 +47,7 @@ public class CronJobService implements ApplicationRunner { private final AgentMapper agentMapper; private final AgentService agentService; private final ConversationService conversationService; - private final ApplicationEventPublisher eventPublisher; + private final ConversationCompletionPublisher completionPublisher; private final ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); private final ConcurrentHashMap> scheduledTasks = new ConcurrentHashMap<>(); @@ -279,13 +278,7 @@ public class CronJobService implements ApplicationRunner { conversationService.saveMessage(conversationId, "assistant", result); // 发布对话完成事件 - try { - int msgCount = conversationService.getMessageCount(conversationId); - eventPublisher.publishEvent(new ConversationCompletedEvent( - job.getAgentId(), conversationId, userMessage, result, msgCount, "cron")); - } catch (Exception ex) { - log.debug("[Memory] Failed to publish ConversationCompletedEvent: {}", ex.getMessage()); - } + completionPublisher.publish(job.getAgentId(), conversationId, userMessage, result, "cron"); // 合并更新 lastRunTime + nextRunTime，单次 DB 写入 updateRunTimes(job.getId(), job.getCronExpression(), job.getTimezone()); diff --git a/mateclaw-server/src/main/java/vip/mate/memory/event/ConversationCompletionPublisher.java b/mateclaw-server/src/main/java/vip/mate/memory/event/ConversationCompletionPublisher.java new file mode 100644 index 00000000..f3258cbd --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/memory/event/ConversationCompletionPublisher.java @@ -0,0 +1,71 @@ +package vip.mate.memory.event; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.stereotype.Component; +import vip.mate.workspace.conversation.ConversationService; + +/** + * Single source of truth for publishing {@link ConversationCompletedEvent}. + *

+ * Callers (chat controllers, channel routers, cron executor, talk-mode handler …) + * should invoke {@link #publish(Long, String, String, String, String)} once per + * fully completed turn — i.e. after both the user and assistant messages have + * been persisted via {@link ConversationService}. + *

+ * This helper hides the boilerplate of looking up message count, building the + * event record, and swallowing publish-time exceptions so that a memory hook + * failure never surfaces to the user. + *

+ * Introduced to close the P0 gap in RFC-040 §6.1: three historical call sites + * hand-rolled the same try/catch, and several real entry points forgot to + * publish the event at all. Routing every site through this bean makes the + * memory production pipeline observable and uniform. + * + * @author MateClaw Team + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class ConversationCompletionPublisher { + + private final ApplicationEventPublisher eventPublisher; + private final ConversationService conversationService; + + /** + * Publish a {@link ConversationCompletedEvent} for the given turn. + * Must be called after the assistant message is persisted, because + * {@link PostConversationMemoryListener} uses {@code messageCount} as a + * summarization threshold and downstream summarizer reads the DB. + * + * @param agentId agent id the turn ran under + * @param conversationId conversation/session id + * @param userMessage last user message (for summarizer heuristics) + * @param assistantReply agent's final reply (may be empty for replay-only turns) + * @param source trigger source label: {@code web}, {@code channel}, + * {@code cron}, {@code talk}, {@code replay}, … + */ + public void publish(Long agentId, + String conversationId, + String userMessage, + String assistantReply, + String source) { + if (agentId == null || conversationId == null || conversationId.isBlank()) { + return; + } + try { + int messageCount = conversationService.getMessageCount(conversationId); + eventPublisher.publishEvent(new ConversationCompletedEvent( + agentId, + conversationId, + userMessage != null ? userMessage : "", + assistantReply != null ? assistantReply : "", + messageCount, + source != null ? source : "unknown")); + } catch (Exception e) { + log.debug("[Memory] Failed to publish ConversationCompletedEvent (source={}, conv={}): {}", + source, conversationId, e.getMessage()); + } + } +}