mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 11:13:43 +08:00
feat(conversation): rewind-to-message endpoint and duplicate-free regenerate across web, webchat, and console UI
This commit is contained in:
matevip
parent
a183519d69
commit
717f15e91f
@ -39,8 +39,6 @@ import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
@ -160,7 +158,7 @@ public class ChatController {
|
||||
|
||||
// ---- 分支 B：正常请求 ----
|
||||
Long agentId = request.getAgentId();
|
||||
String message = request.getMessage() != null ? request.getMessage() : "";
|
||||
String requestMessage = request.getMessage() != null ? request.getMessage() : "";
|
||||
if (auth == null) {
|
||||
try {
|
||||
sendEvent(emitter, "error", Map.of("message", "未登录，请先登录"));
|
||||
@ -194,7 +192,7 @@ public class ChatController {
|
||||
}
|
||||
|
||||
// ---- 审批命令拦截：/approve、/deny 走 SSE 流式 replay ----
|
||||
String normalizedMsg = message.trim().toLowerCase();
|
||||
String normalizedMsg = requestMessage.trim().toLowerCase();
|
||||
boolean isApprovalCommand = "/approve".equals(normalizedMsg) || "approve".equals(normalizedMsg);
|
||||
boolean isDenyCommand = "/deny".equals(normalizedMsg) || "deny".equals(normalizedMsg);
|
||||
|
||||
@ -249,7 +247,7 @@ public class ChatController {
|
||||
AtomicBoolean approvalEmitterDone = new AtomicBoolean(false);
|
||||
|
||||
sseExecutor.execute(() -> {
|
||||
StreamAccumulator accumulator = new StreamAccumulator();
|
||||
AgentStreamAccumulator accumulator = newAccumulator();
|
||||
AtomicBoolean finalized = new AtomicBoolean(false);
|
||||
try {
|
||||
// 广播 approval_resolved 事件
|
||||
@ -518,6 +516,34 @@ public class ChatController {
|
||||
return emitter;
|
||||
}
|
||||
|
||||
// ---- 重新生成（regenerate=true）：删除会话末尾的 assistant 回答块，
|
||||
// 复用 DB 中的种子 user 消息作为本轮输入，不重复持久化 user 行。
|
||||
// message 字段被忽略，以持久化的种子为准。 ----
|
||||
final boolean regenerate = Boolean.TRUE.equals(request.getRegenerate());
|
||||
final ConversationService.RegenerateSeed regenerateSeed;
|
||||
if (regenerate) {
|
||||
if (!conversationService.isConversationOwner(conversationId, username)) {
|
||||
sendErrorDoneAndComplete(emitter, "无权操作该会话");
|
||||
return emitter;
|
||||
}
|
||||
if (streamTracker.isRunning(conversationId)) {
|
||||
sendErrorDoneAndComplete(emitter, "正在生成回复，请先停止再重新生成");
|
||||
return emitter;
|
||||
}
|
||||
regenerateSeed = conversationService.prepareRegenerate(conversationId);
|
||||
if (regenerateSeed == null) {
|
||||
sendErrorDoneAndComplete(emitter, "当前没有可重新生成的回答");
|
||||
return emitter;
|
||||
}
|
||||
log.info("SSE regenerate: conversationId={}, seedMessageId={}",
|
||||
conversationId, regenerateSeed.seedMessageId());
|
||||
} else {
|
||||
regenerateSeed = null;
|
||||
}
|
||||
final String message = regenerateSeed != null
|
||||
? (regenerateSeed.content() != null ? regenerateSeed.content() : "")
|
||||
: requestMessage;
|
||||
|
||||
// ---- 正常请求：注册流状态并附着首个订阅者 ----
|
||||
streamTracker.register(conversationId);
|
||||
streamTracker.bindRunMeta(conversationId, agentId, username);
|
||||
@ -541,7 +567,7 @@ public class ChatController {
|
||||
AtomicBoolean emitterDone = new AtomicBoolean(false);
|
||||
|
||||
sseExecutor.execute(() -> {
|
||||
StreamAccumulator accumulator = new StreamAccumulator();
|
||||
AgentStreamAccumulator accumulator = newAccumulator();
|
||||
AtomicBoolean finalized = new AtomicBoolean(false);
|
||||
try {
|
||||
conversationService.getOrCreateConversation(conversationId, agentId, username, workspaceId);
|
||||
@ -550,9 +576,15 @@ public class ChatController {
|
||||
// of every other conversation.
|
||||
conversationService.updateConversationModel(conversationId,
|
||||
request.getModelProvider(), request.getModelName());
|
||||
List<MessageContentPart> requestParts = normalizeRequestParts(request);
|
||||
List<MessageContentPart> requestParts = regenerateSeed != null
|
||||
? regenerateSeed.parts()
|
||||
: normalizeRequestParts(request);
|
||||
String promptText = buildPromptText(message, requestParts);
|
||||
conversationService.saveMessage(conversationId, "user", message, requestParts);
|
||||
if (regenerateSeed == null) {
|
||||
// Regenerate reuses the already-persisted seed user row —
|
||||
// inserting again would duplicate it (issue #547).
|
||||
conversationService.saveMessage(conversationId, "user", message, requestParts);
|
||||
}
|
||||
conversationService.updateStreamStatus(conversationId, "running");
|
||||
|
||||
broadcastEvent(conversationId, "session", Map.of(
|
||||
@ -1339,6 +1371,12 @@ public class ChatController {
|
||||
* end-user when one MateClaw account fronts many of them.
|
||||
*/
|
||||
private String endUserId;
|
||||
/**
|
||||
* true 表示重新生成：删除会话末尾的 assistant 回答块，复用其前最近一条
|
||||
* 已持久化的 user 消息作为本轮输入（{@link #message} 字段被忽略），且不
|
||||
* 重复插入 user 行。生成中的会话拒绝该请求。
|
||||
*/
|
||||
private Boolean regenerate;
|
||||
}
|
||||
|
||||
/**
|
||||
@ -1400,7 +1438,7 @@ public class ChatController {
|
||||
streamTracker.attach(conversationId, emitter);
|
||||
|
||||
// 启动新的流（复用现有 sseExecutor.execute 的逻辑模式）
|
||||
StreamAccumulator accumulator = new StreamAccumulator();
|
||||
AgentStreamAccumulator accumulator = newAccumulator();
|
||||
AtomicBoolean finalized = new AtomicBoolean(false);
|
||||
|
||||
broadcastEvent(conversationId, "message_start", Map.of("role", "assistant"));
|
||||
@ -1532,6 +1570,22 @@ public class ChatController {
|
||||
() -> emergencySaveAccumulator(conversationId, accumulator));
|
||||
}
|
||||
|
||||
/**
|
||||
* Terminal error path for requests rejected before a stream is registered:
|
||||
* emit an {@code error} + terminal {@code done} pair and complete the
|
||||
* emitter, so the client's SSE reader exits cleanly instead of waiting
|
||||
* for a timeout.
|
||||
*/
|
||||
private void sendErrorDoneAndComplete(SseEmitter emitter, String errorMessage) {
|
||||
try {
|
||||
sendEvent(emitter, "error", Map.of("message", errorMessage));
|
||||
sendEvent(emitter, "done", Map.of("status", "completed"));
|
||||
} catch (IOException e) {
|
||||
log.warn("SSE pre-stream error send failed: {}", e.getMessage());
|
||||
}
|
||||
emitter.complete();
|
||||
}
|
||||
|
||||
private void sendEvent(SseEmitter emitter, String name, Object data) throws IOException {
|
||||
String payload;
|
||||
try {
|
||||
@ -1616,7 +1670,7 @@ public class ChatController {
|
||||
}
|
||||
|
||||
private MessageEntity saveEmptyAssistantPlaceholder(String conversationId, String status,
|
||||
StreamAccumulator accumulator, String source) {
|
||||
AgentStreamAccumulator accumulator, String source) {
|
||||
log.warn("{} with empty accumulator: conversationId={}, status={}, finishReason={}, phase={}, hasSegments={}",
|
||||
source, conversationId, status, accumulator.getFinishReason(),
|
||||
accumulator.getCurrentPhase(), !accumulator.segmentsEmpty());
|
||||
@ -1722,7 +1776,7 @@ public class ChatController {
|
||||
* (race window is sub-second between dispose and save) and acceptable. Skipping
|
||||
* save when nothing to save avoids empty rows.
|
||||
*/
|
||||
private void emergencySaveAccumulator(String conversationId, StreamAccumulator accumulator) {
|
||||
private void emergencySaveAccumulator(String conversationId, AgentStreamAccumulator accumulator) {
|
||||
try {
|
||||
String text = accumulator.getContent();
|
||||
List<MessageContentPart> parts = accumulator.toAssistantParts();
|
||||
@ -1867,491 +1921,23 @@ public class ChatController {
|
||||
|| lower.contains("client abort") || lower.contains("closed");
|
||||
}
|
||||
|
||||
/** Markdown link pointing at a generated-file download URL. Used by the
|
||||
* StreamAccumulator to surface generated artifacts in the run-overview rail. */
|
||||
private static final Pattern GENERATED_FILE_LINK_PATTERN =
|
||||
Pattern.compile("\\[([^\\]]+)\\]\\(((?:https?://[^/\\s)\\]]+)?/api/v1/files/generated/[A-Za-z0-9-]+)\\)");
|
||||
|
||||
/**
|
||||
* 流式累积器 — 收集 StreamDelta 事件，持久化到 DB。
|
||||
* <p>
|
||||
* 维护两份数据：
|
||||
* <ul>
|
||||
* <li>{@code toolCalls} — 兼容旧逻辑（执行面板等 UI 使用）</li>
|
||||
* <li>{@code segments} — 按事件到达顺序记录的有序时间线（前端分段渲染用）</li>
|
||||
* </ul>
|
||||
* 两份数据从同一事件流构建，保证一致。segments 保留了 thinking → tools → content
|
||||
* 的真实交错顺序，toolCalls 是 segments 中 tool_call 类型的平铺视图。
|
||||
* Build a per-stream accumulator wired to this controller's SSE
|
||||
* broadcast and phase tracking. Kept as a factory so every stream gets
|
||||
* its own instance while the fan-out semantics stay in one place.
|
||||
*/
|
||||
private final class StreamAccumulator {
|
||||
private final StringBuilder content = new StringBuilder();
|
||||
private final StringBuilder thinking = new StringBuilder();
|
||||
private final List<Map<String, Object>> toolCalls = new ArrayList<>();
|
||||
/** 有序事件时间线 — 前端分段渲染的权威数据源 */
|
||||
private final List<Map<String, Object>> segments = new ArrayList<>();
|
||||
private final List<Map<String, Object>> browserActions = new ArrayList<>();
|
||||
private final List<String> warnings = new ArrayList<>();
|
||||
private final List<Map<String, Object>> planStepResults = new ArrayList<>();
|
||||
/** RFC-052: tool names whose returnDirect output was folded into the assistant message */
|
||||
private final List<String> directToolNames = new ArrayList<>();
|
||||
/** Generated file artifacts extracted from tool results — surfaced in the run-overview rail. */
|
||||
private final List<Map<String, Object>> generatedFiles = new ArrayList<>();
|
||||
private int segCounter = 0;
|
||||
private int promptTokens = 0;
|
||||
private int completionTokens = 0;
|
||||
private int cacheReadTokens = 0;
|
||||
private int cacheWriteTokens = 0;
|
||||
private int reasoningTokens = 0;
|
||||
private String runtimeModelName = "";
|
||||
private String runtimeProviderId = "";
|
||||
private boolean awaitingApproval = false;
|
||||
private String currentPhase = "";
|
||||
/**
|
||||
* Graph-emitted FinishReason for the turn (e.g. {@code "incomplete"},
|
||||
* {@code "stopped"}, {@code "evidence_insufficient"}). Sourced from
|
||||
* the {@code finish_reason} {@link vip.mate.agent.GraphEventPublisher}
|
||||
* event that {@code FinalAnswerNode} attaches to its PENDING_EVENTS
|
||||
* output — same pipeline the SSE accumulator already drains, so the
|
||||
* value is delivered alongside the assistant content (not via a
|
||||
* sibling SSE-only broadcast that would bypass this accumulator).
|
||||
* Persisted into message metadata so downstream filters
|
||||
* (memory promotion gate) see a machine-readable status instead of
|
||||
* having to guess from text. Empty string until the event arrives.
|
||||
*/
|
||||
private String finishReason = "";
|
||||
/**
|
||||
* Recovery affordance payload from {@link
|
||||
* vip.mate.agent.GraphEventPublisher#feedback}. Persisted into
|
||||
* {@code metadata.feedbackEvent} so a page reload still surfaces
|
||||
* the retry/regenerate/report card on the failed assistant
|
||||
* bubble. Null when the turn ended cleanly.
|
||||
*/
|
||||
private Map<String, Object> feedbackEvent = null;
|
||||
private Long planId = null;
|
||||
private List<String> planSteps = List.of();
|
||||
private Integer currentPlanStep = null;
|
||||
private Map<String, Object> pendingApproval = null;
|
||||
/**
|
||||
* Multimodal sidecar routing decision for this turn (null when no
|
||||
* routing happened). Captured from the {@code _routing_decision}
|
||||
* event emitted before the graph stream and folded into
|
||||
* {@code metadata.routing} on persistence so the chat UI can show
|
||||
* which sidecar (if any) was invoked.
|
||||
*/
|
||||
private Map<String, Object> routingDecision = null;
|
||||
|
||||
synchronized void accept(AgentService.StreamDelta delta, String conversationId) {
|
||||
if (delta == null) return;
|
||||
|
||||
if (delta.isEvent()) {
|
||||
if ("_usage_final".equals(delta.eventType())) {
|
||||
Map<String, Object> data = delta.eventData();
|
||||
promptTokens = ((Number) data.getOrDefault("promptTokens", 0)).intValue();
|
||||
completionTokens = ((Number) data.getOrDefault("completionTokens", 0)).intValue();
|
||||
cacheReadTokens = ((Number) data.getOrDefault("cacheReadTokens", 0)).intValue();
|
||||
cacheWriteTokens = ((Number) data.getOrDefault("cacheWriteTokens", 0)).intValue();
|
||||
reasoningTokens = ((Number) data.getOrDefault("reasoningTokens", 0)).intValue();
|
||||
runtimeModelName = String.valueOf(data.getOrDefault("runtimeModelName", ""));
|
||||
runtimeProviderId = String.valueOf(data.getOrDefault("runtimeProviderId", ""));
|
||||
return;
|
||||
}
|
||||
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");
|
||||
}
|
||||
}
|
||||
if ("finish_reason".equals(delta.eventType())) {
|
||||
Object reason = delta.eventData().get("reason");
|
||||
if (reason != null) {
|
||||
// Last-write-wins: graph normally fires this exactly once
|
||||
// at FinalAnswerNode completion. Replay paths that re-enter
|
||||
// the graph after approval will emit a fresh value, which
|
||||
// is the correct behavior — the latest reason is what gets
|
||||
// persisted with the assistant message.
|
||||
finishReason = String.valueOf(reason);
|
||||
}
|
||||
}
|
||||
if (vip.mate.agent.GraphEventPublisher.EVENT_FEEDBACK
|
||||
.equals(delta.eventType())) {
|
||||
// Snapshot the affordance payload so it persists into
|
||||
// message metadata. The same event is also rebroadcast
|
||||
// live (via the broadcastEvent fall-through below) so
|
||||
// an already-mounted UI sees it instantly without
|
||||
// waiting for the message-save round trip.
|
||||
feedbackEvent = delta.eventData();
|
||||
}
|
||||
if (vip.mate.agent.GraphEventPublisher.EVENT_ROUTING_DECISION.equals(delta.eventType())) {
|
||||
// Captured at turn start; persisted under metadata.routing so the
|
||||
// chat UI can render which sidecar (if any) was invoked. Internal
|
||||
// event — return early to skip rebroadcast on IM channels.
|
||||
routingDecision = delta.eventData();
|
||||
return;
|
||||
}
|
||||
accumulateToolEvent(delta.eventType(), delta.eventData(), conversationId);
|
||||
try {
|
||||
broadcastEvent(conversationId, delta.eventType(), delta.eventData());
|
||||
} catch (Exception e) {
|
||||
log.warn("Failed to broadcast event {}: {}", delta.eventType(), e.getMessage());
|
||||
}
|
||||
return;
|
||||
private AgentStreamAccumulator newAccumulator() {
|
||||
return new AgentStreamAccumulator(objectMapper, new AgentStreamAccumulator.Sink() {
|
||||
@Override
|
||||
public void broadcast(String conversationId, String eventName, Object payload) {
|
||||
broadcastEvent(conversationId, eventName, payload);
|
||||
}
|
||||
|
||||
// content_delta
|
||||
if (delta.content() != null && !delta.content().isBlank()) {
|
||||
// segmentOnly deltas route per-iteration narration to the
|
||||
// segments timeline only — the persisted top-level content
|
||||
// field stays clean so it carries the final answer span,
|
||||
// not "我来…让我…" concatenations across iterations (issue
|
||||
// #120 narration leg). segmentOnly implies persistenceOnly,
|
||||
// so no broadcast either.
|
||||
if (!delta.segmentOnly()) {
|
||||
content.append(delta.content());
|
||||
}
|
||||
streamTracker.updatePhase(conversationId, "drafting_answer");
|
||||
if (!delta.persistenceOnly()) {
|
||||
broadcastEvent(conversationId, "content_delta", Map.of("delta", delta.content()));
|
||||
}
|
||||
// segments: 追加到当前 running content segment，或创建新的
|
||||
var seg = findLastRunning("content");
|
||||
if (seg != null) {
|
||||
seg.put("text", seg.getOrDefault("text", "") + delta.content());
|
||||
} else {
|
||||
finalizeRunningSegments("thinking");
|
||||
var s = newSegment("content");
|
||||
s.put("text", delta.content());
|
||||
segments.add(s);
|
||||
}
|
||||
@Override
|
||||
public void updatePhase(String conversationId, String phase) {
|
||||
streamTracker.updatePhase(conversationId, phase);
|
||||
}
|
||||
|
||||
// thinking_delta
|
||||
if (delta.thinking() != null && !delta.thinking().isBlank()) {
|
||||
if (!delta.segmentOnly()) {
|
||||
thinking.append(delta.thinking());
|
||||
}
|
||||
if (!delta.persistenceOnly()) {
|
||||
broadcastEvent(conversationId, "thinking_delta", Map.of("delta", delta.thinking()));
|
||||
}
|
||||
var seg = findLastRunning("thinking");
|
||||
if (seg != null) {
|
||||
seg.put("thinkingText", seg.getOrDefault("thinkingText", "") + delta.thinking());
|
||||
} else {
|
||||
var s = newSegment("thinking");
|
||||
s.put("thinkingText", delta.thinking());
|
||||
segments.add(s);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
boolean isAwaitingApproval() { return awaitingApproval; }
|
||||
|
||||
private void accumulateToolEvent(String eventType, Map<String, Object> 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<String, Object> 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<String, Object> tc = new LinkedHashMap<>();
|
||||
// toolCallId is required for history replay to pair the persisted
|
||||
// assistant tool_call with its tool_response — providers reject any
|
||||
// sequence whose ids don't match. Always record it (empty string
|
||||
// when the upstream event didn't carry one, e.g. forced tool calls).
|
||||
tc.put("toolCallId", String.valueOf(data.getOrDefault("toolCallId", "")));
|
||||
tc.put("name", data.getOrDefault("toolName", ""));
|
||||
tc.put("arguments", data.getOrDefault("arguments", ""));
|
||||
tc.put("status", "running");
|
||||
toolCalls.add(tc);
|
||||
// segments: 关闭 running thinking/content，插入 tool_call
|
||||
finalizeRunningSegments("thinking", "content");
|
||||
var seg = newSegment("tool_call");
|
||||
seg.put("toolCallId", String.valueOf(data.getOrDefault("toolCallId", "")));
|
||||
seg.put("toolName", data.getOrDefault("toolName", ""));
|
||||
seg.put("toolArgs", data.getOrDefault("arguments", ""));
|
||||
segments.add(seg);
|
||||
} else if ("tool_direct_result".equals(eventType)) {
|
||||
// RFC-052: returnDirect tool — track the tool name so history
|
||||
// replay can render a "data returned directly by tool" badge.
|
||||
// The actual textual content reaches the user/persistence layer
|
||||
// through the regular content_delta path (FinalAnswerNode's
|
||||
// FINAL_ANSWER → StateGraphReActAgent → StreamDelta), so we
|
||||
// intentionally do NOT add a content-bearing segment here to
|
||||
// avoid the user seeing the same text twice.
|
||||
String toolName = String.valueOf(data.getOrDefault("toolName", ""));
|
||||
if (!toolName.isBlank() && !directToolNames.contains(toolName)) {
|
||||
directToolNames.add(toolName);
|
||||
}
|
||||
} else if ("tool_call_completed".equals(eventType)) {
|
||||
String toolName = String.valueOf(data.getOrDefault("toolName", ""));
|
||||
String toolCallId = String.valueOf(data.getOrDefault("toolCallId", ""));
|
||||
// toolCalls（兼容）— prefer toolCallId match so parallel calls of
|
||||
// the same tool don't collide on the running+toolName fallback.
|
||||
for (int i = toolCalls.size() - 1; i >= 0; i--) {
|
||||
Map<String, Object> tc = toolCalls.get(i);
|
||||
boolean matches = (!toolCallId.isEmpty()
|
||||
&& toolCallId.equals(String.valueOf(tc.getOrDefault("toolCallId", ""))))
|
||||
|| (toolCallId.isEmpty()
|
||||
&& "running".equals(tc.get("status"))
|
||||
&& toolName.equals(tc.get("name")));
|
||||
if (matches) {
|
||||
tc.put("result", data.getOrDefault("result", ""));
|
||||
tc.put("success", data.getOrDefault("success", true));
|
||||
tc.put("status", "completed");
|
||||
break;
|
||||
}
|
||||
}
|
||||
// segments: 标记对应 tool_call 完成
|
||||
for (int i = segments.size() - 1; i >= 0; i--) {
|
||||
var seg = segments.get(i);
|
||||
if (!"tool_call".equals(seg.get("type"))) continue;
|
||||
boolean matches = (!toolCallId.isEmpty()
|
||||
&& toolCallId.equals(String.valueOf(seg.getOrDefault("toolCallId", ""))))
|
||||
|| (toolCallId.isEmpty()
|
||||
&& "running".equals(seg.get("status"))
|
||||
&& toolName.equals(seg.get("toolName")));
|
||||
if (matches) {
|
||||
seg.put("status", "completed");
|
||||
seg.put("toolResult", data.getOrDefault("result", ""));
|
||||
seg.put("toolSuccess", data.getOrDefault("success", true));
|
||||
break;
|
||||
}
|
||||
}
|
||||
// Extract generated-file links from the tool result so the
|
||||
// run-overview rail can surface artifacts without re-scanning
|
||||
// segments on the frontend.
|
||||
extractGeneratedFiles(String.valueOf(data.getOrDefault("result", "")), toolName);
|
||||
}
|
||||
}
|
||||
|
||||
/** Scan a tool result for markdown links pointing at generated-file
|
||||
* download URLs and collect them into {@link #generatedFiles}.
|
||||
* De-duplicates by URL so a link echoed in later tool results doesn't
|
||||
* produce duplicate entries in the run-overview rail. */
|
||||
private void extractGeneratedFiles(String result, String toolName) {
|
||||
if (result == null || result.isBlank()) return;
|
||||
Matcher m = GENERATED_FILE_LINK_PATTERN.matcher(result);
|
||||
while (m.find()) {
|
||||
String url = m.group(2);
|
||||
boolean dup = generatedFiles.stream()
|
||||
.anyMatch(f -> url.equals(String.valueOf(f.get("url"))));
|
||||
if (dup) continue;
|
||||
Map<String, Object> file = new LinkedHashMap<>();
|
||||
file.put("filename", m.group(1));
|
||||
file.put("url", url);
|
||||
file.put("toolName", toolName);
|
||||
generatedFiles.add(file);
|
||||
}
|
||||
}
|
||||
|
||||
private void ensurePlanStepCapacity(int size) {
|
||||
while (planStepResults.size() < size) {
|
||||
planStepResults.add(null);
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== Segment helpers ====================
|
||||
|
||||
private Map<String, Object> newSegment(String type) {
|
||||
Map<String, Object> seg = new LinkedHashMap<>();
|
||||
seg.put("id", type.substring(0, 2) + "-" + segCounter++);
|
||||
seg.put("type", type);
|
||||
seg.put("status", "running");
|
||||
return seg;
|
||||
}
|
||||
|
||||
private Map<String, Object> findLastRunning(String type) {
|
||||
for (int i = segments.size() - 1; i >= 0; i--) {
|
||||
var seg = segments.get(i);
|
||||
if (type.equals(seg.get("type")) && "running".equals(seg.get("status"))) return seg;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private void finalizeRunningSegments(String... types) {
|
||||
var typeSet = java.util.Set.of(types);
|
||||
for (var seg : segments) {
|
||||
if ("running".equals(seg.get("status")) && typeSet.contains(seg.get("type"))) {
|
||||
seg.put("status", "completed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== 原有访问器 ====================
|
||||
|
||||
String getContent() { return content.toString().trim(); }
|
||||
String getThinking() { return thinking.toString().trim(); }
|
||||
int getPromptTokens() { return promptTokens; }
|
||||
int getCompletionTokens() { return completionTokens; }
|
||||
int getCacheReadTokens() { return cacheReadTokens; }
|
||||
int getCacheWriteTokens() { return cacheWriteTokens; }
|
||||
int getReasoningTokens() { return reasoningTokens; }
|
||||
String getRuntimeModelName() { return runtimeModelName; }
|
||||
String getRuntimeProviderId() { return runtimeProviderId; }
|
||||
String getCurrentPhase() { return currentPhase; }
|
||||
String getFinishReason() { return finishReason; }
|
||||
boolean segmentsEmpty() { return segments.isEmpty(); }
|
||||
|
||||
synchronized List<MessageContentPart> toAssistantParts() {
|
||||
List<MessageContentPart> parts = new ArrayList<>();
|
||||
if (!getContent().isBlank()) {
|
||||
MessageContentPart textPart = new MessageContentPart();
|
||||
textPart.setType("text");
|
||||
textPart.setText(getContent());
|
||||
parts.add(textPart);
|
||||
}
|
||||
if (!getThinking().isBlank()) {
|
||||
MessageContentPart thinkingPart = new MessageContentPart();
|
||||
thinkingPart.setType("thinking");
|
||||
thinkingPart.setText(getThinking());
|
||||
parts.add(thinkingPart);
|
||||
}
|
||||
for (Map<String, Object> tc : toolCalls) {
|
||||
try {
|
||||
parts.add(MessageContentPart.toolCall(objectMapper.writeValueAsString(tc)));
|
||||
} catch (Exception e) {
|
||||
log.warn("Failed to serialize tool call: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
return parts;
|
||||
}
|
||||
|
||||
void finalizeToolCalls() {
|
||||
for (Map<String, Object> tc : toolCalls) {
|
||||
if ("running".equals(tc.get("status"))) tc.put("status", "completed");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 生成 metadata JSON：包含 toolCalls + segments。
|
||||
* toolCalls 保留兼容旧 UI，segments 是按事件顺序的完整时间线。
|
||||
*/
|
||||
synchronized String toMetadataJson() {
|
||||
finalizeToolCalls();
|
||||
finalizeRunningSegments("thinking", "content", "tool_call");
|
||||
SegmentSupersedeDetector.markSuperseded(segments);
|
||||
try {
|
||||
Map<String, Object> metadata = new LinkedHashMap<>();
|
||||
if (!toolCalls.isEmpty()) {
|
||||
metadata.put("toolCalls", toolCalls);
|
||||
}
|
||||
if (!segments.isEmpty()) {
|
||||
metadata.put("segments", segments);
|
||||
}
|
||||
if (!currentPhase.isBlank()) {
|
||||
metadata.put("currentPhase", currentPhase);
|
||||
}
|
||||
if (planId != null || !planSteps.isEmpty() || currentPlanStep != null) {
|
||||
Map<String, Object> 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);
|
||||
}
|
||||
if (!directToolNames.isEmpty()) {
|
||||
// RFC-052 §3.3: only the tool names go into metadata —
|
||||
// the full content already lives in mate_message.content
|
||||
// (assembled by FinalAnswerNode). UI uses this to badge
|
||||
// historical messages as "data returned directly by tool".
|
||||
metadata.put("directToolNames", directToolNames);
|
||||
}
|
||||
if (!generatedFiles.isEmpty()) {
|
||||
metadata.put("generatedFiles", generatedFiles);
|
||||
}
|
||||
if (!finishReason.isEmpty()) {
|
||||
// Surface graph FinishReason so MemorySummarizationGate and
|
||||
// any other downstream consumer can branch on a structured
|
||||
// status (e.g. skip INCOMPLETE / STOPPED / ERROR_FALLBACK
|
||||
// turns from long-term memory promotion) instead of doing
|
||||
// brittle text matching on the assistant content.
|
||||
metadata.put("finishReason", finishReason);
|
||||
}
|
||||
if (feedbackEvent != null && !feedbackEvent.isEmpty()) {
|
||||
// Persist the recovery-affordance payload so the
|
||||
// retry/regenerate/report card survives page reload.
|
||||
// Stored as-is (errorType, errorMessage, actions,
|
||||
// timestamp) — frontend MessageBubble reads
|
||||
// metadata.feedbackEvent and renders one button per
|
||||
// entry in `actions`.
|
||||
metadata.put("feedbackEvent", feedbackEvent);
|
||||
}
|
||||
if (routingDecision != null && !routingDecision.isEmpty()) {
|
||||
metadata.put("routing", routingDecision);
|
||||
}
|
||||
return objectMapper.writeValueAsString(metadata);
|
||||
} catch (Exception e) {
|
||||
log.warn("Failed to serialize metadata: {}", e.getMessage());
|
||||
return "{}";
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private static Long parseLongOrNull(String s) {
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
package vip.mate.channel.webchat;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
@ -193,7 +194,11 @@ public class WebChatController {
|
||||
// 保存用户消息（含访客本轮引用的附件）。附件元数据一律服务端按 fileId 回查，
|
||||
// 不信客户端传入；path 用于 Agent 侧工具读取，对外消息视图会被剥离。
|
||||
List<MessageContentPart> userParts = buildUserParts(conversationId, message, request.getAttachmentIds());
|
||||
conversationService.saveMessage(conversationId, "user", message, userParts);
|
||||
if (!request.isInternalSkipUserPersist()) {
|
||||
// Regenerate reuses the already-persisted seed user row —
|
||||
// inserting again would duplicate it.
|
||||
conversationService.saveMessage(conversationId, "user", message, userParts);
|
||||
}
|
||||
|
||||
// 初始化 SSE 流跟踪
|
||||
streamTracker.register(conversationId);
|
||||
@ -1425,29 +1430,27 @@ public class WebChatController {
|
||||
// right disposable; multi-node is a separate epic.
|
||||
streamTracker.requestStop(conversationId);
|
||||
|
||||
MessageEntity lastAssistant = conversationService.findLastMessageByRole(conversationId, "assistant");
|
||||
if (lastAssistant != null) {
|
||||
conversationService.deleteMessageById(lastAssistant.getId());
|
||||
}
|
||||
MessageEntity lastUser = conversationService.findLastMessageByRole(conversationId, "user");
|
||||
if (lastUser == null) {
|
||||
ConversationService.RegenerateSeed seed = conversationService.prepareRegenerate(conversationId);
|
||||
if (seed == null) {
|
||||
sendErrorAndComplete(emitter, "No user message to regenerate from");
|
||||
return emitter;
|
||||
}
|
||||
|
||||
log.info("[WebChat] Regenerate: conversationId={}, visitor={}, seedMessageId={}",
|
||||
conversationId, visitorId, lastUser.getId());
|
||||
conversationId, visitorId, seed.seedMessageId());
|
||||
audit(channel, visitorId, "webchat.regenerate-session", conversationId,
|
||||
"{\"sessionId\":\"" + sid + "\",\"seedMessageId\":" + lastUser.getId() + "}");
|
||||
"{\"sessionId\":\"" + sid + "\",\"seedMessageId\":" + seed.seedMessageId() + "}");
|
||||
|
||||
// Reuse chatStream: it'll resolve the agent again (cheap), re-derive
|
||||
// conversationId, saveMessage user (new id, same content), and start
|
||||
// the agent turn. visitorId echoes through to keep the visitor-scoped
|
||||
// memory owner consistent.
|
||||
// conversationId and start the agent turn. The seed user row is reused
|
||||
// as-is — internalSkipUserPersist stops chatStream from inserting a
|
||||
// duplicate user row. visitorId echoes through to keep the
|
||||
// visitor-scoped memory owner consistent.
|
||||
WebChatRequest req = new WebChatRequest();
|
||||
req.setMessage(lastUser.getContent());
|
||||
req.setMessage(seed.content());
|
||||
req.setVisitorId(visitorId);
|
||||
req.setSessionId(sid);
|
||||
req.setInternalSkipUserPersist(true);
|
||||
return chatStream(apiKey, req);
|
||||
}
|
||||
|
||||
@ -1656,7 +1659,7 @@ public class WebChatController {
|
||||
return full;
|
||||
}
|
||||
return "webchat:" + key8 + ":#"
|
||||
+ sha256Hex(visitorId + " " + (sessionId == null ? "" : sessionId)).substring(0, 40);
|
||||
+ sha256Hex(visitorId + "\0" + (sessionId == null ? "" : sessionId)).substring(0, 40);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -1917,6 +1920,13 @@ public class WebChatController {
|
||||
* for this conversation. Metadata is resolved server-side; unknown / foreign / expired
|
||||
* ids are dropped. */
|
||||
private List<String> attachmentIds;
|
||||
/**
|
||||
* Internal-only regenerate flag: the seed user row is already
|
||||
* persisted, so {@code chatStream} must not insert a duplicate.
|
||||
* Excluded from JSON binding — never client-settable.
|
||||
*/
|
||||
@JsonIgnore
|
||||
private boolean internalSkipUserPersist;
|
||||
}
|
||||
|
||||
/** Compact view of one of a visitor's conversation threads. */
|
||||
|
||||
@ -794,27 +794,112 @@ public class ConversationService {
|
||||
}
|
||||
|
||||
/**
|
||||
* Find the most recent message of a given role in a conversation.
|
||||
* Used by the webchat regenerate flow to find the seed user message and
|
||||
* locate the assistant reply to delete. Returns null if no match.
|
||||
* Post-rewind snapshot handed back to the controller so the UI can sync
|
||||
* the sidebar (message count + preview) without a follow-up query.
|
||||
*/
|
||||
public MessageEntity findLastMessageByRole(String conversationId, String role) {
|
||||
List<MessageEntity> msgs = messageMapper.selectList(new LambdaQueryWrapper<MessageEntity>()
|
||||
.eq(MessageEntity::getConversationId, conversationId)
|
||||
.eq(MessageEntity::getRole, role)
|
||||
.orderByDesc(MessageEntity::getId)
|
||||
.last("LIMIT 1"));
|
||||
return msgs.isEmpty() ? null : msgs.get(0);
|
||||
public record RewindResult(int deletedCount, int messageCount, String lastMessage) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete a single message by its primary key. Used by the webchat
|
||||
* regenerate flow to drop the last assistant reply before re-running.
|
||||
* Does NOT touch the conversation's messageCount counter — that is
|
||||
* rewritten when the new assistant message is persisted by saveMessage.
|
||||
* Delete the given message and every message after it (compression
|
||||
* boundary rows included), returning the conversation to the state just
|
||||
* before that message. Aggregate counters ({@code messageCount},
|
||||
* {@code lastMessage}, {@code lastActiveTime}) are recomputed from the
|
||||
* surviving rows.
|
||||
*
|
||||
* <p>回退到指定消息：删除该消息及其之后的所有消息，并重算会话统计。
|
||||
*
|
||||
* @return snapshot of the post-rewind state, or {@code null} when the
|
||||
* message does not belong to the conversation
|
||||
*/
|
||||
public void deleteMessageById(Long messageId) {
|
||||
messageMapper.deleteById(messageId);
|
||||
@Transactional
|
||||
public RewindResult rewindToMessage(String conversationId, Long messageId) {
|
||||
List<MessageEntity> all = listMessages(conversationId);
|
||||
int index = -1;
|
||||
for (int i = 0; i < all.size(); i++) {
|
||||
if (messageId.equals(all.get(i).getId())) {
|
||||
index = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (index < 0) {
|
||||
return null;
|
||||
}
|
||||
List<Long> doomedIds = all.subList(index, all.size()).stream()
|
||||
.map(MessageEntity::getId)
|
||||
.toList();
|
||||
messageMapper.delete(new LambdaQueryWrapper<MessageEntity>()
|
||||
.eq(MessageEntity::getConversationId, conversationId)
|
||||
.in(MessageEntity::getId, doomedIds));
|
||||
|
||||
List<MessageEntity> remaining = all.subList(0, index);
|
||||
String lastMessage = remaining.stream()
|
||||
.filter(m -> "assistant".equals(m.getRole()))
|
||||
.reduce((first, second) -> second)
|
||||
.map(this::assistantPreview)
|
||||
.orElse(null);
|
||||
ConversationEntity conv = conversationMapper.selectOne(new LambdaQueryWrapper<ConversationEntity>()
|
||||
.eq(ConversationEntity::getConversationId, conversationId));
|
||||
if (conv != null) {
|
||||
conv.setMessageCount(remaining.size());
|
||||
conv.setLastMessage(lastMessage);
|
||||
conv.setLastActiveTime(LocalDateTime.now());
|
||||
conversationMapper.updateById(conv);
|
||||
}
|
||||
log.info("[Conversation] Rewound conv={} to before message {}, deleted {} rows",
|
||||
conversationId, messageId, doomedIds.size());
|
||||
return new RewindResult(doomedIds.size(), remaining.size(), lastMessage);
|
||||
}
|
||||
|
||||
/**
|
||||
* Seed for a regenerate turn: the persisted user message whose reply is
|
||||
* being regenerated. {@code parts} are already deserialized so the caller
|
||||
* can rebuild the prompt exactly as the original turn saw it.
|
||||
*/
|
||||
public record RegenerateSeed(Long seedMessageId, String content, List<MessageContentPart> parts) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Prepare a regenerate turn: locate the most recent user message, delete
|
||||
* every row after it (the assistant reply block, including any trailing
|
||||
* system/boundary rows), and return that user message as the new turn's
|
||||
* input. The caller re-runs the agent WITHOUT persisting a new user row,
|
||||
* so {@code mate_message} stays free of duplicates.
|
||||
*
|
||||
* <p>When the user message is already the conversation tail (the reply
|
||||
* never got persisted — stream died mid-turn), nothing is deleted and the
|
||||
* seed is still returned, which doubles as the recovery path.
|
||||
*
|
||||
* <p>准备重新生成：删除最近一条 user 消息之后的所有行并返回该消息作为种子。
|
||||
*
|
||||
* @return the seed, or {@code null} when the conversation has no user
|
||||
* message to regenerate from
|
||||
*/
|
||||
@Transactional
|
||||
public RegenerateSeed prepareRegenerate(String conversationId) {
|
||||
List<MessageEntity> all = listMessages(conversationId);
|
||||
int i = all.size() - 1;
|
||||
while (i >= 0 && !"user".equals(all.get(i).getRole())) {
|
||||
i--;
|
||||
}
|
||||
if (i < 0) {
|
||||
return null;
|
||||
}
|
||||
MessageEntity seed = all.get(i);
|
||||
if (i + 1 < all.size()) {
|
||||
rewindToMessage(conversationId, all.get(i + 1).getId());
|
||||
}
|
||||
return new RegenerateSeed(seed.getId(), seed.getContent(), parseMessageParts(seed));
|
||||
}
|
||||
|
||||
/**
|
||||
* Sidebar preview of an assistant reply — same summarization and 50-char
|
||||
* truncation that {@link #saveMessage} applies to the
|
||||
* {@code last_message} column.
|
||||
*/
|
||||
private String assistantPreview(MessageEntity message) {
|
||||
String summary = summarizeMessage(message.getContent(), parseMessageParts(message));
|
||||
return summary.length() > 50 ? summary.substring(0, 50) + "..." : summary;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@ -215,6 +215,30 @@ public class ConversationController {
|
||||
return R.ok(deleted);
|
||||
}
|
||||
|
||||
/**
|
||||
* 回退到指定消息：删除该消息及其之后的所有消息，会话回到该消息之前的状态。
|
||||
* 生成中的会话拒绝回退（409），避免与在途流写入竞争。
|
||||
*/
|
||||
@Operation(summary = "回退会话到指定消息之前")
|
||||
@PostMapping("/{conversationId}/messages/{messageId}/rewind")
|
||||
public R<ConversationService.RewindResult> rewindToMessage(@PathVariable String conversationId,
|
||||
@PathVariable Long messageId,
|
||||
Authentication auth) {
|
||||
String username = auth != null ? auth.getName() : "anonymous";
|
||||
if (!conversationService.isConversationOwner(conversationId, username)) {
|
||||
return R.fail(403, "无权操作该会话");
|
||||
}
|
||||
if (streamTracker.isRunning(conversationId)) {
|
||||
return R.fail(409, "正在生成回复，请先停止再回退");
|
||||
}
|
||||
ConversationService.RewindResult result =
|
||||
conversationService.rewindToMessage(conversationId, messageId);
|
||||
if (result == null) {
|
||||
return R.fail(404, "消息不存在或不属于该会话");
|
||||
}
|
||||
return R.ok(result);
|
||||
}
|
||||
|
||||
/**
|
||||
* 清空会话消息（保留会话记录）
|
||||
*/
|
||||
|
||||
@ -0,0 +1,174 @@
|
||||
package vip.mate.workspace.conversation;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.jupiter.api.DisplayName;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.InjectMocks;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.Spy;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import vip.mate.workspace.conversation.model.ConversationEntity;
|
||||
import vip.mate.workspace.conversation.model.MessageEntity;
|
||||
import vip.mate.workspace.conversation.repository.ConversationMapper;
|
||||
import vip.mate.workspace.conversation.repository.MessageMapper;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* Rewind + regenerate data contract (issue #547):
|
||||
* <ul>
|
||||
* <li>{@code rewindToMessage} deletes the target row and everything after
|
||||
* it, then recomputes the conversation aggregates (messageCount /
|
||||
* lastMessage) from the surviving rows.</li>
|
||||
* <li>{@code prepareRegenerate} drops the trailing reply block (assistant +
|
||||
* trailing system rows) and returns the most recent user message as the
|
||||
* seed, WITHOUT deleting that user row — the caller reuses it instead of
|
||||
* inserting a duplicate.</li>
|
||||
* </ul>
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class ConversationServiceRewindAndRegenerateTest {
|
||||
|
||||
@Mock private ConversationMapper conversationMapper;
|
||||
@Mock private MessageMapper messageMapper;
|
||||
@Spy private ObjectMapper objectMapper = new ObjectMapper();
|
||||
|
||||
@InjectMocks private ConversationService service;
|
||||
|
||||
private static MessageEntity msg(long id, String role, String content) {
|
||||
MessageEntity m = new MessageEntity();
|
||||
m.setId(id);
|
||||
m.setConversationId("conv-1");
|
||||
m.setRole(role);
|
||||
m.setContent(content);
|
||||
return m;
|
||||
}
|
||||
|
||||
private ConversationEntity stubConversation() {
|
||||
ConversationEntity conv = new ConversationEntity();
|
||||
conv.setConversationId("conv-1");
|
||||
conv.setMessageCount(99);
|
||||
when(conversationMapper.selectOne(any())).thenReturn(conv);
|
||||
return conv;
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("rewindToMessage deletes the target and everything after, recomputing aggregates")
|
||||
void rewindMiddleMessage() {
|
||||
when(messageMapper.selectList(any())).thenReturn(List.of(
|
||||
msg(1, "user", "Q1"),
|
||||
msg(2, "assistant", "A1"),
|
||||
msg(3, "user", "Q2"),
|
||||
msg(4, "assistant", "A2")));
|
||||
ConversationEntity conv = stubConversation();
|
||||
|
||||
ConversationService.RewindResult result = service.rewindToMessage("conv-1", 3L);
|
||||
|
||||
assertThat(result).isNotNull();
|
||||
assertThat(result.deletedCount()).isEqualTo(2);
|
||||
assertThat(result.messageCount()).isEqualTo(2);
|
||||
assertThat(result.lastMessage()).isEqualTo("A1");
|
||||
verify(messageMapper).delete(any());
|
||||
ArgumentCaptor<ConversationEntity> updated = ArgumentCaptor.forClass(ConversationEntity.class);
|
||||
verify(conversationMapper).updateById(updated.capture());
|
||||
assertThat(updated.getValue().getMessageCount()).isEqualTo(2);
|
||||
assertThat(updated.getValue().getLastMessage()).isEqualTo("A1");
|
||||
assertThat(conv.getLastActiveTime()).isNotNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("rewinding to the first message clears lastMessage entirely")
|
||||
void rewindToFirstMessageClearsPreview() {
|
||||
when(messageMapper.selectList(any())).thenReturn(List.of(
|
||||
msg(1, "user", "Q1"),
|
||||
msg(2, "assistant", "A1")));
|
||||
stubConversation();
|
||||
|
||||
ConversationService.RewindResult result = service.rewindToMessage("conv-1", 1L);
|
||||
|
||||
assertThat(result).isNotNull();
|
||||
assertThat(result.deletedCount()).isEqualTo(2);
|
||||
assertThat(result.messageCount()).isZero();
|
||||
assertThat(result.lastMessage()).isNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("rewindToMessage returns null for a message not in the conversation")
|
||||
void rewindUnknownMessageReturnsNull() {
|
||||
when(messageMapper.selectList(any())).thenReturn(List.of(msg(1, "user", "Q1")));
|
||||
|
||||
ConversationService.RewindResult result = service.rewindToMessage("conv-1", 42L);
|
||||
|
||||
assertThat(result).isNull();
|
||||
verify(messageMapper, never()).delete(any());
|
||||
verify(conversationMapper, never()).updateById(any(ConversationEntity.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("prepareRegenerate drops the trailing assistant block and returns the seed user row")
|
||||
void prepareRegenerateDeletesTrailingReply() {
|
||||
when(messageMapper.selectList(any())).thenReturn(List.of(
|
||||
msg(1, "user", "Q1"),
|
||||
msg(2, "assistant", "A1"),
|
||||
msg(3, "user", "Q2"),
|
||||
msg(4, "assistant", "A2"),
|
||||
msg(5, "system", "boundary")));
|
||||
stubConversation();
|
||||
|
||||
ConversationService.RegenerateSeed seed = service.prepareRegenerate("conv-1");
|
||||
|
||||
assertThat(seed).isNotNull();
|
||||
assertThat(seed.seedMessageId()).isEqualTo(3L);
|
||||
assertThat(seed.content()).isEqualTo("Q2");
|
||||
// The reply block (assistant + trailing system rows) is gone; the seed
|
||||
// user row survives, so the aggregates reflect rows 1-3.
|
||||
verify(messageMapper).delete(any());
|
||||
ArgumentCaptor<ConversationEntity> updated = ArgumentCaptor.forClass(ConversationEntity.class);
|
||||
verify(conversationMapper).updateById(updated.capture());
|
||||
assertThat(updated.getValue().getMessageCount()).isEqualTo(3);
|
||||
assertThat(updated.getValue().getLastMessage()).isEqualTo("A1");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("prepareRegenerate with the user message already at the tail deletes nothing (recovery path)")
|
||||
void prepareRegenerateUserAtTailDeletesNothing() {
|
||||
when(messageMapper.selectList(any())).thenReturn(List.of(
|
||||
msg(1, "user", "Q1"),
|
||||
msg(2, "assistant", "A1"),
|
||||
msg(3, "user", "Q2")));
|
||||
|
||||
ConversationService.RegenerateSeed seed = service.prepareRegenerate("conv-1");
|
||||
|
||||
assertThat(seed).isNotNull();
|
||||
assertThat(seed.seedMessageId()).isEqualTo(3L);
|
||||
assertThat(seed.content()).isEqualTo("Q2");
|
||||
verify(messageMapper, never()).delete(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("prepareRegenerate returns null when the conversation has no user message")
|
||||
void prepareRegenerateWithoutUserMessageReturnsNull() {
|
||||
when(messageMapper.selectList(any())).thenReturn(List.of(
|
||||
msg(1, "system", "boundary"),
|
||||
msg(2, "assistant", "greeting")));
|
||||
|
||||
assertThat(service.prepareRegenerate("conv-1")).isNull();
|
||||
verify(messageMapper, never()).delete(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("prepareRegenerate on an empty conversation returns null")
|
||||
void prepareRegenerateEmptyConversationReturnsNull() {
|
||||
when(messageMapper.selectList(any())).thenReturn(List.of());
|
||||
|
||||
assertThat(service.prepareRegenerate("conv-1")).isNull();
|
||||
}
|
||||
}
|
||||
@ -212,6 +212,9 @@ export const conversationApi = {
|
||||
http.delete(`/conversations/${encId(conversationId)}`),
|
||||
clearMessages: (conversationId: string) =>
|
||||
http.delete(`/conversations/${encId(conversationId)}/messages`),
|
||||
// messageId is a snowflake ID — keep it a string end-to-end (never Number()).
|
||||
rewindMessage: (conversationId: string, messageId: string) =>
|
||||
http.post(`/conversations/${encId(conversationId)}/messages/${messageId}/rewind`),
|
||||
rename: (conversationId: string, title: string) =>
|
||||
http.put(`/conversations/${encId(conversationId)}/title`, { title }),
|
||||
setPinned: (conversationId: string, pinned: boolean) =>
|
||||
|
||||
@ -412,9 +412,9 @@
|
||||
<el-icon v-else-if="ttsState === 'playing'"><VideoPause /></el-icon>
|
||||
<el-icon v-else><Microphone /></el-icon>
|
||||
</button>
|
||||
<!-- 重新生成（仅 assistant） -->
|
||||
<!-- 重新生成（仅会话末尾的 assistant 回答；服务端只支持对末条回答重生成） -->
|
||||
<button
|
||||
v-if="role === 'assistant' && !isGenerating"
|
||||
v-if="role === 'assistant' && !isGenerating && isLast"
|
||||
class="action-btn"
|
||||
type="button"
|
||||
:title="$t('chat.regenerate')"
|
||||
@ -422,6 +422,17 @@
|
||||
>
|
||||
<el-icon><RefreshRight /></el-icon>
|
||||
</button>
|
||||
<!-- 回退到此处：删除本条及之后的所有消息。仅已持久化的消息可回退
|
||||
（客户端临时 id 带下划线，持久化雪花 id 是纯数字） -->
|
||||
<button
|
||||
v-if="!isGenerating && canRewind"
|
||||
class="action-btn"
|
||||
type="button"
|
||||
:title="$t('chat.rewindHere')"
|
||||
@click="$emit('rewind')"
|
||||
>
|
||||
<el-icon><RefreshLeft /></el-icon>
|
||||
</button>
|
||||
<!-- Reply model attribution (assistant only) -->
|
||||
<span
|
||||
v-if="role === 'assistant' && replyModel"
|
||||
@ -549,6 +560,7 @@ import {
|
||||
Loading,
|
||||
Microphone,
|
||||
Opportunity,
|
||||
RefreshLeft,
|
||||
RefreshRight,
|
||||
Select,
|
||||
Tools,
|
||||
@ -610,6 +622,7 @@ const props = withDefaults(defineProps<Props>(), {
|
||||
|
||||
const emit = defineEmits<{
|
||||
regenerate: []
|
||||
rewind: []
|
||||
'toggle-thinking': [expanded: boolean]
|
||||
approve: [pendingId: string]
|
||||
deny: [pendingId: string]
|
||||
@ -619,6 +632,9 @@ const emit = defineEmits<{
|
||||
const role = computed(() => props.message.role)
|
||||
const status = computed(() => props.message.status)
|
||||
const isGenerating = computed(() => status.value === 'generating' || status.value === 'awaiting_approval')
|
||||
// Only persisted messages can be rewound: DB snowflake ids are pure digits,
|
||||
// client temp ids carry an underscore (`${Date.now()}_${random}`).
|
||||
const canRewind = computed(() => /^\d+$/.test(String(props.message.id ?? '')))
|
||||
const hovered = ref(false)
|
||||
|
||||
const avatarIcon = computed(() => {
|
||||
|
||||
@ -72,6 +72,7 @@
|
||||
:user-icon="userIcon"
|
||||
:show-cursor="showCursorForMessage(msg)"
|
||||
@regenerate="$emit('regenerate', msg)"
|
||||
@rewind="$emit('rewind', msg)"
|
||||
@toggle-thinking="(expanded) => $emit('toggle-thinking', msg, expanded)"
|
||||
@approve="(pendingId) => $emit('approve', pendingId)"
|
||||
@deny="(pendingId) => $emit('deny', pendingId)"
|
||||
@ -159,6 +160,7 @@ const props = withDefaults(defineProps<Props>(), {
|
||||
|
||||
const emit = defineEmits<{
|
||||
regenerate: [message: Message]
|
||||
rewind: [message: Message]
|
||||
'toggle-thinking': [message: Message, expanded: boolean]
|
||||
'suggestion-click': [suggestion: string]
|
||||
scroll: [event: Event]
|
||||
|
||||
@ -183,6 +183,12 @@ export interface SendMessageOptions {
|
||||
modelProvider?: string
|
||||
/** Model id picked for this conversation. Paired with modelProvider. */
|
||||
modelName?: string
|
||||
/**
|
||||
* Regenerate mode: the server deletes the trailing assistant reply block and
|
||||
* reuses the persisted seed user message as this turn's input — no new user
|
||||
* row is inserted and `content` is ignored server-side.
|
||||
*/
|
||||
regenerate?: boolean
|
||||
}
|
||||
|
||||
export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
@ -1931,7 +1937,9 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
lifecycleStage.value = { stage: 'connecting', since: Date.now() }
|
||||
|
||||
try {
|
||||
if (!isApprovalCommand) {
|
||||
if (!isApprovalCommand && !options.regenerate) {
|
||||
// Regenerate reuses the seed user message already in the list — only
|
||||
// a fresh send needs a new local user bubble.
|
||||
createUserMessage(content, contentParts, conversationId)
|
||||
}
|
||||
|
||||
@ -1956,6 +1964,9 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
body.modelProvider = options.modelProvider
|
||||
body.modelName = options.modelName
|
||||
}
|
||||
if (options.regenerate) {
|
||||
body.regenerate = true
|
||||
}
|
||||
await stream.connect(body)
|
||||
} catch (e) {
|
||||
error.value = e instanceof Error ? e : new Error(String(e))
|
||||
@ -2212,27 +2223,23 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
}
|
||||
}
|
||||
|
||||
// Regenerate a message
|
||||
// Regenerate the trailing assistant reply. The server owns the data change:
|
||||
// it drops the trailing assistant block and reuses the persisted seed user
|
||||
// message, so no duplicate user row is created — the local list just mirrors
|
||||
// that truncation before reconnecting.
|
||||
const regenerate = async (messageId: string | number) => {
|
||||
const message = getMessage(messageId)
|
||||
if (!message) return
|
||||
if (!message || message.role !== 'assistant') return
|
||||
|
||||
const index = messages.value.findIndex(m => m.id === messageId)
|
||||
if (index <= 0) return
|
||||
if (index < 0) return
|
||||
|
||||
const userMessage = messages.value[index - 1]
|
||||
if (userMessage.role !== 'user') return
|
||||
messages.value = messages.value.slice(0, index)
|
||||
|
||||
messages.value = messages.value.filter(m => m.id !== messageId)
|
||||
|
||||
const text = userMessage.contentParts
|
||||
.filter(p => p.type === 'text')
|
||||
.map(p => p.text || '')
|
||||
.join('\n') || userMessage.content || ''
|
||||
|
||||
await sendMessage(text, {
|
||||
conversationId: userMessage.conversationId,
|
||||
await sendMessage('', {
|
||||
conversationId: message.conversationId,
|
||||
agentId: '',
|
||||
regenerate: true,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@ -37,7 +37,6 @@ export interface UseMessagesReturn {
|
||||
/** 追加消息内容 */
|
||||
appendMessageContent: (id: string | number, content: string, type?: 'text' | 'thinking') => void
|
||||
/** 删除消息 */
|
||||
removeMessage: (id: string | number) => void
|
||||
/** 清空消息 */
|
||||
clearMessages: () => void
|
||||
/** 设置消息状态 */
|
||||
@ -164,15 +163,6 @@ export function useMessages(options: UseMessagesOptions = {}): UseMessagesReturn
|
||||
})
|
||||
}
|
||||
|
||||
// 删除消息
|
||||
const removeMessage = (id: string | number) => {
|
||||
const index = messages.value.findIndex(m => m.id === id)
|
||||
if (index === -1) return
|
||||
|
||||
messages.value = messages.value.filter(m => m.id !== id)
|
||||
onUpdate?.(messages.value)
|
||||
}
|
||||
|
||||
// 清空消息
|
||||
const clearMessages = () => {
|
||||
messages.value = []
|
||||
@ -238,7 +228,6 @@ export function useMessages(options: UseMessagesOptions = {}): UseMessagesReturn
|
||||
addMessage,
|
||||
updateMessage,
|
||||
appendMessageContent,
|
||||
removeMessage,
|
||||
clearMessages,
|
||||
setMessageStatus,
|
||||
getMessage,
|
||||
|
||||
@ -198,6 +198,10 @@ export default {
|
||||
downloadExpired: 'This file has expired or is no longer available. Ask the assistant to generate it again, then download.',
|
||||
downloadFailed: 'Download failed: {reason}',
|
||||
regenerate: 'Regenerate',
|
||||
regenerateFailed: 'Regenerate failed',
|
||||
rewindHere: 'Rewind to here',
|
||||
rewindConfirm: 'This will delete this message and everything after it ({count} messages). Continue?',
|
||||
rewindFailed: 'Rewind failed',
|
||||
replyModel: 'Reply model: {model}',
|
||||
tokenUsageTooltip: 'This turn used {total} tokens ({input} in · {output} out)',
|
||||
tokenUsageTooltipDelegated: 'This turn used {total} tokens ({input} in · {output} out), of which {delegated} came from delegated sub-agents',
|
||||
|
||||
@ -198,6 +198,10 @@ export default {
|
||||
downloadExpired: '文件已失效或过期，请重新让助手生成后再下载',
|
||||
downloadFailed: '下载失败：{reason}',
|
||||
regenerate: '重新生成',
|
||||
regenerateFailed: '重新生成失败',
|
||||
rewindHere: '回退到此处',
|
||||
rewindConfirm: '将删除该消息及其之后的共 {count} 条消息，确定继续吗？',
|
||||
rewindFailed: '回退失败',
|
||||
replyModel: '本条回复模型: {model}',
|
||||
tokenUsageTooltip: '本轮共消耗 {total} tokens（输入 {input} · 输出 {output}）',
|
||||
tokenUsageTooltipDelegated: '本轮共消耗 {total} tokens（输入 {input} · 输出 {output}），其中子 Agent 委派占 {delegated}',
|
||||
|
||||
@ -113,6 +113,7 @@
|
||||
:subtitle="blockingPrompt ? modelPromptText.desc : $t('chat.subtitle')"
|
||||
:suggestions="blockingPrompt ? [] : suggestions"
|
||||
@regenerate="handleRegenerate"
|
||||
@rewind="handleRewind"
|
||||
@suggestion-click="sendSuggestion"
|
||||
@toggle-thinking="handleToggleThinking"
|
||||
@approve="handleApprove"
|
||||
@ -274,6 +275,7 @@ let cachedAgents: import('@/types').Agent[] = []
|
||||
|
||||
<script setup lang="ts">
|
||||
import { ElMessage } from 'element-plus/es/components/message/index'
|
||||
import { ElMessageBox } from 'element-plus/es/components/message-box/index'
|
||||
import { ref, computed, onMounted, onBeforeUnmount, onActivated, onDeactivated, watch, nextTick } from 'vue'
|
||||
import { useRoute, useRouter } from 'vue-router'
|
||||
import { useI18n } from 'vue-i18n'
|
||||
@ -2008,21 +2010,60 @@ function handleStopStream() {
|
||||
stopChatGeneration()
|
||||
}
|
||||
|
||||
function handleRegenerate(message: Message) {
|
||||
if (isGenerating.value) return
|
||||
async function handleRegenerate(message: Message) {
|
||||
if (isGenerating.value || !currentConversationId.value || !selectedAgentId.value) return
|
||||
const idx = messages.value.indexOf(message)
|
||||
if (idx >= 0) {
|
||||
messages.value.splice(idx, 1)
|
||||
// The server drops the trailing assistant block and reuses the persisted
|
||||
// seed user message (no duplicate user row) — mirror the truncation locally.
|
||||
messages.value.splice(idx)
|
||||
}
|
||||
const lastUserMsg = messages.value.findLast(m => m.role === 'user')
|
||||
if (!lastUserMsg) return
|
||||
try {
|
||||
await sendChatMessage('', {
|
||||
conversationId: currentConversationId.value,
|
||||
agentId: selectedAgentId.value,
|
||||
contentParts: [],
|
||||
thinkingLevel: thinkingLevel.value,
|
||||
modelProvider: activeModels.value?.activeLlm?.providerId,
|
||||
modelName: activeModels.value?.activeLlm?.model,
|
||||
regenerate: true,
|
||||
})
|
||||
} catch (e: any) {
|
||||
console.error('Regenerate failed:', e)
|
||||
mcToast.error(e?.message || t('chat.regenerateFailed'))
|
||||
}
|
||||
}
|
||||
|
||||
const text = lastUserMsg.contentParts
|
||||
.filter(p => p.type === 'text')
|
||||
.map(p => p.text || '')
|
||||
.join('\n') || lastUserMsg.content || ''
|
||||
|
||||
handleSendMessage(text)
|
||||
async function handleRewind(message: Message) {
|
||||
if (isGenerating.value || !currentConversationId.value) return
|
||||
const idx = messages.value.indexOf(message)
|
||||
if (idx < 0) return
|
||||
const count = messages.value.length - idx
|
||||
try {
|
||||
await ElMessageBox.confirm(
|
||||
t('chat.rewindConfirm', { count }),
|
||||
t('chat.rewindHere'),
|
||||
{ type: 'warning' }
|
||||
)
|
||||
} catch {
|
||||
return // user cancelled
|
||||
}
|
||||
try {
|
||||
const res = await conversationApi.rewindMessage(
|
||||
currentConversationId.value,
|
||||
String(message.id)
|
||||
)
|
||||
messages.value.splice(idx)
|
||||
// Sync the sidebar entry from the server-recomputed aggregates.
|
||||
const data = res.data as { deletedCount: number; messageCount: number; lastMessage: string | null } | undefined
|
||||
const conv = conversations.value.find(c => c.conversationId === currentConversationId.value)
|
||||
if (conv && data) {
|
||||
conv.lastMessage = data.lastMessage ?? ''
|
||||
conv.messageCount = data.messageCount
|
||||
}
|
||||
} catch (e: any) {
|
||||
mcToast.error(e?.message || t('chat.rewindFailed'))
|
||||
}
|
||||
}
|
||||
|
||||
function sendSuggestion(text: string) {
|
||||
|
||||
Loading…
Reference in New Issue
Block a user