mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-14 19:45:08 +08:00
feat(context): preserve spill markers across compaction phases, persist enriched boundary, broadcast compact_status SSE (#110)
This commit is contained in:
parent
86a6829102
commit
51b5eceb7d
@ -22,6 +22,7 @@ import vip.mate.workspace.conversation.ConversationService;
|
|||||||
|
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -131,6 +132,21 @@ public class ConversationWindowManager {
|
|||||||
this.toolResultStorage = toolResultStorage;
|
this.toolResultStorage = toolResultStorage;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Optional stream tracker for broadcasting {@code compact_status}
|
||||||
|
* SSE events. Wired via setter so unit tests can leave it {@code null}
|
||||||
|
* without dragging in the channel layer. When present, every
|
||||||
|
* compaction emits start/skipped/summarize/done events so the
|
||||||
|
* frontend can render a boundary card and a status line in real
|
||||||
|
* time.
|
||||||
|
*/
|
||||||
|
private vip.mate.channel.web.ChatStreamTracker streamTracker;
|
||||||
|
|
||||||
|
@org.springframework.beans.factory.annotation.Autowired(required = false)
|
||||||
|
public void setStreamTracker(vip.mate.channel.web.ChatStreamTracker streamTracker) {
|
||||||
|
this.streamTracker = streamTracker;
|
||||||
|
}
|
||||||
|
|
||||||
// ==================== 状态 ====================
|
// ==================== 状态 ====================
|
||||||
|
|
||||||
/** 摘要缓存:key = "conversationId:oldMessageCount" */
|
/** 摘要缓存:key = "conversationId:oldMessageCount" */
|
||||||
@ -205,6 +221,8 @@ public class ConversationWindowManager {
|
|||||||
if (messages == null || messages.isEmpty()) {
|
if (messages == null || messages.isEmpty()) {
|
||||||
return messages;
|
return messages;
|
||||||
}
|
}
|
||||||
|
long spillsAtEntry = (toolResultStorage != null) ? toolResultStorage.getSpillCount() : 0L;
|
||||||
|
|
||||||
messages = pruneOldToolResultsForModelInput(messages, conversationId, workspaceBasePath);
|
messages = pruneOldToolResultsForModelInput(messages, conversationId, workspaceBasePath);
|
||||||
|
|
||||||
int effectiveMax = (maxInputTokens != null && maxInputTokens > 0)
|
int effectiveMax = (maxInputTokens != null && maxInputTokens > 0)
|
||||||
@ -229,7 +247,7 @@ public class ConversationWindowManager {
|
|||||||
|
|
||||||
// 可用于历史的 token 预算 = max - system - currentMsg - tools - 安全余量
|
// 可用于历史的 token 预算 = max - system - currentMsg - tools - 安全余量
|
||||||
int reservedTokens = systemTokens + currentMsgTokens + toolsTokens + (int) (effectiveMax * 0.05);
|
int reservedTokens = systemTokens + currentMsgTokens + toolsTokens + (int) (effectiveMax * 0.05);
|
||||||
// RFC-025 Change 1: reserve 硬封顶到 effectiveMax 的 50%。
|
// 预留 reserve 硬封顶到 effectiveMax 的 50%。
|
||||||
// 小上下文模型(Ollama 16K、本地 8K)下,systemTokens + currentMsgTokens 很容易
|
// 小上下文模型(Ollama 16K、本地 8K)下,systemTokens + currentMsgTokens 很容易
|
||||||
// 接近或超过 effectiveMax,不封顶会让 historyBudget 变负数导致死循环压缩
|
// 接近或超过 effectiveMax,不封顶会让 historyBudget 变负数导致死循环压缩
|
||||||
// (压缩目标比压缩前还大 → 压缩后又触发压缩)。
|
// (压缩目标比压缩前还大 → 压缩后又触发压缩)。
|
||||||
@ -244,7 +262,8 @@ public class ConversationWindowManager {
|
|||||||
// 尾部保护 token 预算:阈值的 20%(与 Hermes 一致)
|
// 尾部保护 token 预算:阈值的 20%(与 Hermes 一致)
|
||||||
int tailTokenBudget = (int) (triggerThreshold * 0.20);
|
int tailTokenBudget = (int) (triggerThreshold * 0.20);
|
||||||
|
|
||||||
return compactMessages(messages, historyBudget, tailTokenBudget, chatModel, conversationId, agentId);
|
return compactMessages(messages, historyBudget, tailTokenBudget, chatModel,
|
||||||
|
conversationId, agentId, totalTokens, spillsAtEntry);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -260,15 +279,40 @@ public class ConversationWindowManager {
|
|||||||
|
|
||||||
// ==================== 核心压缩逻辑 ====================
|
// ==================== 核心压缩逻辑 ====================
|
||||||
|
|
||||||
|
/** Broadcast a single compact_status event; silent no-op when no tracker is wired. */
|
||||||
|
private void broadcastCompactStatus(String conversationId, String status, Map<String, Object> extra) {
|
||||||
|
if (streamTracker == null || conversationId == null || conversationId.isEmpty()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
Map<String, Object> payload = new java.util.LinkedHashMap<>();
|
||||||
|
payload.put("status", status);
|
||||||
|
payload.put("timestamp", System.currentTimeMillis());
|
||||||
|
if (extra != null) payload.putAll(extra);
|
||||||
|
streamTracker.broadcastObject(conversationId, "compact_status", payload);
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.debug("[ConversationWindow] broadcast compact_status failed: {}", e.getMessage());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private List<Message> compactMessages(List<Message> messages, int historyBudget,
|
private List<Message> compactMessages(List<Message> messages, int historyBudget,
|
||||||
int tailTokenBudget, ChatModel chatModel,
|
int tailTokenBudget, ChatModel chatModel,
|
||||||
String conversationId, Long agentId) {
|
String conversationId, Long agentId,
|
||||||
|
int preTokens, long spillsAtEntry) {
|
||||||
|
broadcastCompactStatus(conversationId, "start", Map.of(
|
||||||
|
"preTokens", preTokens,
|
||||||
|
"messagesIn", messages.size(),
|
||||||
|
"trigger", "token_threshold"
|
||||||
|
));
|
||||||
|
|
||||||
// 动态计算尾部保护边界(替代固定 preserveRecentPairs)
|
// 动态计算尾部保护边界(替代固定 preserveRecentPairs)
|
||||||
int headEnd = 0; // 头部保护:暂不保护(system prompt 已在外部计算)
|
int headEnd = 0; // 头部保护:暂不保护(system prompt 已在外部计算)
|
||||||
int tailStart = findTailBoundary(messages, headEnd, tailTokenBudget);
|
int tailStart = findTailBoundary(messages, headEnd, tailTokenBudget);
|
||||||
|
|
||||||
if (tailStart <= headEnd) {
|
if (tailStart <= headEnd) {
|
||||||
log.debug("[ConversationWindow] 消息数不足以拆分,跳过压缩");
|
log.debug("[ConversationWindow] 消息数不足以拆分,跳过压缩");
|
||||||
|
broadcastCompactStatus(conversationId, "skipped",
|
||||||
|
Map.of("reason", "insufficient_messages"));
|
||||||
return messages;
|
return messages;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -281,8 +325,14 @@ public class ConversationWindowManager {
|
|||||||
// turn.
|
// turn.
|
||||||
int pairSafeCut = enforcePairSafeBoundary(messages, headEnd, tailStart);
|
int pairSafeCut = enforcePairSafeBoundary(messages, headEnd, tailStart);
|
||||||
if (pairSafeCut <= headEnd) {
|
if (pairSafeCut <= headEnd) {
|
||||||
|
broadcastCompactStatus(conversationId, "skipped",
|
||||||
|
Map.of("reason", "pair_boundary_collapsed"));
|
||||||
return messages;
|
return messages;
|
||||||
}
|
}
|
||||||
|
if (pairSafeCut != tailStart) {
|
||||||
|
broadcastCompactStatus(conversationId, "pair_safe", Map.of(
|
||||||
|
"movedFrom", tailStart, "movedTo", pairSafeCut));
|
||||||
|
}
|
||||||
tailStart = pairSafeCut;
|
tailStart = pairSafeCut;
|
||||||
|
|
||||||
List<Message> oldMessages = new ArrayList<>(messages.subList(headEnd, tailStart));
|
List<Message> oldMessages = new ArrayList<>(messages.subList(headEnd, tailStart));
|
||||||
@ -340,13 +390,20 @@ public class ConversationWindowManager {
|
|||||||
// 计算动态摘要预算
|
// 计算动态摘要预算
|
||||||
int summaryBudget = computeSummaryBudget(forSummary);
|
int summaryBudget = computeSummaryBudget(forSummary);
|
||||||
|
|
||||||
|
broadcastCompactStatus(conversationId, "summarize", Map.of(
|
||||||
|
"messagesToSummarize", oldMessages.size(),
|
||||||
|
"summaryBudget", summaryBudget
|
||||||
|
));
|
||||||
|
|
||||||
// 检查缓存
|
// 检查缓存
|
||||||
String cacheKey = conversationId + ":" + oldMessages.size();
|
String cacheKey = conversationId + ":" + oldMessages.size();
|
||||||
CachedSummary cached = summaryCache.get(cacheKey);
|
CachedSummary cached = summaryCache.get(cacheKey);
|
||||||
String summary;
|
String summary;
|
||||||
|
boolean fromCache = false;
|
||||||
|
|
||||||
if (cached != null && !cached.isExpired(CACHE_TTL_MS)) {
|
if (cached != null && !cached.isExpired(CACHE_TTL_MS)) {
|
||||||
summary = cached.summary();
|
summary = cached.summary();
|
||||||
|
fromCache = true;
|
||||||
log.debug("[ConversationWindow] 命中摘要缓存, conv={}", conversationId);
|
log.debug("[ConversationWindow] 命中摘要缓存, conv={}", conversationId);
|
||||||
} else {
|
} else {
|
||||||
summary = generateSummary(forSummary, chatModel, conversationId, summaryBudget, memoryExtraContext);
|
summary = generateSummary(forSummary, chatModel, conversationId, summaryBudget, memoryExtraContext);
|
||||||
@ -355,21 +412,12 @@ public class ConversationWindowManager {
|
|||||||
int count = compressionCounts.merge(conversationId, 1, Integer::sum);
|
int count = compressionCounts.merge(conversationId, 1, Integer::sum);
|
||||||
log.info("[ConversationWindow] 生成结构化摘要 ({} 字符, 第 {} 次压缩), 压缩 {} 条旧消息, conv={}",
|
log.info("[ConversationWindow] 生成结构化摘要 ({} 字符, 第 {} 次压缩), 压缩 {} 条旧消息, conv={}",
|
||||||
summary.length(), count, oldMessages.size(), conversationId);
|
summary.length(), count, oldMessages.size(), conversationId);
|
||||||
|
|
||||||
// 持久化摘要到 DB:下次加载历史时可直接从摘要位置开始,跳过重复压缩
|
|
||||||
if (conversationService != null) {
|
|
||||||
try {
|
|
||||||
conversationService.saveCompressionSummary(
|
|
||||||
conversationId, SUMMARY_PREFIX + summary, oldMessages.size());
|
|
||||||
} catch (Exception e) {
|
|
||||||
log.warn("[ConversationWindow] Failed to persist compression summary: {}", e.getMessage());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 组装结果
|
// 组装结果
|
||||||
List<Message> result = new ArrayList<>();
|
List<Message> result = new ArrayList<>();
|
||||||
|
boolean anchored = false;
|
||||||
if (summary != null && !summary.isBlank()) {
|
if (summary != null && !summary.isBlank()) {
|
||||||
result.add(new UserMessage(SUMMARY_PREFIX + summary));
|
result.add(new UserMessage(SUMMARY_PREFIX + summary));
|
||||||
|
|
||||||
@ -380,11 +428,16 @@ public class ConversationWindowManager {
|
|||||||
Message anchor = buildFirstUserAnchor(oldMessages);
|
Message anchor = buildFirstUserAnchor(oldMessages);
|
||||||
if (anchor != null) {
|
if (anchor != null) {
|
||||||
result.add(anchor);
|
result.add(anchor);
|
||||||
|
anchored = true;
|
||||||
}
|
}
|
||||||
} else if (!oldMessages.isEmpty()) {
|
} else if (!oldMessages.isEmpty()) {
|
||||||
log.warn("[ConversationWindow] 摘要生成失败,降级为保留最近 4 条旧消息, conv={}", conversationId);
|
log.warn("[ConversationWindow] 摘要生成失败,降级为保留最近 4 条旧消息, conv={}", conversationId);
|
||||||
int fallbackKeep = Math.min(4, oldMessages.size());
|
int fallbackKeep = Math.min(4, oldMessages.size());
|
||||||
result.addAll(oldMessages.subList(oldMessages.size() - fallbackKeep, oldMessages.size()));
|
result.addAll(oldMessages.subList(oldMessages.size() - fallbackKeep, oldMessages.size()));
|
||||||
|
broadcastCompactStatus(conversationId, "failed", Map.of(
|
||||||
|
"reason", "summary_generation_failed",
|
||||||
|
"fallbackKept", fallbackKeep
|
||||||
|
));
|
||||||
}
|
}
|
||||||
result.addAll(recentMessages);
|
result.addAll(recentMessages);
|
||||||
|
|
||||||
@ -393,6 +446,42 @@ public class ConversationWindowManager {
|
|||||||
if (resultTokens > historyBudget && result.size() > 2) {
|
if (resultTokens > historyBudget && result.size() > 2) {
|
||||||
log.warn("[ConversationWindow] 压缩后仍超预算: {} > {}, 执行二次裁剪", resultTokens, historyBudget);
|
log.warn("[ConversationWindow] 压缩后仍超预算: {} > {}, 执行二次裁剪", resultTokens, historyBudget);
|
||||||
result = trimToFit(result, historyBudget);
|
result = trimToFit(result, historyBudget);
|
||||||
|
resultTokens = TokenEstimator.estimateTokens(result);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Persist the boundary + announce completion only when the summary
|
||||||
|
// actually wrote a row. Failed-summary fallback already broadcast
|
||||||
|
// its own event above.
|
||||||
|
if (summary != null && !summary.isBlank() && conversationService != null && !fromCache) {
|
||||||
|
long spillsThisTurn = (toolResultStorage != null)
|
||||||
|
? Math.max(0L, toolResultStorage.getSpillCount() - spillsAtEntry)
|
||||||
|
: 0L;
|
||||||
|
Map<String, Object> boundaryMetadata = new java.util.LinkedHashMap<>();
|
||||||
|
boundaryMetadata.put("trigger", "token_threshold");
|
||||||
|
boundaryMetadata.put("preTokens", preTokens);
|
||||||
|
boundaryMetadata.put("postTokens", resultTokens);
|
||||||
|
boundaryMetadata.put("messagesSummarized", oldMessages.size());
|
||||||
|
boundaryMetadata.put("tailKept", recentMessages.size());
|
||||||
|
boundaryMetadata.put("toolResultsSpilled", spillsThisTurn);
|
||||||
|
boundaryMetadata.put("anchored", anchored);
|
||||||
|
try {
|
||||||
|
conversationService.saveCompressionSummary(
|
||||||
|
conversationId, SUMMARY_PREFIX + summary, oldMessages.size(),
|
||||||
|
boundaryMetadata);
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.warn("[ConversationWindow] Failed to persist compression boundary: {}", e.getMessage());
|
||||||
|
}
|
||||||
|
broadcastCompactStatus(conversationId, "done", boundaryMetadata);
|
||||||
|
} else if (summary != null && !summary.isBlank() && fromCache) {
|
||||||
|
// Cached summary path — no new DB row, but emit done so the
|
||||||
|
// frontend status bar still updates.
|
||||||
|
broadcastCompactStatus(conversationId, "done", Map.of(
|
||||||
|
"preTokens", preTokens,
|
||||||
|
"postTokens", resultTokens,
|
||||||
|
"messagesSummarized", oldMessages.size(),
|
||||||
|
"tailKept", recentMessages.size(),
|
||||||
|
"fromCache", true
|
||||||
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
return result;
|
return result;
|
||||||
@ -780,15 +869,36 @@ public class ConversationWindowManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Phase 1 - Soft trim:对工具结果做 head+tail 裁剪(保留首尾各 200 字符)。
|
* Spill-marker responses already point at an on-disk full copy via
|
||||||
|
* {@code path=...} in their body. Trimming, replacing, or pre-pruning
|
||||||
|
* them would destroy the very pointer the model needs to recover the
|
||||||
|
* original output with {@code read_file} — which is the whole reason
|
||||||
|
* we spilled in the first place. All three compaction phases consult
|
||||||
|
* this guard before touching a response.
|
||||||
*/
|
*/
|
||||||
private int softTrimToolResults(List<Message> messages) {
|
static boolean isSpillMarker(ToolResponseMessage.ToolResponse r) {
|
||||||
|
return r != null
|
||||||
|
&& r.responseData() != null
|
||||||
|
&& r.responseData().startsWith(ToolResultStorage.SPILL_MARKER_PREFIX);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Phase 1 - Soft trim:对工具结果做 head+tail 裁剪(保留首尾各 200 字符)。
|
||||||
|
* <p>Spill-marker responses are left untouched so their on-disk pointer
|
||||||
|
* survives intact across compaction.
|
||||||
|
*/
|
||||||
|
int softTrimToolResults(List<Message> messages) {
|
||||||
int trimmed = 0;
|
int trimmed = 0;
|
||||||
for (int i = 0; i < messages.size(); i++) {
|
for (int i = 0; i < messages.size(); i++) {
|
||||||
if (messages.get(i) instanceof ToolResponseMessage trm) {
|
if (messages.get(i) instanceof ToolResponseMessage trm) {
|
||||||
List<ToolResponseMessage.ToolResponse> newResponses = new ArrayList<>();
|
List<ToolResponseMessage.ToolResponse> newResponses = new ArrayList<>();
|
||||||
boolean changed = false;
|
boolean changed = false;
|
||||||
for (ToolResponseMessage.ToolResponse r : trm.getResponses()) {
|
for (ToolResponseMessage.ToolResponse r : trm.getResponses()) {
|
||||||
|
if (isSpillMarker(r)) {
|
||||||
|
// Pointer + preview already; trimming would lose the path.
|
||||||
|
newResponses.add(r);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
String data = r.responseData();
|
String data = r.responseData();
|
||||||
if (data != null && data.length() > 500) {
|
if (data != null && data.length() > 500) {
|
||||||
String head = data.substring(0, 200);
|
String head = data.substring(0, 200);
|
||||||
@ -811,16 +921,28 @@ public class ConversationWindowManager {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Phase 2 - Hard clear:将所有旧工具结果替换为占位符。
|
* Phase 2 - Hard clear:将所有旧工具结果替换为占位符。
|
||||||
|
* <p>Spill-marker responses are left untouched so the on-disk pointer
|
||||||
|
* survives — a placeholder here would force the model to abandon a
|
||||||
|
* tool output it could otherwise recover via {@code read_file}.
|
||||||
*/
|
*/
|
||||||
private int hardClearToolResults(List<Message> messages) {
|
int hardClearToolResults(List<Message> messages) {
|
||||||
int cleared = 0;
|
int cleared = 0;
|
||||||
for (int i = 0; i < messages.size(); i++) {
|
for (int i = 0; i < messages.size(); i++) {
|
||||||
if (messages.get(i) instanceof ToolResponseMessage trm) {
|
if (messages.get(i) instanceof ToolResponseMessage trm) {
|
||||||
List<ToolResponseMessage.ToolResponse> placeholders = trm.getResponses().stream()
|
boolean changed = false;
|
||||||
.map(r -> new ToolResponseMessage.ToolResponse(r.id(), r.name(), "[tool result removed]"))
|
List<ToolResponseMessage.ToolResponse> replaced = new ArrayList<>(trm.getResponses().size());
|
||||||
.toList();
|
for (ToolResponseMessage.ToolResponse r : trm.getResponses()) {
|
||||||
messages.set(i, ToolResponseMessage.builder().responses(placeholders).build());
|
if (isSpillMarker(r)) {
|
||||||
cleared++;
|
replaced.add(r);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
replaced.add(new ToolResponseMessage.ToolResponse(r.id(), r.name(), "[tool result removed]"));
|
||||||
|
changed = true;
|
||||||
|
}
|
||||||
|
if (changed) {
|
||||||
|
messages.set(i, ToolResponseMessage.builder().responses(replaced).build());
|
||||||
|
cleared++;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return cleared;
|
return cleared;
|
||||||
@ -828,18 +950,27 @@ public class ConversationWindowManager {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Phase 3 Pre-prune:在 LLM 摘要前,将工具输出替换为占位符(减少摘要输入 token)。
|
* Phase 3 Pre-prune:在 LLM 摘要前,将工具输出替换为占位符(减少摘要输入 token)。
|
||||||
|
* <p>Spill-marker responses are left untouched so the summary input
|
||||||
|
* still has the on-disk path the model might cite back in its summary.
|
||||||
*/
|
*/
|
||||||
private int prePruneForSummary(List<Message> messages) {
|
int prePruneForSummary(List<Message> messages) {
|
||||||
int pruned = 0;
|
int pruned = 0;
|
||||||
for (int i = 0; i < messages.size(); i++) {
|
for (int i = 0; i < messages.size(); i++) {
|
||||||
if (messages.get(i) instanceof ToolResponseMessage trm) {
|
if (messages.get(i) instanceof ToolResponseMessage trm) {
|
||||||
boolean hasSubstantial = trm.getResponses().stream()
|
boolean hasSubstantial = trm.getResponses().stream()
|
||||||
.anyMatch(r -> r.responseData() != null && r.responseData().length() > 200);
|
.anyMatch(r -> !isSpillMarker(r)
|
||||||
|
&& r.responseData() != null
|
||||||
|
&& r.responseData().length() > 200);
|
||||||
if (hasSubstantial) {
|
if (hasSubstantial) {
|
||||||
List<ToolResponseMessage.ToolResponse> placeholders = trm.getResponses().stream()
|
List<ToolResponseMessage.ToolResponse> placeholders = new ArrayList<>(trm.getResponses().size());
|
||||||
.map(r -> new ToolResponseMessage.ToolResponse(r.id(), r.name(),
|
for (ToolResponseMessage.ToolResponse r : trm.getResponses()) {
|
||||||
"[旧工具输出已清理以节省上下文空间]"))
|
if (isSpillMarker(r)) {
|
||||||
.toList();
|
placeholders.add(r);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
placeholders.add(new ToolResponseMessage.ToolResponse(r.id(), r.name(),
|
||||||
|
"[旧工具输出已清理以节省上下文空间]"));
|
||||||
|
}
|
||||||
messages.set(i, ToolResponseMessage.builder().responses(placeholders).build());
|
messages.set(i, ToolResponseMessage.builder().responses(placeholders).build());
|
||||||
pruned++;
|
pruned++;
|
||||||
}
|
}
|
||||||
|
|||||||
@ -99,12 +99,17 @@ public class ToolResultProperties {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Days to retain spill files before the scheduled cleanup deletes them.
|
* Days to retain spill files before the scheduled cleanup deletes them.
|
||||||
* Spill files exist to let the model recover full tool output via
|
* <p><b>Default 0 means time-based cleanup is disabled</b> — spill files
|
||||||
* {@code read_file} during the active conversation; once the conversation
|
* stay on disk until the owning conversation is explicitly deleted (which
|
||||||
* is dormant for this many days the agent is extremely unlikely to ever
|
* fires {@code purgeConversation} via {@code ConversationService}).
|
||||||
* read the file again, and disk pressure starts to matter.
|
* This preserves the "recoverable" invariant: a summary or preview that
|
||||||
|
* cites a spill path will keep working for the whole life of the
|
||||||
|
* conversation, no matter how long it sits dormant.
|
||||||
|
* <p>Set to a positive value if disk pressure outweighs recoverability
|
||||||
|
* for your deployment. The scheduled sweep will then delete files whose
|
||||||
|
* mtime falls outside the retention horizon.
|
||||||
*/
|
*/
|
||||||
private int retentionDays = 7;
|
private int retentionDays = 0;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Cron expression for the spill-cleanup task. Defaults to once a day at
|
* Cron expression for the spill-cleanup task. Defaults to once a day at
|
||||||
|
|||||||
@ -428,18 +428,65 @@ public class ConversationService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 将压缩摘要持久化为 role=system 的特殊消息。
|
* Persist a compaction boundary as a role=system message. The body is
|
||||||
* 下次加载历史时识别此消息,跳过它之前的已压缩消息。
|
* the summary text; the metadata describes <em>what happened</em> at
|
||||||
|
* this boundary (trigger, pre/post tokens, how many messages were
|
||||||
|
* summarised, how many spill files were produced, how many tail
|
||||||
|
* messages survived). On the next load this row is the cut-off — older
|
||||||
|
* messages are skipped, the model picks up from the summary forward.
|
||||||
|
*
|
||||||
|
* <p>Backward-compat overload: legacy callers that only know the row
|
||||||
|
* count still work and produce a minimal metadata block.
|
||||||
*/
|
*/
|
||||||
public void saveCompressionSummary(String conversationId, String summary, int compressedCount) {
|
public void saveCompressionSummary(String conversationId, String summary, int compressedCount) {
|
||||||
|
saveCompressionSummary(conversationId, summary, compressedCount, Map.of());
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Same as the 3-arg overload but accepts extra structured fields that
|
||||||
|
* are merged into the boundary's metadata JSON. Fields the frontend
|
||||||
|
* and observability pipeline care about:
|
||||||
|
* <ul>
|
||||||
|
* <li>{@code trigger} — what fired this boundary
|
||||||
|
* ({@code token_threshold}, {@code user_compact}, etc.)</li>
|
||||||
|
* <li>{@code preTokens} / {@code postTokens} — context size before
|
||||||
|
* and after, for the in-prompt status row</li>
|
||||||
|
* <li>{@code messagesSummarized} / {@code tailKept} — partition
|
||||||
|
* counts the user sees in the boundary card</li>
|
||||||
|
* <li>{@code toolResultsSpilled} — how many bodies the spill store
|
||||||
|
* absorbed during this boundary</li>
|
||||||
|
* <li>{@code summaryId} — stable id (the inserted message id) for
|
||||||
|
* deep-linking from the SSE event</li>
|
||||||
|
* </ul>
|
||||||
|
* <p>{@code type=compression_summary} is always present — the loader
|
||||||
|
* keys off it. {@code compressedCount} is kept for backward compat.
|
||||||
|
*/
|
||||||
|
public void saveCompressionSummary(String conversationId, String summary, int compressedCount,
|
||||||
|
Map<String, Object> extraMetadata) {
|
||||||
MessageEntity entity = new MessageEntity();
|
MessageEntity entity = new MessageEntity();
|
||||||
entity.setConversationId(conversationId);
|
entity.setConversationId(conversationId);
|
||||||
entity.setRole("system");
|
entity.setRole("system");
|
||||||
entity.setContent(summary);
|
entity.setContent(summary);
|
||||||
entity.setStatus("completed");
|
entity.setStatus("completed");
|
||||||
entity.setMetadata("{\"type\":\"compression_summary\",\"compressedCount\":" + compressedCount + "}");
|
|
||||||
|
Map<String, Object> metadata = new java.util.LinkedHashMap<>();
|
||||||
|
metadata.put("type", "compression_summary");
|
||||||
|
metadata.put("compressedCount", compressedCount);
|
||||||
|
if (extraMetadata != null) {
|
||||||
|
extraMetadata.forEach((k, v) -> {
|
||||||
|
if (v != null) metadata.put(k, v);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
entity.setMetadata(objectMapper.writeValueAsString(metadata));
|
||||||
|
} catch (com.fasterxml.jackson.core.JsonProcessingException e) {
|
||||||
|
log.warn("[Conversation] Failed to serialise compaction metadata, falling back to minimal: {}",
|
||||||
|
e.getMessage());
|
||||||
|
entity.setMetadata("{\"type\":\"compression_summary\",\"compressedCount\":" + compressedCount + "}");
|
||||||
|
}
|
||||||
messageMapper.insert(entity);
|
messageMapper.insert(entity);
|
||||||
log.info("[Conversation] Saved compression summary for conv={}, compressedCount={}", conversationId, compressedCount);
|
log.info("[Conversation] Saved compression boundary conv={}, compressedCount={}, metadata={}",
|
||||||
|
conversationId, compressedCount, entity.getMetadata());
|
||||||
}
|
}
|
||||||
|
|
||||||
public List<MessageVO> listMessageViews(String conversationId) {
|
public List<MessageVO> listMessageViews(String conversationId) {
|
||||||
|
|||||||
@ -229,10 +229,13 @@ mate:
|
|||||||
excluded-tools:
|
excluded-tools:
|
||||||
- read_file
|
- read_file
|
||||||
- read_workspace_memory_file
|
- read_workspace_memory_file
|
||||||
# Spill files are deleted after this many days. Set to 0 to disable the
|
# Spill files are deleted after this many days. Default 0 disables the
|
||||||
# scheduled sweep entirely (files still get purged when the conversation
|
# scheduled sweep entirely so a summary/preview that points at a spill
|
||||||
# is deleted explicitly via ConversationService.deleteConversation).
|
# path stays valid for the whole life of the conversation. Files are
|
||||||
retention-days: 7
|
# still purged when the conversation is deleted explicitly via
|
||||||
|
# ConversationService.deleteConversation. Raise to a positive value if
|
||||||
|
# disk pressure outweighs recoverability for your deployment.
|
||||||
|
retention-days: 0
|
||||||
cleanup-cron: "0 0 3 * * ?"
|
cleanup-cron: "0 0 3 * * ?"
|
||||||
conversation:
|
conversation:
|
||||||
window:
|
window:
|
||||||
|
|||||||
@ -0,0 +1,157 @@
|
|||||||
|
package vip.mate.agent.context;
|
||||||
|
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.springframework.ai.chat.messages.Message;
|
||||||
|
import org.springframework.ai.chat.messages.ToolResponseMessage;
|
||||||
|
import vip.mate.agent.graph.executor.ToolResultStorage;
|
||||||
|
import vip.mate.config.ConversationWindowProperties;
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The three compaction phases (soft trim, hard clear, pre-prune for
|
||||||
|
* summary) must never destroy a spill-marker body — doing so would erase
|
||||||
|
* the {@code path=...} pointer the model needs to recover the original
|
||||||
|
* full output via {@code read_file}, which is the whole reason that body
|
||||||
|
* was spilled in the first place.
|
||||||
|
*
|
||||||
|
* <p>This is the "recoverable" invariant: once a tool output makes it
|
||||||
|
* into the spill store, the in-context representation stays a stable
|
||||||
|
* preview + path for the rest of the conversation regardless of how
|
||||||
|
* aggressively the window manager has to compress the prefix.
|
||||||
|
*/
|
||||||
|
class ConversationWindowManagerSpillMarkerPreservationTest {
|
||||||
|
|
||||||
|
private static final String SPILL_BODY = ToolResultStorage.SPILL_MARKER_PREFIX
|
||||||
|
+ " tool=web_search full_chars=22000 path=/tmp/x.txt\n"
|
||||||
|
+ "[Preview — first 800 of 22000 chars. Use read_file with the path above to retrieve the rest.]\n"
|
||||||
|
+ "preview body fragment that contributes most of the inline size...";
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void softTrimLeavesSpillMarkerUntouched() {
|
||||||
|
ConversationWindowManager mgr = new ConversationWindowManager(
|
||||||
|
new ConversationWindowProperties(), null, null);
|
||||||
|
|
||||||
|
List<Message> messages = new ArrayList<>(List.of(
|
||||||
|
toolMessage("call-spill", "web_search", SPILL_BODY),
|
||||||
|
toolMessage("call-big", "search", "x".repeat(2000))
|
||||||
|
));
|
||||||
|
|
||||||
|
// Pass the same list through Phase 1.
|
||||||
|
int trimmed = mgr.softTrimToolResults(messages);
|
||||||
|
|
||||||
|
// The non-spill body must have been trimmed (it was > 500 chars).
|
||||||
|
// The spill body must remain identical to the original.
|
||||||
|
ToolResponseMessage trm0 = (ToolResponseMessage) messages.get(0);
|
||||||
|
ToolResponseMessage trm1 = (ToolResponseMessage) messages.get(1);
|
||||||
|
assertEquals(SPILL_BODY, trm0.getResponses().getFirst().responseData(),
|
||||||
|
"Phase 1 soft trim must not modify a spill-marker body");
|
||||||
|
assertTrue(trm1.getResponses().getFirst().responseData().contains("[trimmed "),
|
||||||
|
"non-spill bodies should still be trimmed by Phase 1");
|
||||||
|
assertEquals(1, trimmed,
|
||||||
|
"trim counter should reflect only the non-spill body that was actually shortened");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void hardClearLeavesSpillMarkerUntouched() {
|
||||||
|
ConversationWindowManager mgr = new ConversationWindowManager(
|
||||||
|
new ConversationWindowProperties(), null, null);
|
||||||
|
|
||||||
|
List<Message> messages = new ArrayList<>(List.of(
|
||||||
|
toolMessage("call-spill", "web_search", SPILL_BODY),
|
||||||
|
toolMessage("call-big", "search", "y".repeat(2000))
|
||||||
|
));
|
||||||
|
|
||||||
|
int cleared = mgr.hardClearToolResults(messages);
|
||||||
|
|
||||||
|
ToolResponseMessage trm0 = (ToolResponseMessage) messages.get(0);
|
||||||
|
ToolResponseMessage trm1 = (ToolResponseMessage) messages.get(1);
|
||||||
|
assertEquals(SPILL_BODY, trm0.getResponses().getFirst().responseData(),
|
||||||
|
"Phase 2 hard clear must not replace a spill-marker body with [tool result removed]");
|
||||||
|
assertEquals("[tool result removed]", trm1.getResponses().getFirst().responseData(),
|
||||||
|
"non-spill bodies should still be replaced by Phase 2");
|
||||||
|
assertEquals(1, cleared,
|
||||||
|
"clear counter should reflect only the non-spill body that was actually replaced");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void prePruneForSummaryLeavesSpillMarkerUntouched() {
|
||||||
|
ConversationWindowManager mgr = new ConversationWindowManager(
|
||||||
|
new ConversationWindowProperties(), null, null);
|
||||||
|
|
||||||
|
List<Message> messages = new ArrayList<>(List.of(
|
||||||
|
toolMessage("call-spill", "web_search", SPILL_BODY),
|
||||||
|
toolMessage("call-big", "search", "z".repeat(2000))
|
||||||
|
));
|
||||||
|
|
||||||
|
int pruned = mgr.prePruneForSummary(messages);
|
||||||
|
|
||||||
|
ToolResponseMessage trm0 = (ToolResponseMessage) messages.get(0);
|
||||||
|
ToolResponseMessage trm1 = (ToolResponseMessage) messages.get(1);
|
||||||
|
assertEquals(SPILL_BODY, trm0.getResponses().getFirst().responseData(),
|
||||||
|
"Phase 3 pre-prune must not replace a spill-marker body with the cleared-output placeholder");
|
||||||
|
assertTrue(trm1.getResponses().getFirst().responseData().contains("旧工具输出已清理"),
|
||||||
|
"non-spill bodies should still be replaced by Phase 3");
|
||||||
|
assertEquals(1, pruned);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void mixedMessageWithSpillAndNonSpillResponsesPreservesOnlyTheMarker() {
|
||||||
|
// A single ToolResponseMessage can hold multiple ToolResponses (one
|
||||||
|
// assistant tool_calls turn could ask for several tools at once).
|
||||||
|
// The phase guards must operate at the response level, not the
|
||||||
|
// message level — the spill response stays, the non-spill response
|
||||||
|
// gets the placeholder.
|
||||||
|
ConversationWindowManager mgr = new ConversationWindowManager(
|
||||||
|
new ConversationWindowProperties(), null, null);
|
||||||
|
|
||||||
|
ToolResponseMessage mixed = ToolResponseMessage.builder().responses(List.of(
|
||||||
|
new ToolResponseMessage.ToolResponse("call-spill", "web_search", SPILL_BODY),
|
||||||
|
new ToolResponseMessage.ToolResponse("call-big", "search", "q".repeat(2000))
|
||||||
|
)).build();
|
||||||
|
List<Message> messages = new ArrayList<>(List.of(mixed));
|
||||||
|
|
||||||
|
mgr.hardClearToolResults(messages);
|
||||||
|
|
||||||
|
ToolResponseMessage trm = (ToolResponseMessage) messages.getFirst();
|
||||||
|
assertEquals(SPILL_BODY, trm.getResponses().get(0).responseData(),
|
||||||
|
"the spill response in a mixed message must survive Phase 2");
|
||||||
|
assertEquals("[tool result removed]", trm.getResponses().get(1).responseData(),
|
||||||
|
"the non-spill response in a mixed message must still be cleared");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void smallSpillMarkerStillStaysVerbatim() {
|
||||||
|
// Edge case: even when the preview is short (under the 500-char
|
||||||
|
// soft-trim threshold), the marker check should still apply. This
|
||||||
|
// protects against future changes to the trim threshold.
|
||||||
|
ConversationWindowManager mgr = new ConversationWindowManager(
|
||||||
|
new ConversationWindowProperties(), null, null);
|
||||||
|
|
||||||
|
String tinySpill = ToolResultStorage.SPILL_MARKER_PREFIX
|
||||||
|
+ " tool=test full_chars=600 path=/tmp/t.txt\n[tiny]";
|
||||||
|
List<Message> messages = new ArrayList<>(List.of(
|
||||||
|
toolMessage("call-1", "test", tinySpill)
|
||||||
|
));
|
||||||
|
|
||||||
|
mgr.softTrimToolResults(messages);
|
||||||
|
mgr.hardClearToolResults(messages);
|
||||||
|
mgr.prePruneForSummary(messages);
|
||||||
|
|
||||||
|
ToolResponseMessage trm = (ToolResponseMessage) messages.getFirst();
|
||||||
|
assertEquals(tinySpill, trm.getResponses().getFirst().responseData(),
|
||||||
|
"the marker check is what protects the body — not the size of the preview");
|
||||||
|
}
|
||||||
|
|
||||||
|
// ------------------------------------------------------------------ helpers
|
||||||
|
|
||||||
|
private static ToolResponseMessage toolMessage(String id, String name, String data) {
|
||||||
|
return ToolResponseMessage.builder()
|
||||||
|
.responses(List.of(new ToolResponseMessage.ToolResponse(id, name, data)))
|
||||||
|
.build();
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -103,6 +103,20 @@ class LaneDExecutorAndConfigTest {
|
|||||||
"storageBaseDir should still default to empty string");
|
"storageBaseDir should still default to empty string");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
@DisplayName("retentionDays defaults to 0 so spill files outlive their conversation")
|
||||||
|
void retentionDaysDefaultsToZero() {
|
||||||
|
// The recoverability invariant: a summary or preview that cites
|
||||||
|
// a spill path must keep working for the whole life of the
|
||||||
|
// conversation. Time-based deletion is opt-in; operators with
|
||||||
|
// disk pressure can raise this value explicitly.
|
||||||
|
ToolResultProperties props = new ToolResultProperties();
|
||||||
|
assertEquals(0, props.getRetentionDays(),
|
||||||
|
"retentionDays must default to 0 — time-based purge is opt-in to preserve recoverability");
|
||||||
|
assertTrue(props.getCleanupCron() != null && !props.getCleanupCron().isBlank(),
|
||||||
|
"cleanupCron stays defined; it is a no-op while retentionDays=0");
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@DisplayName("Per-result threshold matches the executor inline hard cap so spill and truncate share one ladder")
|
@DisplayName("Per-result threshold matches the executor inline hard cap so spill and truncate share one ladder")
|
||||||
void thresholdMatchesExecutorHardCap() throws Exception {
|
void thresholdMatchesExecutorHardCap() throws Exception {
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user