From 51b5eceb7d44f622ebcb8f67c5376ea5a71d4e09 Mon Sep 17 00:00:00 2001 From: matevip Date: Wed, 13 May 2026 08:50:28 +0800 Subject: [PATCH] feat(context): preserve spill markers across compaction phases, persist enriched boundary, broadcast compact_status SSE (#110) --- .../context/ConversationWindowManager.java | 185 +++++++++++++++--- .../graph/executor/ToolResultProperties.java | 15 +- .../conversation/ConversationService.java | 55 +++++- .../src/main/resources/application.yml | 11 +- ...dowManagerSpillMarkerPreservationTest.java | 157 +++++++++++++++ .../executor/LaneDExecutorAndConfigTest.java | 14 ++ 6 files changed, 397 insertions(+), 40 deletions(-) create mode 100644 mateclaw-server/src/test/java/vip/mate/agent/context/ConversationWindowManagerSpillMarkerPreservationTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java b/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java index a50e8753..119b6615 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/context/ConversationWindowManager.java @@ -22,6 +22,7 @@ import vip.mate.workspace.conversation.ConversationService; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; /** @@ -131,6 +132,21 @@ public class ConversationWindowManager { 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" */ @@ -205,6 +221,8 @@ public class ConversationWindowManager { if (messages == null || messages.isEmpty()) { return messages; } + long spillsAtEntry = (toolResultStorage != null) ? toolResultStorage.getSpillCount() : 0L; + messages = pruneOldToolResultsForModelInput(messages, conversationId, workspaceBasePath); int effectiveMax = (maxInputTokens != null && maxInputTokens > 0) @@ -229,7 +247,7 @@ public class ConversationWindowManager { // 可用于历史的 token 预算 = max - system - currentMsg - tools - 安全余量 int reservedTokens = systemTokens + currentMsgTokens + toolsTokens + (int) (effectiveMax * 0.05); - // RFC-025 Change 1: reserve 硬封顶到 effectiveMax 的 50%。 + // 预留 reserve 硬封顶到 effectiveMax 的 50%。 // 小上下文模型（Ollama 16K、本地 8K）下，systemTokens + currentMsgTokens 很容易 // 接近或超过 effectiveMax，不封顶会让 historyBudget 变负数导致死循环压缩 // （压缩目标比压缩前还大 → 压缩后又触发压缩）。 @@ -244,7 +262,8 @@ public class ConversationWindowManager { // 尾部保护 token 预算：阈值的 20%（与 Hermes 一致） 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 extra) { + if (streamTracker == null || conversationId == null || conversationId.isEmpty()) { + return; + } + try { + Map 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 compactMessages(List messages, int historyBudget, 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） int headEnd = 0; // 头部保护：暂不保护（system prompt 已在外部计算） int tailStart = findTailBoundary(messages, headEnd, tailTokenBudget); if (tailStart <= headEnd) { log.debug("[ConversationWindow] 消息数不足以拆分，跳过压缩"); + broadcastCompactStatus(conversationId, "skipped", + Map.of("reason", "insufficient_messages")); return messages; } @@ -281,8 +325,14 @@ public class ConversationWindowManager { // turn. int pairSafeCut = enforcePairSafeBoundary(messages, headEnd, tailStart); if (pairSafeCut <= headEnd) { + broadcastCompactStatus(conversationId, "skipped", + Map.of("reason", "pair_boundary_collapsed")); return messages; } + if (pairSafeCut != tailStart) { + broadcastCompactStatus(conversationId, "pair_safe", Map.of( + "movedFrom", tailStart, "movedTo", pairSafeCut)); + } tailStart = pairSafeCut; List oldMessages = new ArrayList<>(messages.subList(headEnd, tailStart)); @@ -340,13 +390,20 @@ public class ConversationWindowManager { // 计算动态摘要预算 int summaryBudget = computeSummaryBudget(forSummary); + broadcastCompactStatus(conversationId, "summarize", Map.of( + "messagesToSummarize", oldMessages.size(), + "summaryBudget", summaryBudget + )); + // 检查缓存 String cacheKey = conversationId + ":" + oldMessages.size(); CachedSummary cached = summaryCache.get(cacheKey); String summary; + boolean fromCache = false; if (cached != null && !cached.isExpired(CACHE_TTL_MS)) { summary = cached.summary(); + fromCache = true; log.debug("[ConversationWindow] 命中摘要缓存, conv={}", conversationId); } else { summary = generateSummary(forSummary, chatModel, conversationId, summaryBudget, memoryExtraContext); @@ -355,21 +412,12 @@ public class ConversationWindowManager { int count = compressionCounts.merge(conversationId, 1, Integer::sum); log.info("[ConversationWindow] 生成结构化摘要 ({} 字符, 第 {} 次压缩), 压缩 {} 条旧消息, conv={}", 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 result = new ArrayList<>(); + boolean anchored = false; if (summary != null && !summary.isBlank()) { result.add(new UserMessage(SUMMARY_PREFIX + summary)); @@ -380,11 +428,16 @@ public class ConversationWindowManager { Message anchor = buildFirstUserAnchor(oldMessages); if (anchor != null) { result.add(anchor); + anchored = true; } } else if (!oldMessages.isEmpty()) { log.warn("[ConversationWindow] 摘要生成失败，降级为保留最近 4 条旧消息, conv={}", conversationId); int fallbackKeep = Math.min(4, 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); @@ -393,6 +446,42 @@ public class ConversationWindowManager { if (resultTokens > historyBudget && result.size() > 2) { log.warn("[ConversationWindow] 压缩后仍超预算: {} > {}, 执行二次裁剪", resultTokens, 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 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; @@ -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 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 字符）。 + *

Spill-marker responses are left untouched so their on-disk pointer + * survives intact across compaction. + */ + int softTrimToolResults(List messages) { int trimmed = 0; for (int i = 0; i < messages.size(); i++) { if (messages.get(i) instanceof ToolResponseMessage trm) { List newResponses = new ArrayList<>(); boolean changed = false; 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(); if (data != null && data.length() > 500) { String head = data.substring(0, 200); @@ -811,16 +921,28 @@ public class ConversationWindowManager { /** * Phase 2 - Hard clear：将所有旧工具结果替换为占位符。 + *

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 messages) { + int hardClearToolResults(List messages) { int cleared = 0; for (int i = 0; i < messages.size(); i++) { if (messages.get(i) instanceof ToolResponseMessage trm) { - List placeholders = trm.getResponses().stream() - .map(r -> new ToolResponseMessage.ToolResponse(r.id(), r.name(), "[tool result removed]")) - .toList(); - messages.set(i, ToolResponseMessage.builder().responses(placeholders).build()); - cleared++; + boolean changed = false; + List replaced = new ArrayList<>(trm.getResponses().size()); + for (ToolResponseMessage.ToolResponse r : trm.getResponses()) { + if (isSpillMarker(r)) { + 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; @@ -828,18 +950,27 @@ public class ConversationWindowManager { /** * Phase 3 Pre-prune：在 LLM 摘要前，将工具输出替换为占位符（减少摘要输入 token）。 + *

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 messages) { + int prePruneForSummary(List messages) { int pruned = 0; for (int i = 0; i < messages.size(); i++) { if (messages.get(i) instanceof ToolResponseMessage trm) { 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) { - List placeholders = trm.getResponses().stream() - .map(r -> new ToolResponseMessage.ToolResponse(r.id(), r.name(), - "[旧工具输出已清理以节省上下文空间]")) - .toList(); + List placeholders = new ArrayList<>(trm.getResponses().size()); + for (ToolResponseMessage.ToolResponse r : trm.getResponses()) { + if (isSpillMarker(r)) { + placeholders.add(r); + continue; + } + placeholders.add(new ToolResponseMessage.ToolResponse(r.id(), r.name(), + "[旧工具输出已清理以节省上下文空间]")); + } messages.set(i, ToolResponseMessage.builder().responses(placeholders).build()); pruned++; } diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java index 15fdf0a8..e53c5fac 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultProperties.java @@ -99,12 +99,17 @@ public class ToolResultProperties { /** * Days to retain spill files before the scheduled cleanup deletes them. - * Spill files exist to let the model recover full tool output via - * {@code read_file} during the active conversation; once the conversation - * is dormant for this many days the agent is extremely unlikely to ever - * read the file again, and disk pressure starts to matter. + *

Default 0 means time-based cleanup is disabled — spill files + * stay on disk until the owning conversation is explicitly deleted (which + * fires {@code purgeConversation} via {@code ConversationService}). + * 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. + *

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 diff --git a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java index a93a05a1..1d5d1895 100644 --- a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java +++ b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java @@ -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 what happened 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. + * + *

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) { + 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: + *

    + *
  • {@code trigger} — what fired this boundary + * ({@code token_threshold}, {@code user_compact}, etc.)
    • + *
  • {@code preTokens} / {@code postTokens} — context size before + * and after, for the in-prompt status row
    • + *
  • {@code messagesSummarized} / {@code tailKept} — partition + * counts the user sees in the boundary card
    • + *
  • {@code toolResultsSpilled} — how many bodies the spill store + * absorbed during this boundary
    • + *
  • {@code summaryId} — stable id (the inserted message id) for + * deep-linking from the SSE event
    • + *
+ *

{@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 extraMetadata) { MessageEntity entity = new MessageEntity(); entity.setConversationId(conversationId); entity.setRole("system"); entity.setContent(summary); entity.setStatus("completed"); - entity.setMetadata("{\"type\":\"compression_summary\",\"compressedCount\":" + compressedCount + "}"); + + Map 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); - 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 listMessageViews(String conversationId) { diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index 34debce0..77b1c9eb 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -229,10 +229,13 @@ mate: excluded-tools: - read_file - read_workspace_memory_file - # Spill files are deleted after this many days. Set to 0 to disable the - # scheduled sweep entirely (files still get purged when the conversation - # is deleted explicitly via ConversationService.deleteConversation). - retention-days: 7 + # Spill files are deleted after this many days. Default 0 disables the + # scheduled sweep entirely so a summary/preview that points at a spill + # path stays valid for the whole life of the conversation. Files are + # 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 * * ?" conversation: window: diff --git a/mateclaw-server/src/test/java/vip/mate/agent/context/ConversationWindowManagerSpillMarkerPreservationTest.java b/mateclaw-server/src/test/java/vip/mate/agent/context/ConversationWindowManagerSpillMarkerPreservationTest.java new file mode 100644 index 00000000..b6112f82 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/context/ConversationWindowManagerSpillMarkerPreservationTest.java @@ -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. + * + *

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 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 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 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 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 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(); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/LaneDExecutorAndConfigTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/LaneDExecutorAndConfigTest.java index a13047d9..29ba7ed1 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/LaneDExecutorAndConfigTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/LaneDExecutorAndConfigTest.java @@ -103,6 +103,20 @@ class LaneDExecutorAndConfigTest { "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 @DisplayName("Per-result threshold matches the executor inline hard cap so spill and truncate share one ladder") void thresholdMatchesExecutorHardCap() throws Exception {