From f28ad8d9d48aebf5785cdce17adb6665e8cc612d Mon Sep 17 00:00:00 2001 From: matevip Date: Sat, 11 Apr 2026 00:16:33 +0800 Subject: [PATCH] fix(chat): preserve refresh history and thinking order --- .../vip/mate/channel/web/ChatController.java | 227 +++++++++++++----- .../src/components/chat/MessageBubble.vue | 23 +- 2 files changed, 181 insertions(+), 69 deletions(-) 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 05df5d95..bcf86d5b 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 @@ -23,6 +23,7 @@ import vip.mate.approval.PendingApproval; import vip.mate.memory.event.ConversationCompletedEvent; import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.model.MessageContentPart; +import vip.mate.workspace.conversation.model.MessageEntity; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; @@ -211,11 +212,13 @@ public class ChatController { if ("denied".equals(decision)) { String denyMsg = "用户拒绝执行工具 " + pending.getToolName(); - conversationService.saveMessage(conversationId, "assistant", denyMsg); + MessageEntity savedAssistant = conversationService.saveMessage(conversationId, "assistant", denyMsg); broadcastEvent(conversationId, "message_start", Map.of("role", "assistant")); broadcastEvent(conversationId, "content_delta", Map.of("delta", denyMsg)); broadcastEvent(conversationId, "message_complete", Map.of("status", "completed")); - broadcastEvent(conversationId, "done", Map.of("status", "completed")); + broadcastEvent(conversationId, "done", buildDonePayload( + conversationId, "completed", savedAssistant, 0, 0, true, + conversationService.getMessageCount(conversationId))); // deny 是正常 turn 终结，用户可能在 awaiting_approval 阶段排了消息 ChatStreamTracker.CompletionResult denyCr = streamTracker.completeAndConsumeIfLast(conversationId); if (denyCr.allDone() && denyCr.queuedInput() != null) { @@ -262,10 +265,11 @@ public class ChatController { .doOnComplete(() -> { if (!finalized.compareAndSet(false, true)) return; try { + MessageEntity savedAssistant = null; List parts = accumulator.toAssistantParts(); String text = accumulator.getContent(); if (!text.isBlank() || !parts.isEmpty()) { - conversationService.saveMessage(conversationId, "assistant", text, parts, + savedAssistant = conversationService.saveMessage(conversationId, "assistant", text, parts, "completed", accumulator.getPromptTokens(), accumulator.getCompletionTokens(), @@ -279,12 +283,8 @@ public class ChatController { "hasContent", !text.isBlank() )); int msgCount = conversationService.getMessageCount(conversationId); - broadcastEvent(conversationId, "done", Map.of( - "conversationId", conversationId, - "status", "completed", - "persisted", true, - "messageCount", msgCount - )); + broadcastEvent(conversationId, "done", buildDonePayload( + conversationId, "completed", savedAssistant, 0, 0, true, msgCount)); } catch (Exception e) { log.warn("SSE replay complete error: {}", e.getMessage()); } finally { @@ -318,12 +318,13 @@ public class ChatController { } try { + MessageEntity savedAssistant = null; List replayParts = accumulator.toAssistantParts(); String replayText = accumulator.getContent(); if (!replayText.isBlank() || !replayParts.isEmpty()) { String savedText = replayText.isBlank() && isUserStop ? (replayIsFollowup ? "[已中断]" : "[已停止生成]") : replayText; - conversationService.saveMessage(conversationId, "assistant", savedText, replayParts, + savedAssistant = conversationService.saveMessage(conversationId, "assistant", savedText, replayParts, errStatus, accumulator.getPromptTokens(), accumulator.getCompletionTokens(), @@ -331,7 +332,7 @@ public class ChatController { accumulator.getRuntimeProviderId(), accumulator.toMetadataJson()); } else if (isUserStop) { - conversationService.saveMessage(conversationId, "assistant", + savedAssistant = conversationService.saveMessage(conversationId, "assistant", replayIsFollowup ? "[已中断]" : "[已停止生成]", null, errStatus); } @@ -352,15 +353,13 @@ public class ChatController { "hasContent", !replayText.isBlank() )); int stoppedMsgCount = conversationService.getMessageCount(conversationId); - broadcastEvent(conversationId, "done", Map.of( - "conversationId", conversationId, - "status", "stopped", - "persisted", true, - "messageCount", stoppedMsgCount - )); + broadcastEvent(conversationId, "done", buildDonePayload( + conversationId, "stopped", savedAssistant, 0, 0, true, stoppedMsgCount)); } else { - broadcastEvent(conversationId, "error", Map.of("message", - e.getMessage() != null ? e.getMessage() : "replay error")); + broadcastEvent(conversationId, "error", buildErrorPayload( + conversationId, + e.getMessage() != null ? e.getMessage() : "replay error", + savedAssistant)); } } catch (Exception ex) { log.warn("SSE replay error finalize failed: {}", ex.getMessage()); @@ -446,12 +445,13 @@ public class ChatController { persistStatus = isInterruptFollowup ? "interrupted" : "stopped"; } try { + MessageEntity savedAssistant = null; List assistantParts = accumulator.toAssistantParts(); String assistantText = accumulator.getContent(); if (!assistantText.isBlank() || !assistantParts.isEmpty()) { String savedText = assistantText.isBlank() && wasStopped ? (isInterruptFollowup ? "[已中断]" : "[已停止生成]") : assistantText; - conversationService.saveMessage(conversationId, "assistant", savedText, assistantParts, + savedAssistant = conversationService.saveMessage(conversationId, "assistant", savedText, assistantParts, persistStatus, accumulator.getPromptTokens(), accumulator.getCompletionTokens(), @@ -459,7 +459,7 @@ public class ChatController { accumulator.getRuntimeProviderId(), accumulator.toMetadataJson()); } else if (wasStopped) { - conversationService.saveMessage(conversationId, "assistant", + savedAssistant = conversationService.saveMessage(conversationId, "assistant", isInterruptFollowup ? "[已中断]" : "[已停止生成]", null, persistStatus); } // 发布对话完成事件（仅正常完成时，停止/中断不触发记忆提取） @@ -490,14 +490,9 @@ public class ChatController { "hasContent", !assistantText.isBlank() )); int msgCount = conversationService.getMessageCount(conversationId); - broadcastEvent(conversationId, "done", Map.of( - "conversationId", conversationId, - "status", persistStatus, - "promptTokens", accumulator.getPromptTokens(), - "completionTokens", accumulator.getCompletionTokens(), - "persisted", true, - "messageCount", msgCount - )); + broadcastEvent(conversationId, "done", buildDonePayload( + conversationId, persistStatus, savedAssistant, + accumulator.getPromptTokens(), accumulator.getCompletionTokens(), true, msgCount)); } } catch (Exception e) { log.warn("SSE complete error: {}", e.getMessage()); @@ -534,12 +529,13 @@ public class ChatController { log.info("SSE stream cancelled ({}): conversationId={}", status, conversationId); try { + MessageEntity savedAssistant = null; List assistantParts = accumulator.toAssistantParts(); String assistantText = accumulator.getContent(); if (!assistantText.isBlank() || !assistantParts.isEmpty()) { String savedText = assistantText.isBlank() ? (isInterruptFollowup ? "[已中断]" : "[已停止生成]") : assistantText; - conversationService.saveMessage(conversationId, "assistant", savedText, assistantParts, + savedAssistant = conversationService.saveMessage(conversationId, "assistant", savedText, assistantParts, status, accumulator.getPromptTokens(), accumulator.getCompletionTokens(), @@ -547,7 +543,7 @@ public class ChatController { accumulator.getRuntimeProviderId(), accumulator.toMetadataJson()); } else { - conversationService.saveMessage(conversationId, "assistant", + savedAssistant = conversationService.saveMessage(conversationId, "assistant", isInterruptFollowup ? "[已中断]" : "[已停止生成]", null, status); } @@ -568,12 +564,8 @@ public class ChatController { "hasContent", !assistantText.isBlank() )); int stoppedMsgCount = conversationService.getMessageCount(conversationId); - broadcastEvent(conversationId, "done", Map.of( - "conversationId", conversationId, - "status", "stopped", - "persisted", true, - "messageCount", stoppedMsgCount - )); + broadcastEvent(conversationId, "done", buildDonePayload( + conversationId, "stopped", savedAssistant, 0, 0, true, stoppedMsgCount)); } } catch (Exception e) { log.warn("SSE stop finalize error: {}", e.getMessage()); @@ -623,10 +615,11 @@ public class ChatController { log.info("SSE doOnError saving: conversationId={}, status={}, textLen={}, partsCount={}", conversationId, status, assistantText.length(), assistantParts.size()); String errorMsg = e.getMessage() != null ? e.getMessage() : "unknown error"; + MessageEntity savedAssistant = null; if (!assistantText.isBlank() || !assistantParts.isEmpty()) { String savedText = assistantText.isBlank() && isUserStop ? (isInterruptFollowup ? "[已中断]" : "[已停止生成]") : assistantText; - conversationService.saveMessage(conversationId, "assistant", savedText, assistantParts, + savedAssistant = conversationService.saveMessage(conversationId, "assistant", savedText, assistantParts, status, accumulator.getPromptTokens(), accumulator.getCompletionTokens(), @@ -634,10 +627,10 @@ public class ChatController { accumulator.getRuntimeProviderId(), accumulator.toMetadataJson()); } else if (isUserStop) { - conversationService.saveMessage(conversationId, "assistant", + savedAssistant = conversationService.saveMessage(conversationId, "assistant", isInterruptFollowup ? "[已中断]" : "[已停止生成]", null, status); } else { - conversationService.saveMessage(conversationId, "assistant", "[错误] " + errorMsg, null, "failed"); + savedAssistant = conversationService.saveMessage(conversationId, "assistant", "[错误] " + errorMsg, null, "failed"); } if (isInterruptFollowup) { @@ -657,17 +650,10 @@ public class ChatController { "hasContent", !assistantText.isBlank() )); int stoppedMsgCount = conversationService.getMessageCount(conversationId); - broadcastEvent(conversationId, "done", Map.of( - "conversationId", conversationId, - "status", "stopped", - "persisted", true, - "messageCount", stoppedMsgCount - )); + broadcastEvent(conversationId, "done", buildDonePayload( + conversationId, "stopped", savedAssistant, 0, 0, true, stoppedMsgCount)); } else { - broadcastEvent(conversationId, "error", Map.of( - "message", errorMsg, - "conversationId", conversationId - )); + broadcastEvent(conversationId, "error", buildErrorPayload(conversationId, errorMsg, savedAssistant)); } } catch (Exception ioException) { log.error("SSE doOnError save/broadcast failed: conversationId={}, error={}", @@ -1011,10 +997,11 @@ public class ChatController { .doOnComplete(() -> { if (!finalized.compareAndSet(false, true)) return; try { + MessageEntity savedAssistant = null; List parts = accumulator.toAssistantParts(); String text = accumulator.getContent(); if (!text.isBlank() || !parts.isEmpty()) { - conversationService.saveMessage(conversationId, "assistant", text, parts, + savedAssistant = conversationService.saveMessage(conversationId, "assistant", text, parts, "completed", accumulator.getPromptTokens(), accumulator.getCompletionTokens(), @@ -1027,11 +1014,10 @@ public class ChatController { "hasThinking", !accumulator.getThinking().isBlank(), "hasContent", !text.isBlank() )); - broadcastEvent(conversationId, "done", Map.of( - "status", "completed", - "promptTokens", accumulator.getPromptTokens(), - "completionTokens", accumulator.getCompletionTokens() - )); + broadcastEvent(conversationId, "done", buildDonePayload( + conversationId, "completed", savedAssistant, + accumulator.getPromptTokens(), accumulator.getCompletionTokens(), true, + conversationService.getMessageCount(conversationId))); } catch (Exception e) { log.warn("SSE queued complete error: {}", e.getMessage()); } finally { @@ -1055,10 +1041,11 @@ public class ChatController { log.error("SSE queued stream error: conversationId={}, cause={}", conversationId, e.getMessage()); // 持久化已累积的 assistant 消息（修复：原逻辑未保存导致回答丢失） try { + MessageEntity savedAssistant = null; List parts = accumulator.toAssistantParts(); String text = accumulator.getContent(); if (!text.isBlank() || !parts.isEmpty()) { - conversationService.saveMessage(conversationId, "assistant", text, parts, + savedAssistant = conversationService.saveMessage(conversationId, "assistant", text, parts, "failed", accumulator.getPromptTokens(), accumulator.getCompletionTokens(), @@ -1067,14 +1054,16 @@ public class ChatController { accumulator.toMetadataJson()); } else { String errorMsg = e.getMessage() != null ? e.getMessage() : "queued stream error"; - conversationService.saveMessage(conversationId, "assistant", + savedAssistant = conversationService.saveMessage(conversationId, "assistant", "[错误] " + errorMsg, null, "failed"); } + broadcastEvent(conversationId, "error", buildErrorPayload( + conversationId, + e.getMessage() != null ? e.getMessage() : "queued stream error", + savedAssistant)); } catch (Exception saveEx) { log.error("SSE queued doOnError save failed: {}", saveEx.getMessage()); } - broadcastEvent(conversationId, "error", Map.of( - "message", e.getMessage() != null ? e.getMessage() : "queued stream error")); ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId); if (cr.allDone()) { if (cr.queuedInput() != null) { @@ -1109,6 +1098,32 @@ public class ChatController { streamTracker.broadcast(conversationId, name, payload); } + private Map buildDonePayload(String conversationId, String status, MessageEntity savedAssistant, + int promptTokens, int completionTokens, + boolean persisted, Integer messageCount) { + Map payload = new LinkedHashMap<>(); + if (conversationId != null && !conversationId.isBlank()) payload.put("conversationId", conversationId); + payload.put("status", status); + if (savedAssistant != null && savedAssistant.getId() != null) { + payload.put("assistantMessageId", savedAssistant.getId()); + } + if (promptTokens > 0) payload.put("promptTokens", promptTokens); + if (completionTokens > 0) payload.put("completionTokens", completionTokens); + payload.put("persisted", persisted); + if (messageCount != null) payload.put("messageCount", messageCount); + return payload; + } + + private Map buildErrorPayload(String conversationId, String message, MessageEntity savedAssistant) { + Map payload = new LinkedHashMap<>(); + payload.put("message", message); + if (conversationId != null && !conversationId.isBlank()) payload.put("conversationId", conversationId); + if (savedAssistant != null && savedAssistant.getId() != null) { + payload.put("assistantMessageId", savedAssistant.getId()); + } + return payload; + } + private List normalizeRequestParts(ChatStreamRequest request) { if (request.getContentParts() != null && !request.getContentParts().isEmpty()) { return request.getContentParts(); @@ -1232,12 +1247,20 @@ public class ChatController { private final List> toolCalls = new ArrayList<>(); /** 有序事件时间线 — 前端分段渲染的权威数据源 */ private final List> segments = new ArrayList<>(); + private final List> browserActions = new ArrayList<>(); + private final List warnings = new ArrayList<>(); + private final List> planStepResults = new ArrayList<>(); private int segCounter = 0; private int promptTokens = 0; private int completionTokens = 0; private String runtimeModelName = ""; private String runtimeProviderId = ""; private boolean awaitingApproval = false; + private String currentPhase = ""; + private Long planId = null; + private List planSteps = List.of(); + private Integer currentPlanStep = null; + private Map pendingApproval = null; synchronized void accept(AgentService.StreamDelta delta, String conversationId) { if (delta == null) return; @@ -1254,6 +1277,7 @@ public class ChatController { if ("phase".equals(delta.eventType())) { String phase = String.valueOf(delta.eventData().getOrDefault("phase", "")); if (!phase.isBlank()) { + currentPhase = phase; streamTracker.updatePhase(conversationId, phase); // phase 切换时关闭 running 的 content/thinking segment，保留边界 finalizeRunningSegments("content", "thinking"); @@ -1309,7 +1333,62 @@ public class ChatController { private void accumulateToolEvent(String eventType, Map data, String conversationId) { if ("tool_approval_requested".equals(eventType)) { awaitingApproval = true; + currentPhase = "awaiting_approval"; + pendingApproval = new LinkedHashMap<>(); + pendingApproval.put("pendingId", data.getOrDefault("pendingId", "")); + pendingApproval.put("toolName", data.getOrDefault("toolName", "")); + pendingApproval.put("arguments", data.getOrDefault("arguments", "")); + pendingApproval.put("reason", data.getOrDefault("reason", "")); + pendingApproval.put("status", "pending_approval"); + if (data.containsKey("findings")) pendingApproval.put("findings", data.get("findings")); + if (data.containsKey("maxSeverity")) pendingApproval.put("maxSeverity", data.get("maxSeverity")); + if (data.containsKey("summary")) pendingApproval.put("summary", data.get("summary")); streamTracker.updatePhase(conversationId, "awaiting_approval"); + } else if ("tool_approval_resolved".equals(eventType)) { + if (pendingApproval != null) { + pendingApproval.put("status", + "approved".equals(String.valueOf(data.getOrDefault("decision", ""))) ? "approved" : "denied"); + } + } else if ("plan_created".equals(eventType)) { + Object rawPlanId = data.get("planId"); + if (rawPlanId instanceof Number n) { + planId = n.longValue(); + } else if (rawPlanId != null) { + try { planId = Long.valueOf(String.valueOf(rawPlanId)); } catch (Exception ignored) {} + } + Object steps = data.get("steps"); + if (steps instanceof List list) { + planSteps = list.stream().map(String::valueOf).toList(); + planStepResults.clear(); + for (int i = 0; i < planSteps.size(); i++) { + planStepResults.add(null); + } + } + currentPlanStep = 0; + } else if ("plan_step_started".equals(eventType)) { + Object idx = data.get("index"); + if (idx instanceof Number n) { + currentPlanStep = n.intValue(); + } + } else if ("plan_step_completed".equals(eventType)) { + Object idx = data.get("index"); + if (idx instanceof Number n) { + int index = n.intValue(); + currentPlanStep = index; + ensurePlanStepCapacity(index + 1); + Map stepResult = new LinkedHashMap<>(); + stepResult.put("result", data.getOrDefault("result", "")); + stepResult.put("status", "completed"); + planStepResults.set(index, stepResult); + } + } else if ("browser_action".equals(eventType)) { + browserActions.add(new LinkedHashMap<>(data)); + } else if ("warning".equals(eventType)) { + String warning = String.valueOf(data.getOrDefault("message", + data.getOrDefault("delta", ""))); + if (!warning.isBlank()) { + warnings.add(warning); + } } else if ("tool_call_started".equals(eventType)) { // toolCalls（兼容） Map tc = new LinkedHashMap<>(); @@ -1349,6 +1428,12 @@ public class ChatController { } } + private void ensurePlanStepCapacity(int size) { + while (planStepResults.size() < size) { + planStepResults.add(null); + } + } + // ==================== Segment helpers ==================== private Map newSegment(String type) { @@ -1430,6 +1515,28 @@ public class ChatController { if (!segments.isEmpty()) { metadata.put("segments", segments); } + if (!currentPhase.isBlank()) { + metadata.put("currentPhase", currentPhase); + } + if (planId != null || !planSteps.isEmpty() || currentPlanStep != null) { + Map plan = new LinkedHashMap<>(); + if (planId != null) plan.put("planId", planId); + if (!planSteps.isEmpty()) plan.put("steps", planSteps); + if (currentPlanStep != null) plan.put("currentStep", currentPlanStep); + if (planStepResults.stream().anyMatch(java.util.Objects::nonNull)) { + plan.put("stepResults", planStepResults); + } + metadata.put("plan", plan); + } + if (pendingApproval != null && !pendingApproval.isEmpty()) { + metadata.put("pendingApproval", pendingApproval); + } + if (!browserActions.isEmpty()) { + metadata.put("browserActions", browserActions); + } + if (!warnings.isEmpty()) { + metadata.put("warnings", warnings); + } return objectMapper.writeValueAsString(metadata); } catch (Exception e) { log.warn("Failed to serialize metadata: {}", e.getMessage()); diff --git a/mateclaw-ui/src/components/chat/MessageBubble.vue b/mateclaw-ui/src/components/chat/MessageBubble.vue index 3cef443d..b7a6cf35 100644 --- a/mateclaw-ui/src/components/chat/MessageBubble.vue +++ b/mateclaw-ui/src/components/chat/MessageBubble.vue @@ -594,19 +594,12 @@ const segments = computed(() => { if (props.message.role !== 'assistant') return [] const meta = parsedMetadata.value - // DEBUG: 临时日志，验证后删除 - if (meta?.segments) { - console.debug('[MessageBubble] segments found:', meta.segments.length, 'first:', meta.segments[0]?.type) - } else { - console.debug('[MessageBubble] NO segments in metadata, keys:', Object.keys(meta || {})) - } - // 优先：使用 metadata.segments（流式时由前端写入，历史时由后端持久化） - // 这是按事件顺序记录的完整时间线，保留了 thinking→tools→content 的真实交错 if (meta?.segments && Array.isArray(meta.segments) && meta.segments.length > 0 && typeof meta.segments[0] === 'object' && meta.segments[0]?.type) { - // 补充：如果后端 segments 没有 thinking 但 contentParts 有（非原生 thinking 模型） const segs = [...meta.segments] as MessageSegment[] + + // 补充：如果后端 segments 没有 thinking 但 contentParts 有（非原生 thinking 模型） const hasThinking = segs.some(s => s.type === 'thinking') if (!hasThinking) { const thinkingPart = props.message.contentParts?.find(p => p.type === 'thinking') @@ -614,6 +607,18 @@ const segments = computed(() => { segs.unshift({ id: 'th-fb', type: 'thinking', status: 'completed', thinkingText: thinkingPart.text }) } } + + // 修复历史消息顺序：如果 thinking 被落在 content 后面，提到首个 content 前 + // 只处理单个 thinking 段的常见场景，避免破坏复杂交错时间线 + const thinkingIndices = segs + .map((seg, index) => seg.type === 'thinking' ? index : -1) + .filter(index => index >= 0) + const firstNonThinkingIdx = segs.findIndex((seg: MessageSegment) => seg.type !== 'thinking') + if (thinkingIndices.length === 1 && firstNonThinkingIdx >= 0 && thinkingIndices[0] > firstNonThinkingIdx) { + const [thinkingSeg] = segs.splice(thinkingIndices[0], 1) + segs.splice(0, 0, thinkingSeg) + } + return segs }