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 41bb473d..15fdf0a8 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 @@ -97,6 +97,23 @@ public class ToolResultProperties { */ private List excludedTools = List.of("read_file", "read_workspace_memory_file"); + /** + * 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. + */ + private int retentionDays = 7; + + /** + * Cron expression for the spill-cleanup task. Defaults to once a day at + * 03:00 server-local time so cleanup runs during quiet hours. Set this + * to a Spring-recognised value (six-field cron) or change the bean + * wiring to disable it entirely. + */ + private String cleanupCron = "0 0 3 * * ?"; + public boolean isEnabled() { return enabled; } public void setEnabled(boolean enabled) { this.enabled = enabled; } @@ -130,6 +147,14 @@ public class ToolResultProperties { this.excludedTools = excludedTools == null ? List.of() : excludedTools; } + public int getRetentionDays() { return retentionDays; } + public void setRetentionDays(int retentionDays) { this.retentionDays = retentionDays; } + + public String getCleanupCron() { return cleanupCron; } + public void setCleanupCron(String cleanupCron) { + this.cleanupCron = cleanupCron == null ? "" : cleanupCron; + } + /** O(1) membership test for the exclusion list, used on every tool result. */ public Set excludedToolsSet() { return Set.copyOf(excludedTools); diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultRetentionScheduler.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultRetentionScheduler.java new file mode 100644 index 00000000..fab5ef88 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultRetentionScheduler.java @@ -0,0 +1,54 @@ +package vip.mate.agent.graph.executor; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +/** + * Drives {@link ToolResultStorage#cleanupExpired()} on a cron schedule so + * spill files don't accumulate forever. Kept in its own class instead of + * inlined into {@link ToolResultStorage} for two reasons: + * + * + * + *

The cron expression comes from + * {@link ToolResultProperties#getCleanupCron()} (default {@code 0 0 3 * * ?}, + * i.e. once a day at 03:00 server-local time). The retention horizon comes + * from {@link ToolResultProperties#getRetentionDays()}. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class ToolResultRetentionScheduler { + + private final ToolResultStorage storage; + private final ToolResultProperties props; + + /** + * Cron-fired hook. Failures are logged at WARN so they show up in + * standard log scrapes without aborting the scheduler thread — losing + * a single sweep is fine, the next one will catch the same files. + */ + @Scheduled(cron = "${mate.agent.tool-result.cleanup-cron:0 0 3 * * ?}") + public void cleanup() { + if (props.getRetentionDays() <= 0) { + log.debug("[ToolResultRetentionScheduler] retentionDays<=0, skipping sweep"); + return; + } + try { + int deleted = storage.cleanupExpired(); + if (deleted > 0) { + log.info("[ToolResultRetentionScheduler] sweep deleted {} spill file(s) older than {} days", + deleted, props.getRetentionDays()); + } + } catch (Exception e) { + log.warn("[ToolResultRetentionScheduler] sweep failed: {}", e.getMessage(), e); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java index 558d43bb..88aeb301 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java @@ -58,6 +58,15 @@ public class ToolResultStorage { /** D-6: monotonically increasing spill counter for observability. */ private final java.util.concurrent.atomic.AtomicLong spillCount = new java.util.concurrent.atomic.AtomicLong(); + /** + * Workspace roots observed during this JVM's lifetime. Populated every + * time a successful spill resolves a base directory; consulted by the + * scheduled retention sweep and by {@link #purgeConversation} so we + * don't have to query the database for every workspace path. Cross-JVM + * orphans are not covered — that is documented in the cleanup javadoc. + */ + private final java.util.Set observedRoots = java.util.concurrent.ConcurrentHashMap.newKeySet(); + public ToolResultStorage(ToolResultProperties props) { this.props = props; this.excludedToolsSnapshot = props.excludedToolsSet(); @@ -270,15 +279,162 @@ public class ToolResultStorage { } private Path resolveBaseDir(String workspaceBasePath) { + Path base; if (!props.getStorageBaseDir().isEmpty()) { - return Paths.get(props.getStorageBaseDir()); + base = Paths.get(props.getStorageBaseDir()); + } else if (workspaceBasePath != null && !workspaceBasePath.isBlank()) { + base = Paths.get(workspaceBasePath, ".mateclaw", "tool-results"); + } else { + String tmp = System.getProperty("java.io.tmpdir"); + if (tmp == null || tmp.isEmpty()) return null; + base = Paths.get(tmp, "mateclaw", "tool-results"); } - if (workspaceBasePath != null && !workspaceBasePath.isBlank()) { - return Paths.get(workspaceBasePath, ".mateclaw", "tool-results"); + // Register so the retention sweep and conversation-delete hook can + // reach this root even when the workspace path is no longer in scope. + observedRoots.add(base); + return base; + } + + /** + * Roots currently known to this instance. Exposed package-private so the + * scheduled retention sweep and unit tests can enumerate them without + * touching the underlying set directly. + */ + java.util.Set getObservedRoots() { + return java.util.Collections.unmodifiableSet(observedRoots); + } + + /** + * Best-effort: delete every spill file and per-conversation directory + * older than {@link ToolResultProperties#getRetentionDays()} across all + * roots this storage has seen, plus the configured base dir and the + * tmpdir fallback. Returns the number of files deleted. + * + *

Workspaces that never received a spill in this JVM's lifetime are + * not covered. Persisting an observed-roots registry across restarts + * could fix that, but is intentionally out of scope — the operator-side + * remedy is to run a one-off cleanup with {@code storage-base-dir} + * pointed at the historical workspace. + */ + public int cleanupExpired() { + if (props.getRetentionDays() <= 0) { + return 0; + } + long cutoffEpochMillis = System.currentTimeMillis() + - (long) props.getRetentionDays() * 24L * 60L * 60L * 1000L; + + java.util.Set roots = new java.util.LinkedHashSet<>(observedRoots); + if (!props.getStorageBaseDir().isEmpty()) { + roots.add(Paths.get(props.getStorageBaseDir())); } String tmp = System.getProperty("java.io.tmpdir"); - if (tmp == null || tmp.isEmpty()) return null; - return Paths.get(tmp, "mateclaw", "tool-results"); + if (tmp != null && !tmp.isEmpty()) { + roots.add(Paths.get(tmp, "mateclaw", "tool-results")); + } + + int deleted = 0; + for (Path root : roots) { + deleted += deleteExpiredUnder(root, cutoffEpochMillis); + } + if (deleted > 0) { + log.info("[ToolResultStorage] cleanup: {} spill files removed across {} root(s)", + deleted, roots.size()); + } + return deleted; + } + + private int deleteExpiredUnder(Path root, long cutoffEpochMillis) { + if (root == null || !java.nio.file.Files.isDirectory(root)) { + return 0; + } + int deleted = 0; + try (java.util.stream.Stream stream = java.nio.file.Files.walk(root, 2)) { + for (Path p : (Iterable) stream::iterator) { + if (p.equals(root)) continue; + if (!java.nio.file.Files.isRegularFile(p)) continue; + try { + long mtime = java.nio.file.Files.getLastModifiedTime(p).toMillis(); + if (mtime < cutoffEpochMillis) { + java.nio.file.Files.deleteIfExists(p); + deleted++; + } + } catch (java.io.IOException ioe) { + log.warn("[ToolResultStorage] failed to inspect spill file {}: {}", p, ioe.getMessage()); + } + } + } catch (java.io.IOException ioe) { + log.warn("[ToolResultStorage] cleanup walk failed under {}: {}", root, ioe.getMessage()); + return deleted; + } + // Best-effort: remove emptied per-conversation directories. + try (java.util.stream.Stream stream = java.nio.file.Files.list(root)) { + for (Path child : (Iterable) stream::iterator) { + if (!java.nio.file.Files.isDirectory(child)) continue; + try (java.util.stream.Stream kids = java.nio.file.Files.list(child)) { + if (kids.findAny().isEmpty()) { + java.nio.file.Files.deleteIfExists(child); + } + } catch (java.io.IOException ignored) { + // empty-check failure is not fatal — leave the directory alone + } + } + } catch (java.io.IOException ioe) { + log.warn("[ToolResultStorage] empty-dir sweep failed under {}: {}", root, ioe.getMessage()); + } + return deleted; + } + + /** + * Delete every spill file produced for {@code conversationId} across + * all observed roots, plus the configured base and tmpdir fallback. + * Called by {@code ConversationService.deleteConversation} so spill + * directories don't outlive the conversation that owns them. + * + *

Silently no-ops when nothing matches — a conversation that never + * spilled, or one whose workspace root was never observed in this JVM, + * is simply left alone. Returns the number of files deleted. + */ + public int purgeConversation(String conversationId) { + if (conversationId == null || conversationId.isEmpty()) { + return 0; + } + String safeConv = sanitize(conversationId); + java.util.Set roots = new java.util.LinkedHashSet<>(observedRoots); + if (!props.getStorageBaseDir().isEmpty()) { + roots.add(Paths.get(props.getStorageBaseDir())); + } + String tmp = System.getProperty("java.io.tmpdir"); + if (tmp != null && !tmp.isEmpty()) { + roots.add(Paths.get(tmp, "mateclaw", "tool-results")); + } + int deleted = 0; + for (Path root : roots) { + Path convDir = root.resolve(safeConv); + if (!java.nio.file.Files.isDirectory(convDir)) continue; + try (java.util.stream.Stream stream = java.nio.file.Files.list(convDir)) { + for (Path p : (Iterable) stream::iterator) { + try { + if (java.nio.file.Files.isRegularFile(p)) { + java.nio.file.Files.deleteIfExists(p); + deleted++; + } + } catch (java.io.IOException ioe) { + log.warn("[ToolResultStorage] failed to delete spill file {}: {}", p, ioe.getMessage()); + } + } + } catch (java.io.IOException ioe) { + log.warn("[ToolResultStorage] purge walk failed under {}: {}", convDir, ioe.getMessage()); + } + try { + java.nio.file.Files.deleteIfExists(convDir); + } catch (java.io.IOException ignored) { + // non-empty after deletes (another writer raced us) — fine, leave it + } + } + if (deleted > 0) { + log.info("[ToolResultStorage] purged {} spill file(s) for conversation {}", deleted, conversationId); + } + return deleted; } /** Strip path separators and reserved characters so user-supplied IDs cannot escape the directory. */ 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 6de391be..a93a05a1 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 @@ -65,6 +65,19 @@ public class ConversationService { private final ChannelSessionMapper channelSessionMapper; private final ApplicationEventPublisher eventPublisher; + /** + * Optional spill store. Injected via a setter so the existing @RequiredArgsConstructor + * stays stable and tests that build the service directly don't need to wire + * tool-result storage. When present, deleteConversation also purges any spill + * files this conversation produced so they don't outlive the row that owned them. + */ + private vip.mate.agent.graph.executor.ToolResultStorage toolResultStorage; + + @org.springframework.beans.factory.annotation.Autowired(required = false) + public void setToolResultStorage(vip.mate.agent.graph.executor.ToolResultStorage toolResultStorage) { + this.toolResultStorage = toolResultStorage; + } + /** * 获取用户的会话列表（返回 VO，包含 agentName/agentIcon/status） */ @@ -501,15 +514,34 @@ public class ConversationService { @Override public void afterCommit() { cleanAttachmentFiles(conversationId); + purgeToolResultSpill(conversationId); eventPublisher.publishEvent(new ConversationDeletedEvent(conversationId)); } }); } else { cleanAttachmentFiles(conversationId); + purgeToolResultSpill(conversationId); eventPublisher.publishEvent(new ConversationDeletedEvent(conversationId)); } } + /** + * Best-effort: ask the spill store to delete every tool-result file this + * conversation produced. No-op when no spill store is wired in (legacy + * deployments or tests that don't need spill). Failures are logged but + * never propagated — leaving an extra file on disk is a small price + * compared to surfacing IO errors as a 500 on the delete endpoint. + */ + private void purgeToolResultSpill(String conversationId) { + if (toolResultStorage == null) return; + try { + toolResultStorage.purgeConversation(conversationId); + } catch (Exception e) { + log.warn("[Conversation] tool-result spill purge failed for {}: {}", + conversationId, e.getMessage()); + } + } + /** * 清空会话消息（同时清理附件文件） */ diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index 73da99a1..34debce0 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -229,6 +229,11 @@ 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 + cleanup-cron: "0 0 3 * * ?" conversation: window: # 测试时临时调低：2000 token ≈ 2000 中文字，3 轮对话即可触发压缩 diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/ToolResultStorageRetentionTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/ToolResultStorageRetentionTest.java new file mode 100644 index 00000000..d2132e97 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/executor/ToolResultStorageRetentionTest.java @@ -0,0 +1,161 @@ +package vip.mate.agent.graph.executor; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Retention sweep + per-conversation purge for {@link ToolResultStorage}. + * + *

The store keeps a per-JVM "observed roots" registry — every time a + * spill resolves a directory, that directory is remembered so the + * retention sweep can reach it even after the workspace path has gone + * out of scope. These tests verify the registry behaviour, the + * mtime-based deletion contract, and the targeted per-conversation purge + * called from {@code ConversationService.deleteConversation}. + */ +class ToolResultStorageRetentionTest { + + @Test + void successfulSpillRegistersTheRoot(@TempDir Path tempDir) { + ToolResultStorage storage = newStorage(tempDir, /*threshold*/ 100, /*retention*/ 7); + + // Trigger a spill so resolveBaseDir() is invoked. + String out = storage.persistIfOversized( + "x".repeat(500), "web_search", "call-1", "conv-A", tempDir.toString()); + assertTrue(out.startsWith(ToolResultStorage.SPILL_MARKER_PREFIX)); + + assertTrue(storage.getObservedRoots().contains(tempDir), + "successful spill must register its resolved root for later cleanup"); + } + + @Test + void cleanupDeletesFilesOlderThanRetention(@TempDir Path tempDir) throws Exception { + // retention = 1 day + ToolResultStorage storage = newStorage(tempDir, /*threshold*/ 100, /*retention*/ 1); + + // Drop a "fresh" spill via the public API. + String fresh = storage.persistIfOversized( + "fresh".repeat(200), "web_search", "call-fresh", "conv-A", tempDir.toString()); + assertTrue(fresh.startsWith(ToolResultStorage.SPILL_MARKER_PREFIX)); + Path freshFile = pathFromPreview(fresh); + + // Drop a "stale" spill and back-date its mtime by 8 days. + String stale = storage.persistIfOversized( + "stale".repeat(200), "web_search", "call-stale", "conv-B", tempDir.toString()); + Path staleFile = pathFromPreview(stale); + Files.setLastModifiedTime(staleFile, + FileTime.from(Instant.now().minus(8, ChronoUnit.DAYS))); + + int deleted = storage.cleanupExpired(); + + assertEquals(1, deleted, "only the stale file should be removed"); + assertTrue(Files.exists(freshFile), "fresh file must survive"); + assertFalse(Files.exists(staleFile), "stale file must be deleted"); + } + + @Test + void cleanupIsNoOpWhenRetentionDisabled(@TempDir Path tempDir) throws Exception { + ToolResultStorage storage = newStorage(tempDir, /*threshold*/ 100, /*retention*/ 0); + + String stale = storage.persistIfOversized( + "stale".repeat(200), "web_search", "call-1", "conv-A", tempDir.toString()); + Path staleFile = pathFromPreview(stale); + Files.setLastModifiedTime(staleFile, + FileTime.from(Instant.now().minus(100, ChronoUnit.DAYS))); + + int deleted = storage.cleanupExpired(); + assertEquals(0, deleted, "retentionDays<=0 must disable the sweep entirely"); + assertTrue(Files.exists(staleFile), "stale file must remain when sweep is disabled"); + } + + @Test + void cleanupRemovesEmptiedConversationDirectories(@TempDir Path tempDir) throws Exception { + ToolResultStorage storage = newStorage(tempDir, /*threshold*/ 100, /*retention*/ 1); + + String stale = storage.persistIfOversized( + "stale".repeat(200), "web_search", "call-stale", "conv-old", tempDir.toString()); + Path staleFile = pathFromPreview(stale); + Path staleDir = staleFile.getParent(); + Files.setLastModifiedTime(staleFile, + FileTime.from(Instant.now().minus(8, ChronoUnit.DAYS))); + + storage.cleanupExpired(); + + assertFalse(Files.exists(staleFile)); + assertFalse(Files.exists(staleDir), + "the empty conv-old/ directory should be cleaned up too"); + } + + @Test + void purgeConversationDeletesAllFilesForOneConversation(@TempDir Path tempDir) throws Exception { + ToolResultStorage storage = newStorage(tempDir, /*threshold*/ 100, /*retention*/ 30); + + // Two spills for conv-A, one spill for conv-B. + String a1 = storage.persistIfOversized( + "a1".repeat(200), "web_search", "call-a1", "conv-A", tempDir.toString()); + String a2 = storage.persistIfOversized( + "a2".repeat(200), "web_search", "call-a2", "conv-A", tempDir.toString()); + String b1 = storage.persistIfOversized( + "b1".repeat(200), "web_search", "call-b1", "conv-B", tempDir.toString()); + Path af1 = pathFromPreview(a1); + Path af2 = pathFromPreview(a2); + Path bf1 = pathFromPreview(b1); + + int deleted = storage.purgeConversation("conv-A"); + + assertEquals(2, deleted, "both A files should be deleted"); + assertFalse(Files.exists(af1)); + assertFalse(Files.exists(af2)); + assertTrue(Files.exists(bf1), "conv-B files must not be touched by a conv-A purge"); + } + + @Test + void purgeConversationIsSilentForUnknownConversation(@TempDir Path tempDir) { + ToolResultStorage storage = newStorage(tempDir, /*threshold*/ 100, /*retention*/ 7); + + // No spill at all → nothing to purge → 0, no exception. + int deleted = storage.purgeConversation("never-existed"); + assertEquals(0, deleted); + } + + @Test + void purgeConversationHandlesBlankIdSafely(@TempDir Path tempDir) { + ToolResultStorage storage = newStorage(tempDir, /*threshold*/ 100, /*retention*/ 7); + + assertEquals(0, storage.purgeConversation(null)); + assertEquals(0, storage.purgeConversation("")); + } + + // ------------------------------------------------------------------ helpers + + private static ToolResultStorage newStorage(Path tempDir, int threshold, int retentionDays) { + ToolResultProperties props = new ToolResultProperties(); + props.setStorageBaseDir(tempDir.toString()); + props.setPerResultThresholdChars(threshold); + props.setPreviewHeadChars(80); + props.setRetentionDays(retentionDays); + props.setExcludedTools(List.of()); + return new ToolResultStorage(props); + } + + private static Path pathFromPreview(String preview) { + java.util.regex.Matcher m = java.util.regex.Pattern.compile("path=(\\S+)").matcher(preview); + assertTrue(m.find(), "preview must include path=..."); + Path p = Path.of(m.group(1)); + assertNotNull(p); + return p; + } +}