opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-09-13 03:13:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(chat): suppress duplicate assistant row on force-recycle

Browse Source
This commit is contained in:
matevip 2026-05-05 10:39:54 +08:00
parent be2235a493
commit 2ab0bd000b
6 changed files with 210 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
mateclaw-server/src/main/java/vip/mate/agent/runtime/AgentRuntimeController.java
View File

@ -12,6 +12,7 @@ import vip.mate.audit.service.AuditEventService;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.common.result.R;
import vip.mate.exception.MateClawException;
import vip.mate.i18n.I18nService;
import vip.mate.workspace.conversation.ConversationService;
import java.util.LinkedHashMap;
@ -37,6 +38,7 @@ public class AgentRuntimeController {
private final SubagentRegistry subagentRegistry;
private final AuditEventService auditEventService;
private final ConversationService conversationService;
private final I18nService i18nService;
@Operation(summary = "Snapshot of every in-flight agent turn")
@GetMapping("/snapshot")
@ -62,16 +64,7 @@ public class AgentRuntimeController {
requireAdmin(auth);
boolean ok = streamTracker.forceRecycle(conversationId);
if (ok) {
// Flip stream_status off 'running' so the sidebar drops the
// 生成中 badge immediately. The downstream doOnCancel /
// doOnComplete still fires later (when the agent yields) and may
// re-set this to 'idle' — same value, no-op.
try {
conversationService.updateStreamStatus(conversationId, "idle");
} catch (Exception e) {
log.warn("recycle: failed to reset stream_status for {}: {}",
conversationId, e.getMessage());
}
finalizeRecycledConversation(conversationId);
}
recordAudit(auth, "agent-runtime.recycle", conversationId, Map.of("result", ok));
return R.ok(Map.of("recycled", ok));
@ -105,12 +98,7 @@ public class AgentRuntimeController {
for (String cid : ids) {
if (streamTracker.forceRecycle(cid)) {
recycled++;
try {
conversationService.updateStreamStatus(cid, "idle");
} catch (Exception e) {
log.warn("sweep: failed to reset stream_status for {}: {}",
cid, e.getMessage());
}
finalizeRecycledConversation(cid);
}
}
recordAudit(auth, "agent-runtime.sweep", "all",
@ -118,6 +106,36 @@ public class AgentRuntimeController {
return R.ok(Map.of("recycled", recycled, "ids", ids));
}
/**
* Common DB-side cleanup after a successful {@code forceRecycle}:
* <ol>
* <li>Flip {@code stream_status} off 'running' so the sidebar drops the
* 生成中 badge immediately. (The late doOnCancel / doOnComplete may
* re-set this to 'idle' when the agent finally yields — same value,
* no-op.)</li>
* <li>If the conversation's last message is still a user turn — i.e.
* the agent was disposed before any text streamed and the
* emergencySaveCallback found nothing to persist — write a
* "已被用户中止" assistant marker so the UI shows what happened
* instead of a blank reply.</li>
* </ol>
*/
private void finalizeRecycledConversation(String conversationId) {
try {
conversationService.updateStreamStatus(conversationId, "idle");
} catch (Exception e) {
log.warn("recycle: failed to reset stream_status for {}: {}",
conversationId, e.getMessage());
}
try {
conversationService.saveStopMarkerIfDangling(
conversationId, i18nService.msg("chat.stopMarker.userAborted"), "stopped");
} catch (Exception e) {
log.warn("recycle: failed to save stop marker for {}: {}",
conversationId, e.getMessage());
}
}
private void requireAdmin(Authentication auth) {
if (auth == null) {
throw new MateClawException(401, "authentication required");

96
mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
View File

@ -308,7 +308,22 @@ public class ChatController {
})
.doOnComplete(() -> {
if (!finalized.compareAndSet(false, true)) return;
// RFC-067 §4.6: replay can re-trigger an approval (the approved tool
// Force-recycle short-circuit: see main doOnComplete below.
if (streamTracker.isRecycled(conversationId)) {
log.info("SSE replay doOnComplete skipped for force-recycled conversation: {}", conversationId);
try {
conversationService.updateStreamStatus(conversationId, "idle");
} catch (Exception e) {
log.debug("recycled-skip: stream_status reset failed for {}: {}",
conversationId, e.getMessage());
}
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
completeEmitterQuietly(emitter, approvalEmitterDone);
}
return;
}
// Replay can re-trigger an approval (the approved tool
// call may chain into another guarded tool). Derive status the same
// way as the normal stream so awaiting_approval doesn't get masked
// as completed.
@ -356,6 +371,22 @@ public class ChatController {
})
.doOnError(e -> {
if (!finalized.compareAndSet(false, true)) return;
// Force-recycle short-circuit: see main doOnComplete below.
if (streamTracker.isRecycled(conversationId)) {
log.info("SSE replay doOnError skipped for force-recycled conversation: {}, cause={}",
conversationId, e.getMessage());
try {
conversationService.updateStreamStatus(conversationId, "idle");
} catch (Exception ex) {
log.debug("recycled-skip: stream_status reset failed for {}: {}",
conversationId, ex.getMessage());
}
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
completeEmitterQuietly(emitter, approvalEmitterDone);
}
return;
}
boolean isUserStop = e instanceof java.util.concurrent.CancellationException
|| (e.getCause() instanceof java.util.concurrent.CancellationException);
@ -503,6 +534,36 @@ public class ChatController {
})
.doOnComplete(() -> {
if (!finalized.compareAndSet(false, true)) return;
// Force-recycle: the recycle path already wrote a
// "[已被用户中止]" placeholder (or the partial
// content via emergencySave). The agent's flux may
// have completed the same millisecond — skip its
// save + broadcast so we don't append a duplicate
// assistant row below the placeholder. Cleanup
// still runs so queue draining + emitter close
// happen normally.
if (streamTracker.isRecycled(conversationId)) {
log.info("SSE doOnComplete skipped for force-recycled conversation: {}", conversationId);
streamTracker.clearInterruptState(conversationId);
// Defensive: keep DB stream_status consistent with the
// "this turn is over" reality even when we skip the
// save. Force-recycle's controller path already wrote
// 'idle' for the recycled run, so this is normally a
// no-op — but if a register() ever fails to clear the
// marker (e.g. a different turn snuck through), this
// prevents the row leaking at 'running' across refresh.
try {
conversationService.updateStreamStatus(conversationId, "idle");
} catch (Exception e) {
log.debug("recycled-skip: stream_status reset failed for {}: {}",
conversationId, e.getMessage());
}
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
completeEmitterQuietly(emitter, emitterDone);
}
return;
}
// 区分四种完成语义：
// 1. 正常完成（stopRequested=false）→ completed
// 2. 用户主动停止 → stopped
@ -609,6 +670,22 @@ public class ChatController {
boolean wasFirst = finalized.compareAndSet(false, true);
log.info("SSE doOnCancel fired: conversationId={}, wasFirst={}", conversationId, wasFirst);
if (!wasFirst) return;
// Force-recycle short-circuit: see doOnComplete above.
if (streamTracker.isRecycled(conversationId)) {
log.info("SSE doOnCancel skipped for force-recycled conversation: {}", conversationId);
streamTracker.clearInterruptState(conversationId);
try {
conversationService.updateStreamStatus(conversationId, "idle");
} catch (Exception e) {
log.debug("recycled-skip: stream_status reset failed for {}: {}",
conversationId, e.getMessage());
}
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
completeEmitterQuietly(emitter, emitterDone);
}
return;
}
// 区分用户主动停止和 interrupt-with-followup
ChatStreamTracker.InterruptType interruptType = streamTracker.getInterruptType(conversationId);
boolean isInterruptFollowup = interruptType == ChatStreamTracker.InterruptType.USER_INTERRUPT_WITH_FOLLOWUP;
@ -676,6 +753,23 @@ public class ChatController {
log.info("SSE doOnError skipped (finalized by doOnCancel): conversationId={}", conversationId);
return;
}
// Force-recycle short-circuit: see doOnComplete above.
if (streamTracker.isRecycled(conversationId)) {
log.info("SSE doOnError skipped for force-recycled conversation: {}, cause={}",
conversationId, e.getMessage());
streamTracker.clearInterruptState(conversationId);
try {
conversationService.updateStreamStatus(conversationId, "idle");
} catch (Exception ex) {
log.debug("recycled-skip: stream_status reset failed for {}: {}",
conversationId, ex.getMessage());
}
ChatStreamTracker.CompletionResult cr = streamTracker.completeAndConsumeIfLast(conversationId);
if (cr.allDone()) {
completeEmitterQuietly(emitter, emitterDone);
}
return;
}
// CancellationException = 用户主动停止或中断续跑
boolean isUserStop = e instanceof java.util.concurrent.CancellationException

51
mateclaw-server/src/main/java/vip/mate/channel/web/ChatStreamTracker.java
View File

@ -233,6 +233,22 @@ public class ChatStreamTracker {
private final ConcurrentHashMap<String, RunState> runs = new ConcurrentHashMap<>();
/**
* Conversations whose run was force-recycled by an admin. Maps to the
* recycle timestamp so a scheduled cleanup can age entries out (TTL
* matches {@link #DONE_RETENTION_MS} — long enough that any in-flight
* doOnComplete / doOnError firing after the dispose still finds the
* marker, short enough not to leak across sessions).
* <p>
* Read by the SSE doOn* handlers in ChatController to skip a duplicate
* saveMessage when the recycle path already wrote the "[已被用户中止]"
* placeholder. Without this, the agent's late-yielding doOnComplete
* inserts a second assistant row carrying whatever the agent produced
* after the user pressed stop — exactly the behavior the user does
* <em>not</em> want when force-recycling.
*/
private final ConcurrentHashMap<String, Long> recycledConversations = new ConcurrentHashMap<>();
/** 事件 relay：子会话事件转发到父会话（用于 Agent 委派进度可见性） */
private final ConcurrentHashMap<String, List<java.util.function.BiConsumer<String, String>>> eventRelays = new ConcurrentHashMap<>();
@ -451,6 +467,15 @@ public class ChatStreamTracker {
log.info("[ChatStreamTracker] Reset stale stopRequested on register: {}", conversationId);
}
}
// Clear the force-recycle marker on new registration — the recycle
// tombstone is meant to suppress the late doOnComplete of the
// *recycled* run only, not future turns on the same conversation. If
// the user re-prompts ("继续", "重试", a new question, etc.) inside
// the 5-min TTL, this turn must be allowed to save its assistant
// message normally.
if (recycledConversations.remove(conversationId) != null) {
log.info("[ChatStreamTracker] Cleared recycle marker on new register: {}", conversationId);
}
startHeartbeat(conversationId);
log.debug("Stream registered: {}", conversationId);
}
@ -506,6 +531,23 @@ public class ChatStreamTracker {
return state != null && state.stopRequested.get();
}
/**
* Whether this conversation was force-recycled by an admin within the
* recycle marker's TTL ({@link #DONE_RETENTION_MS}). The SSE doOn*
* handlers consult this to skip a duplicate saveMessage when the recycle
* path already wrote the placeholder. Survives {@code runs.remove(...)},
* unlike {@link #isStopRequested(String)}.
*/
public boolean isRecycled(String conversationId) {
Long ts = recycledConversations.get(conversationId);
if (ts == null) return false;
if (System.currentTimeMillis() - ts > DONE_RETENTION_MS) {
recycledConversations.remove(conversationId);
return false;
}
return true;
}
/**
* 广播事件到所有订阅者并缓存到 buffer.
* <p>
@ -1454,6 +1496,11 @@ public class ChatStreamTracker {
log.info("[SSE] Cleanup completed: evicted {} stale RunState entries, {} remaining",
evicted, runs.size());
}
// Age out the recycled-marker map alongside RunState cleanup. Same
// 5-minute retention so a delayed doOnComplete still hits the marker
// while we don't keep entries around forever.
recycledConversations.entrySet().removeIf(e -> now - e.getValue() > DONE_RETENTION_MS);
}
/**
@ -1603,6 +1650,10 @@ public class ChatStreamTracker {
public boolean forceRecycle(String conversationId) {
RunState state = runs.get(conversationId);
if (state == null) return false;
// Mark BEFORE dispose so a doOnComplete that fires the same millisecond
// (the upstream agent flux had already buffered/completed concurrently
// with the dispose call) sees the recycled flag and skips its save.
recycledConversations.put(conversationId, System.currentTimeMillis());
// Persist any partial assistant content first — dispose() only severs
// the downstream subscription, the agent's worker thread keeps running
// and may not yield for minutes. Without this, the conversation row

24
mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java
View File

@ -307,6 +307,30 @@ public class ConversationService {
}
}
/**
* Persist an assistant placeholder marker only when the last message is a
* user turn (i.e., the assistant never got to reply). Used by the admin
* force-recycle path so a torn-down turn leaves a visible "已被用户中止"
* marker instead of an empty conversation. Idempotent: if the previous
* emergency-save path already wrote an assistant row, this is a no-op.
*
* @return the saved message, or {@code null} if the marker was not needed
*/
@Transactional
public MessageEntity saveStopMarkerIfDangling(String conversationId, String markerText, String status) {
List<MessageEntity> recent = messageMapper.selectList(
new LambdaQueryWrapper<MessageEntity>()
.eq(MessageEntity::getConversationId, conversationId)
.orderByDesc(MessageEntity::getCreateTime)
.orderByDesc(MessageEntity::getId)
.last("LIMIT 1"));
if (recent.isEmpty()) return null;
MessageEntity last = recent.get(0);
if (!"user".equals(last.getRole())) return null;
return saveMessage(conversationId, "assistant", markerText, null,
status != null ? status : "stopped");
}
/**
* 获取会话最后一条消息内容（用于 rate limit 防护等场景）
*/

3
mateclaw-server/src/main/resources/messages.properties
View File

@ -280,3 +280,6 @@ err.wiki.vision.disabled=图片识别功能未启用
err.wiki.vision.no_provider=未配置可用的图片识别 provider
err.wiki.vision.provider_failed=图片识别 provider 调用失败
err.wiki.vision.all_failed=所有图片识别 provider 调用失败
# --- Chat: assistant stop / interrupt placeholders ---
chat.stopMarker.userAborted=[已被用户中止]

3
mateclaw-server/src/main/resources/messages_en.properties
View File

@ -287,3 +287,6 @@ err.wiki.vision.disabled=Image vision pipeline is currently disabled
err.wiki.vision.no_provider=No image vision provider is configured
err.wiki.vision.provider_failed=Image vision provider call failed
err.wiki.vision.all_failed=All image vision providers failed
# --- Chat: assistant stop / interrupt placeholders ---
chat.stopMarker.userAborted=[Stopped by user]