refactor(memory): unify ConversationCompletedEvent publish

2026-09-13 03:13:41 +08:00 · 2026-04-23 08:09:48 +08:00 · 2026-04-23 08:09:48 +08:00 · 869e0c47e6
commit 869e0c47e6
parent 2e15369465
5 changed files with 92 additions and 41 deletions
--- 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 渠道专用，不走队列） ====================
--- 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);
    }

--- 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));

--- 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<Long, ScheduledFuture<?>> 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());
--- a/mateclaw-server/src/main/java/vip/mate/memory/event/ConversationCompletionPublisher.java
+++ 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}.
+ * <p>
+ * 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}.
+ * <p>
+ * 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.
+ * <p>
+ * 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 <em>after</em> 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());
+        }
+    }
+}