From bfc54175089d259ef2855feccd9d9e6a3dbc6d3e Mon Sep 17 00:00:00 2001 From: mateaix <7333791@qq.com> Date: Sat, 12 Sep 2026 13:03:39 +0800 Subject: [PATCH] feat(memory): harden recall and provider resilience --- .../api/memory/PluginMemoryProvider.java | 7 +- .../java/vip/mate/plugin/mem0/Mem0Config.java | 11 +- .../vip/mate/plugin/mem0/Mem0Exception.java | 5 +- .../java/vip/mate/plugin/mem0/Mem0Plugin.java | 17 +- .../vip/mate/plugin/mem0/Mem0Provider.java | 106 +++++++----- .../src/main/resources/mateclaw-plugin.json | 6 + .../vip/mate/plugin/mem0/Mem0ConfigTest.java | 6 + .../vip/mate/plugin/mem0/Mem0PluginTest.java | 2 + .../mate/plugin/mem0/Mem0ProviderTest.java | 48 +++++- .../vip/mate/memory/MemoryProperties.java | 12 ++ .../PostConversationMemoryListener.java | 13 +- .../mate/memory/model/MemoryRecallEntity.java | 2 +- .../mate/memory/nudge/MemoryNudgeService.java | 22 ++- .../memory/service/MemoryRecallService.java | 159 +++++++++--------- .../service/MemorySummarizationGate.java | 16 +- .../service/MemorySummarizationService.java | 93 ++++++---- .../vip/mate/memory/spi/MemoryManager.java | 134 ++++++++++++++- .../vip/mate/memory/spi/MemoryProvider.java | 7 +- .../decorator/MemoryProviderDecorator.java | 1 + .../decorator/RetryableMemoryProvider.java | 4 +- .../plugin/bridge/PluginMemoryBridge.java | 5 + .../src/main/resources/application.yml | 4 + .../V192__memory_recall_unique_identity.sql | 39 +++++ .../V192__memory_recall_unique_identity.sql | 39 +++++ .../V192__memory_recall_unique_identity.sql | 35 ++++ .../src/main/resources/docs/en/memory.md | 11 +- .../src/main/resources/docs/zh/memory.md | 11 +- .../memory/MemoryManagerResilienceTest.java | 116 +++++++++++++ .../memory/nudge/MemoryNudgeCooldownTest.java | 64 +++++++ .../service/MemoryRecallMigrationTest.java | 89 ++++++++++ .../MemoryRecallOwnerIsolationTest.java | 39 ++++- .../MemorySummarizationCooldownTest.java | 88 ++++++++++ .../service/MemorySummarizationGateTest.java | 2 + .../plugin/bridge/PluginMemoryBridgeTest.java | 15 ++ 34 files changed, 1025 insertions(+), 203 deletions(-) create mode 100644 mateclaw-server/src/main/resources/db/migration/h2/V192__memory_recall_unique_identity.sql create mode 100644 mateclaw-server/src/main/resources/db/migration/kingbase/V192__memory_recall_unique_identity.sql create mode 100644 mateclaw-server/src/main/resources/db/migration/mysql/V192__memory_recall_unique_identity.sql create mode 100644 mateclaw-server/src/test/java/vip/mate/memory/MemoryManagerResilienceTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/memory/nudge/MemoryNudgeCooldownTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/memory/service/MemoryRecallMigrationTest.java create mode 100644 mateclaw-server/src/test/java/vip/mate/memory/service/MemorySummarizationCooldownTest.java diff --git a/mateclaw-plugin-api/src/main/java/vip/mate/plugin/api/memory/PluginMemoryProvider.java b/mateclaw-plugin-api/src/main/java/vip/mate/plugin/api/memory/PluginMemoryProvider.java index b4b47833..7dd82d06 100644 --- a/mateclaw-plugin-api/src/main/java/vip/mate/plugin/api/memory/PluginMemoryProvider.java +++ b/mateclaw-plugin-api/src/main/java/vip/mate/plugin/api/memory/PluginMemoryProvider.java @@ -11,7 +11,7 @@ import java.util.List; * * @author MateClaw Team */ -public interface PluginMemoryProvider { +public interface PluginMemoryProvider extends AutoCloseable { /** * Unique provider identifier, e.g. "vector_memory", "graph_memory". @@ -114,4 +114,9 @@ public interface PluginMemoryProvider { */ default void onSessionEnd(Long agentId, String conversationId) { } + + /** Release provider-owned resources when the plugin is unloaded. */ + @Override + default void close() { + } } diff --git a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Config.java b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Config.java index ea7ae20e..1a424431 100644 --- a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Config.java +++ b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Config.java @@ -13,6 +13,7 @@ package vip.mate.plugin.mem0; * @param syncEnabled whether syncTurn should POST to Mem0 /memories/ * @param maxResults cap on memories returned per recall * @param timeoutMs HTTP timeout for both recall and sync + * @param syncQueueCapacity maximum number of turns waiting for asynchronous sync * @author MateClaw Team */ record Mem0Config( @@ -21,10 +22,18 @@ record Mem0Config( boolean searchEnabled, boolean syncEnabled, int maxResults, - int timeoutMs + int timeoutMs, + int syncQueueCapacity ) { static final int DEFAULT_MAX_RESULTS = 5; static final int DEFAULT_TIMEOUT_MS = 3000; + static final int DEFAULT_SYNC_QUEUE_CAPACITY = 256; + + Mem0Config(String baseUrl, String apiKey, boolean searchEnabled, boolean syncEnabled, + int maxResults, int timeoutMs) { + this(baseUrl, apiKey, searchEnabled, syncEnabled, maxResults, timeoutMs, + DEFAULT_SYNC_QUEUE_CAPACITY); + } /** * Whether this provider should participate at all. diff --git a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Exception.java b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Exception.java index 064fffa3..5ba2c875 100644 --- a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Exception.java +++ b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Exception.java @@ -3,9 +3,8 @@ package vip.mate.plugin.mem0; /** * Raised when a Mem0 REST call fails (non-2xx response, IO error, timeout). *

- * Caught and logged by {@link Mem0Provider} so that Mem0 outages degrade - * gracefully (empty recall / dropped sync) without affecting the agent's - * response path. + * Sync failures are caught by {@link Mem0Provider}; recall failures propagate + * to the platform provider boundary for timeout/circuit-breaker accounting. * * @author MateClaw Team */ diff --git a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Plugin.java b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Plugin.java index 62696403..d4f9283f 100644 --- a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Plugin.java +++ b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Plugin.java @@ -39,6 +39,7 @@ public class Mem0Plugin implements MateClawPlugin { private static final String CONFIG_SYNC_ENABLED = "syncEnabled"; private static final String CONFIG_MAX_RESULTS = "maxResults"; private static final String CONFIG_TIMEOUT_MS = "timeoutMs"; + private static final String CONFIG_SYNC_QUEUE_CAPACITY = "syncQueueCapacity"; private Logger log; @@ -54,11 +55,16 @@ public class Mem0Plugin implements MateClawPlugin { Mem0Client client = new Mem0Client(config); Mem0Provider provider = new Mem0Provider(config, client, log); - context.registerMemoryProvider(provider); + try { + context.registerMemoryProvider(provider); + } catch (RuntimeException e) { + provider.close(); + throw e; + } - log.info("Mem0 plugin loaded: baseUrl={}, searchEnabled={}, syncEnabled={}, maxResults={}, timeoutMs={}", + log.info("Mem0 plugin loaded: baseUrl={}, searchEnabled={}, syncEnabled={}, maxResults={}, timeoutMs={}, syncQueueCapacity={}", maskUrl(config.baseUrl()), config.searchEnabled(), config.syncEnabled(), - config.maxResults(), config.timeoutMs()); + config.maxResults(), config.timeoutMs(), config.syncQueueCapacity()); } @Override @@ -78,6 +84,7 @@ public class Mem0Plugin implements MateClawPlugin { Boolean syncEnabled = ctx.getConfig(CONFIG_SYNC_ENABLED, Boolean.class); Integer maxResults = ctx.getConfig(CONFIG_MAX_RESULTS, Integer.class); Integer timeoutMs = ctx.getConfig(CONFIG_TIMEOUT_MS, Integer.class); + Integer syncQueueCapacity = ctx.getConfig(CONFIG_SYNC_QUEUE_CAPACITY, Integer.class); return new Mem0Config( baseUrl, @@ -85,7 +92,9 @@ public class Mem0Plugin implements MateClawPlugin { searchEnabled == null ? true : searchEnabled, syncEnabled == null ? true : syncEnabled, maxResults == null ? Mem0Config.DEFAULT_MAX_RESULTS : maxResults, - timeoutMs == null ? Mem0Config.DEFAULT_TIMEOUT_MS : timeoutMs + timeoutMs == null ? Mem0Config.DEFAULT_TIMEOUT_MS : timeoutMs, + syncQueueCapacity == null ? Mem0Config.DEFAULT_SYNC_QUEUE_CAPACITY + : Math.max(1, syncQueueCapacity) ); } diff --git a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Provider.java b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Provider.java index fc42efd6..5c66f4dd 100644 --- a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Provider.java +++ b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Provider.java @@ -4,9 +4,11 @@ import org.slf4j.Logger; import vip.mate.plugin.api.memory.PluginMemoryProvider; import java.util.List; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.Executor; -import java.util.concurrent.Executors; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; /** * Memory provider that bridges MateClaw's per-turn lifecycle to a self-hosted @@ -17,13 +19,14 @@ import java.util.concurrent.Executors; *

  • {@code systemPromptBlock} — no-op (returns ""), aligns with SessionSearchProvider
    • *
  • {@code prefetch(agentId, query, ownerKey)} — when {@code searchEnabled} * and {@code ownerKey} is non-blank, calls {@code POST /memories/search/} - * and returns a {@code [Mem0 Recall]} block. Returns "" on any failure - * or when disabled.
    • + * and returns a {@code [Mem0 Recall]} block. Failures propagate to the + * platform's timeout/circuit-breaker boundary. *
  • {@code syncTurn(agentId, conversationId, messages, ownerKey)} — when * {@code syncEnabled} and {@code ownerKey} is non-blank, asynchronously * pushes the turn to {@code POST /memories/} under {@code user_id = * ownerKey}, the same identifier prefetch recalls by. Failures are - * logged and swallowed; never blocks the response path. The four-arg + * logged and swallowed; never blocks the response path. The bounded + * queue drops new writes when saturated. The four-arg * variant (no ownerKey) skips — writing under any other identifier * would produce memories that owner-scoped recall can never surface.
    • *
  • {@code getToolBeans} — empty (no agent-facing tools in v1)
    • @@ -34,8 +37,8 @@ import java.util.concurrent.Executors; * When {@code ownerKey} is null/blank, both recall and sync are skipped — Mem0 * requires {@code user_id}. * - *

    Asynchronous sync: a single-thread daemon executor is used - * so that bursts of turns don't pile up on the platform's request thread. + *

    Asynchronous sync: a single-thread daemon executor with a bounded queue + * prevents an unavailable Mem0 service from growing heap usage without limit. * * @author MateClaw Team */ @@ -46,21 +49,19 @@ class Mem0Provider implements PluginMemoryProvider { private final Mem0Config config; private final Mem0Client client; private final Logger log; - private final Executor async; + private final ThreadPoolExecutor async; + private final AtomicLong droppedSyncCount = new AtomicLong(); Mem0Provider(Mem0Config config, Mem0Client client, Logger log) { this.config = config; this.client = client; this.log = log; - // Single-thread executor is enough — syncTurn calls are sequential per - // agent and not latency-sensitive; the platform's request thread must - // not be blocked. A bounded single-thread queue keeps memory footprint - // predictable even under burst load. - this.async = Executors.newSingleThreadExecutor(r -> { - Thread t = new Thread(r, "mem0-sync"); - t.setDaemon(true); - return t; - }); + this.async = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(Math.max(1, config.syncQueueCapacity())), r -> { + Thread t = new Thread(r, "mem0-sync"); + t.setDaemon(true); + return t; + }, new ThreadPoolExecutor.AbortPolicy()); } @Override @@ -104,20 +105,12 @@ class Mem0Provider implements PluginMemoryProvider { if (userQuery == null || userQuery.isBlank()) { return ""; } - try { - List memories = client.searchMemories( - ownerKey, agentId == null ? null : agentId.toString(), userQuery); - if (memories.isEmpty()) { - return ""; - } - return formatRecallBlock(memories); - } catch (Exception e) { - // Fault isolation: log and return empty so the platform falls back - // to the other (local) providers without affecting the response. - log.warn("[Mem0] prefetch failed for agent={} owner={}: {}", - agentId, ownerKey, e.getMessage()); + List memories = client.searchMemories( + ownerKey, agentId == null ? null : agentId.toString(), userQuery); + if (memories.isEmpty()) { return ""; } + return formatRecallBlock(memories); } @Override @@ -143,15 +136,52 @@ class Mem0Provider implements PluginMemoryProvider { && (assistantReply == null || assistantReply.isBlank())) { return; } - CompletableFuture.runAsync(() -> { - try { - client.addMemories(ownerKey, agentId == null ? null : agentId.toString(), - conversationId, userMessage, assistantReply); - } catch (Exception e) { - log.debug("[Mem0] syncTurn failed for agent={} owner={}: {}", - agentId, ownerKey, e.getMessage()); + try { + async.execute(() -> { + try { + client.addMemories(ownerKey, agentId == null ? null : agentId.toString(), + conversationId, userMessage, assistantReply); + } catch (Exception e) { + log.debug("[Mem0] syncTurn failed for agent={} owner={}: {}", + agentId, ownerKey, e.getMessage()); + } + }); + } catch (RejectedExecutionException e) { + long dropped = droppedSyncCount.incrementAndGet(); + log.warn("[Mem0] sync queue full or provider closed; dropped turn for agent={} owner={} (totalDropped={})", + agentId, ownerKey, dropped); + } + } + + int queuedSyncCount() { + return async.getQueue().size(); + } + + long droppedSyncCount() { + return droppedSyncCount.get(); + } + + boolean isClosed() { + return async.isShutdown(); + } + + @Override + public void close() { + async.shutdown(); + List dropped = List.of(); + try { + long drainMs = Math.min(1000L, Math.max(100L, config.timeoutMs())); + if (!async.awaitTermination(drainMs, TimeUnit.MILLISECONDS)) { + dropped = async.shutdownNow(); } - }, async); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + dropped = async.shutdownNow(); + } + if (!dropped.isEmpty()) { + droppedSyncCount.addAndGet(dropped.size()); + log.warn("[Mem0] provider closed with {} queued sync turn(s) discarded", dropped.size()); + } } @Override diff --git a/mateclaw-plugin-mem0/src/main/resources/mateclaw-plugin.json b/mateclaw-plugin-mem0/src/main/resources/mateclaw-plugin.json index a0e974a9..6a4a3f98 100644 --- a/mateclaw-plugin-mem0/src/main/resources/mateclaw-plugin.json +++ b/mateclaw-plugin-mem0/src/main/resources/mateclaw-plugin.json @@ -43,6 +43,12 @@ "required": false, "secret": false, "description": "HTTP timeout in milliseconds for both recall and sync. Default 3000." + }, + "syncQueueCapacity": { + "type": "integer", + "required": false, + "secret": false, + "description": "Maximum pending asynchronous sync turns. New writes are dropped when full. Default 256." } } } diff --git a/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0ConfigTest.java b/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0ConfigTest.java index f1d0a279..f3eb92b4 100644 --- a/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0ConfigTest.java +++ b/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0ConfigTest.java @@ -35,4 +35,10 @@ class Mem0ConfigTest { Mem0Config c = new Mem0Config("http://localhost:8080", null, true, true, 5, 1000); assertThat(c.normalizedBaseUrl()).isEqualTo("http://localhost:8080"); } + + @Test + void legacyConstructorUsesBoundedQueueDefault() { + Mem0Config c = new Mem0Config("http://localhost:8080", null, true, true, 5, 1000); + assertThat(c.syncQueueCapacity()).isEqualTo(Mem0Config.DEFAULT_SYNC_QUEUE_CAPACITY); + } } diff --git a/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0PluginTest.java b/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0PluginTest.java index b4f0739a..cb13d959 100644 --- a/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0PluginTest.java +++ b/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0PluginTest.java @@ -90,6 +90,7 @@ class Mem0PluginTest { PluginContext ctx = new StubContext(config, registered) { @Override public void registerMemoryProvider(PluginMemoryProvider provider) { + registered.set(provider); throw new PluginException("Only one external memory provider allowed"); } }; @@ -98,6 +99,7 @@ class Mem0PluginTest { assertThatThrownBy(() -> plugin.onLoad(ctx)) .isInstanceOf(PluginException.class) .hasMessageContaining("Only one"); + assertThat(((Mem0Provider) registered.get()).isClosed()).isTrue(); } /** diff --git a/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0ProviderTest.java b/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0ProviderTest.java index 0a84d40d..37f0b2a4 100644 --- a/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0ProviderTest.java +++ b/mateclaw-plugin-mem0/src/test/java/vip/mate/plugin/mem0/Mem0ProviderTest.java @@ -14,8 +14,11 @@ import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; class Mem0ProviderTest { @@ -43,6 +46,7 @@ class Mem0ProviderTest { @AfterEach void tearDown() { + if (provider != null) provider.close(); if (server != null) server.stop(0); } @@ -129,16 +133,15 @@ class Mem0ProviderTest { } @Test - void threeArgPrefetch_returnsEmptyOnServerError() { - // Replace handler to fail; the provider should swallow and return "". + void threeArgPrefetch_propagatesServerErrorToPlatformCircuitBreaker() { server.removeContext("/"); server.createContext("/", ex -> { ex.sendResponseHeaders(500, 0); ex.close(); }); - String result = provider.prefetch(1L, "q", "user:42"); - assertThat(result).isEmpty(); + assertThatThrownBy(() -> provider.prefetch(1L, "q", "user:42")) + .isInstanceOf(Mem0Exception.class); } @Test @@ -215,5 +218,42 @@ class Mem0ProviderTest { Mem0Provider p = new Mem0Provider(cfg, new Mem0Client(cfg), LoggerFactory.getLogger("test")); assertThat(p.prefetch(1L, "q", "user:42")).isEmpty(); assertThat(searchCount.get()).isZero(); + p.close(); + } + + @Test + void syncQueueIsBoundedAndCloseReleasesExecutor() throws Exception { + CountDownLatch firstStarted = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + AtomicInteger writes = new AtomicInteger(); + Mem0Config cfg = new Mem0Config("http://localhost:8080", null, + false, true, 3, 3000, 1); + Mem0Client blockingClient = new Mem0Client(cfg) { + @Override + void addMemories(String userId, String agentId, String conversationId, + String userMessage, String assistantReply) { + writes.incrementAndGet(); + firstStarted.countDown(); + try { + releaseFirst.await(2, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }; + Mem0Provider bounded = new Mem0Provider(cfg, blockingClient, LoggerFactory.getLogger("test")); + try { + bounded.syncTurn(1L, "one", "u", "a", "user:1"); + assertThat(firstStarted.await(1, TimeUnit.SECONDS)).isTrue(); + bounded.syncTurn(1L, "two", "u", "a", "user:1"); + bounded.syncTurn(1L, "three", "u", "a", "user:1"); + + assertThat(bounded.queuedSyncCount()).isEqualTo(1); + assertThat(bounded.droppedSyncCount()).isEqualTo(1); + } finally { + releaseFirst.countDown(); + bounded.close(); + } + assertThat(bounded.isClosed()).isTrue(); } } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/MemoryProperties.java b/mateclaw-server/src/main/java/vip/mate/memory/MemoryProperties.java index b77f3fb3..757f7777 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/MemoryProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/MemoryProperties.java @@ -162,6 +162,18 @@ public class MemoryProperties { /** Enable provider metrics collection */ private boolean providerMetricsEnabled = false; + /** Maximum time allowed for a single provider prefetch; 0 = no per-provider limit. */ + private long providerPrefetchTimeoutMs = 1500; + + /** Maximum time allowed for the complete prefetch chain; 0 = no total limit. */ + private long providerPrefetchTotalBudgetMs = 2500; + + /** Consecutive prefetch failures before a provider circuit opens. */ + private int providerCircuitFailureThreshold = 3; + + /** Time an open provider circuit waits before allowing one probe request. */ + private long providerCircuitCooldownSeconds = 30; + // --- Phase 3: Fact projection --- /** Fact projection configuration */ diff --git a/mateclaw-server/src/main/java/vip/mate/memory/listener/PostConversationMemoryListener.java b/mateclaw-server/src/main/java/vip/mate/memory/listener/PostConversationMemoryListener.java index 2834d2fc..57cf3c10 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/listener/PostConversationMemoryListener.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/listener/PostConversationMemoryListener.java @@ -8,6 +8,7 @@ import org.springframework.stereotype.Component; import vip.mate.memory.MemoryProperties; import vip.mate.memory.event.ConversationCompletedEvent; import vip.mate.memory.nudge.MemoryNudgeService; +import vip.mate.memory.service.MemorySummarizationGate; import vip.mate.memory.service.MemorySummarizationService; /** @@ -38,13 +39,17 @@ public class PostConversationMemoryListener { return; } - // 消息数量不足 - if (event.messageCount() < properties.getMinMessagesForSummarize()) { + // Explicit "remember" requests are durable user intent and must not be + // dropped merely because this is the first turn in a conversation. + boolean explicitRemember = MemorySummarizationGate.isExplicitRememberRequest(event.userMessage()); + + // 消息数量不足（显式记忆请求除外） + if (!explicitRemember && event.messageCount() < properties.getMinMessagesForSummarize()) { return; } - // 用户消息太短 - if (event.userMessage() != null + // 用户消息太短（显式记忆请求除外） + if (!explicitRemember && event.userMessage() != null && event.userMessage().length() < properties.getMinUserMessageLength()) { return; } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/model/MemoryRecallEntity.java b/mateclaw-server/src/main/java/vip/mate/memory/model/MemoryRecallEntity.java index b956c133..00ba7aca 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/model/MemoryRecallEntity.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/model/MemoryRecallEntity.java @@ -56,7 +56,7 @@ public class MemoryRecallEntity { /** Last time this candidate was reviewed during a dream run */ private LocalDateTime lastReviewedAt; - /** Memory subject this recall belongs to (e.g. "user:42"); null for shared/legacy rows. */ + /** Memory subject this recall belongs to (e.g. "user:42"); empty for shared rows. */ private String ownerKey; /** Visibility scope: PERSONAL / TEAM / GLOBAL. Defaults to TEAM at the DB level. */ diff --git a/mateclaw-server/src/main/java/vip/mate/memory/nudge/MemoryNudgeService.java b/mateclaw-server/src/main/java/vip/mate/memory/nudge/MemoryNudgeService.java index 4f13f333..d5c9237c 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/nudge/MemoryNudgeService.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/nudge/MemoryNudgeService.java @@ -85,17 +85,21 @@ public class MemoryNudgeService { } try { - doNudge(agentId, conversationId, ownerKey); - lastNudgeTimes.put(cooldownKey, Instant.now()); + if (doNudge(agentId, conversationId, ownerKey)) { + lastNudgeTimes.put(cooldownKey, Instant.now()); + } } catch (Exception e) { log.warn("[Nudge] Failed for agent={}, conv={}: {}", agentId, conversationId, e.getMessage()); } } - private void doNudge(Long agentId, String conversationId, String ownerKey) { + private boolean doNudge(Long agentId, String conversationId, String ownerKey) { // 1. Load recent messages List messages = conversationService.listMessages(conversationId); + if (messages == null || messages.isEmpty()) { + return false; + } int maxReview = properties.getNudgeMaxMessages(); List recent = messages.size() > maxReview ? messages.subList(messages.size() - maxReview, messages.size()) @@ -103,12 +107,12 @@ public class MemoryNudgeService { if (recent.size() < 4) { log.debug("[Nudge] Not enough messages to review ({}), skipping", recent.size()); - return; + return false; } // 2. Build transcript String transcript = buildTranscript(recent); - if (transcript.isBlank()) return; + if (transcript.isBlank()) return false; // 3. Load existing structured memories for dedup (owner-scoped) String existingMemories = structuredMemoryService.buildMemoryBlock(agentId, ownerKey); @@ -131,11 +135,11 @@ public class MemoryNudgeService { llmResponse = callLlmWithRetry(chatModel, prompt, 2); if (llmResponse == null) { log.warn("[Nudge] LLM returned null after retries for agent={}", agentId); - return; + return false; } } catch (Exception e) { log.warn("[Nudge] LLM call failed for agent={}: {}", agentId, e.getMessage()); - return; + return false; } // 6. Parse and apply @@ -143,7 +147,7 @@ public class MemoryNudgeService { JsonNode root = parseJsonResponse(llmResponse); if (root == null || !root.isArray()) { log.debug("[Nudge] No entries extracted for agent={}", agentId); - return; + return false; } int saved = 0; @@ -163,9 +167,11 @@ public class MemoryNudgeService { if (saved > 0) { log.info("[Nudge] Extracted {} entries for agent={}", saved, agentId); } + return true; } catch (Exception e) { log.warn("[Nudge] Failed to parse nudge response for agent={}: {}", agentId, e.getMessage()); + return false; } } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/service/MemoryRecallService.java b/mateclaw-server/src/main/java/vip/mate/memory/service/MemoryRecallService.java index 9253c7be..f5e2c9b7 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/service/MemoryRecallService.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/service/MemoryRecallService.java @@ -59,14 +59,14 @@ public class MemoryRecallService { recordRecall(agentId, filename, snippetText, userQueryHash, null, MemoryScope.TEAM); } - /** Owner-aware recall ledger write. Shared legacy rows keep a null owner key. */ + /** Owner-aware recall ledger write. Shared rows use the canonical empty owner key. */ public void recordRecall(Long agentId, String filename, String snippetText, String userQueryHash, String ownerKey, String scope) { if (agentId == null || filename == null || filename.isBlank()) { return; } String effectiveScope = normalizeScope(scope); - String effectiveOwner = MemoryScope.PERSONAL.equals(effectiveScope) ? ownerKey : null; + String effectiveOwner = MemoryScope.PERSONAL.equals(effectiveScope) ? ownerKey : ""; if (MemoryScope.PERSONAL.equals(effectiveScope) && (effectiveOwner == null || effectiveOwner.isBlank())) { return; @@ -80,82 +80,91 @@ public class MemoryRecallService { ? snippetText.substring(0, 200) : snippetText; - LambdaQueryWrapper existingQuery = new LambdaQueryWrapper() - .eq(MemoryRecallEntity::getAgentId, agentId) - .eq(MemoryRecallEntity::getFilename, filename) - .eq(MemoryRecallEntity::getScope, effectiveScope) - .eq(MemoryRecallEntity::getDeleted, 0); - applyOwnerIdentity(existingQuery, effectiveOwner, effectiveScope); - MemoryRecallEntity existing = recallMapper.selectOne(existingQuery.last("LIMIT 1")); - LocalDateTime now = LocalDateTime.now(); - if (existing != null) { - existing.setRecallCount(existing.getRecallCount() + 1); - existing.setDailyCount(existing.getDailyCount() + 1); - existing.setLastRecalledAt(now); - existing.setSnippetPreview(preview); + // Update-first makes the hot path a single atomic SQL increment. The + // unique identity migration closes the insert race across threads and + // nodes; the loser retries this same atomic increment. + if (incrementExisting(agentId, filename, effectiveOwner, effectiveScope, preview, now) > 0) { + mergeQueryHash(agentId, filename, effectiveOwner, effectiveScope, userQueryHash); + return; + } + try { + MemoryRecallEntity entity = new MemoryRecallEntity(); + entity.setAgentId(agentId); + entity.setFilename(filename); + entity.setSnippetPreview(preview); + entity.setRecallCount(1); + entity.setDailyCount(1); + entity.setLastRecalledAt(now); + entity.setPromoted(false); + entity.setScore(0.0); + entity.setOwnerKey(effectiveOwner); + entity.setScope(effectiveScope); + entity.setCreateTime(now); + entity.setUpdateTime(now); + entity.setDeleted(0); if (userQueryHash != null) { - List hashes = parseQueryHashes(existing.getQueryHashes()); - if (!hashes.contains(userQueryHash) && hashes.size() < MAX_QUERY_HASHES) { - hashes.add(userQueryHash); - } - existing.setQueryHashes(toJson(hashes)); + entity.setQueryHashes(toJson(List.of(userQueryHash))); } - - recallMapper.updateById(existing); - } else { - // 防并发：trackRecalls 和 trackActiveRetrieval 可能同时插入同一 filename - try { - MemoryRecallEntity entity = new MemoryRecallEntity(); - entity.setAgentId(agentId); - entity.setFilename(filename); - entity.setSnippetPreview(preview); - entity.setRecallCount(1); - entity.setDailyCount(1); - entity.setLastRecalledAt(now); - entity.setPromoted(false); - entity.setScore(0.0); - entity.setOwnerKey(effectiveOwner); - entity.setScope(effectiveScope); - entity.setCreateTime(now); - entity.setUpdateTime(now); - entity.setDeleted(0); - - if (userQueryHash != null) { - entity.setQueryHashes(toJson(List.of(userQueryHash))); - } - - recallMapper.insert(entity); - } catch (org.springframework.dao.DuplicateKeyException e) { - // 并发插入冲突，重新查询后更新（不递归，避免 StackOverflow） - log.debug("[MemoryRecall] Concurrent insert for {}, falling back to update", filename); - LambdaQueryWrapper retryQuery = new LambdaQueryWrapper() - .eq(MemoryRecallEntity::getAgentId, agentId) - .eq(MemoryRecallEntity::getFilename, filename) - .eq(MemoryRecallEntity::getScope, effectiveScope) - .eq(MemoryRecallEntity::getDeleted, 0); - applyOwnerIdentity(retryQuery, effectiveOwner, effectiveScope); - MemoryRecallEntity retry = recallMapper.selectOne(retryQuery.last("LIMIT 1")); - if (retry != null) { - retry.setRecallCount(retry.getRecallCount() + 1); - retry.setDailyCount(retry.getDailyCount() + 1); - retry.setLastRecalledAt(now); - retry.setSnippetPreview(preview); - if (userQueryHash != null) { - List hashes = parseQueryHashes(retry.getQueryHashes()); - if (!hashes.contains(userQueryHash) && hashes.size() < MAX_QUERY_HASHES) { - hashes.add(userQueryHash); - } - retry.setQueryHashes(toJson(hashes)); - } - recallMapper.updateById(retry); - } + recallMapper.insert(entity); + } catch (org.springframework.dao.DuplicateKeyException e) { + log.debug("[MemoryRecall] Concurrent insert for {}, retrying atomic update", filename); + if (incrementExisting(agentId, filename, effectiveOwner, effectiveScope, preview, now) > 0) { + mergeQueryHash(agentId, filename, effectiveOwner, effectiveScope, userQueryHash); + } else { + log.warn("[MemoryRecall] Duplicate insert lost but active row was not found: agent={}, file={}, owner={}", + agentId, filename, effectiveOwner); } } } + private int incrementExisting(Long agentId, String filename, String ownerKey, String scope, + String preview, LocalDateTime now) { + LambdaUpdateWrapper update = new LambdaUpdateWrapper() + .eq(MemoryRecallEntity::getAgentId, agentId) + .eq(MemoryRecallEntity::getFilename, filename) + .eq(MemoryRecallEntity::getScope, scope) + .eq(MemoryRecallEntity::getOwnerKey, ownerKey) + .eq(MemoryRecallEntity::getDeleted, 0) + .setSql("recall_count = COALESCE(recall_count, 0) + 1") + .setSql("daily_count = COALESCE(daily_count, 0) + 1") + .set(MemoryRecallEntity::getLastRecalledAt, now) + .set(MemoryRecallEntity::getSnippetPreview, preview); + return recallMapper.update(null, update); + } + + /** Best-effort optimistic merge; counters remain atomic even under hash contention. */ + private void mergeQueryHash(Long agentId, String filename, String ownerKey, String scope, + String userQueryHash) { + if (userQueryHash == null) return; + for (int attempt = 0; attempt < 3; attempt++) { + MemoryRecallEntity current = recallMapper.selectOne( + new LambdaQueryWrapper() + .eq(MemoryRecallEntity::getAgentId, agentId) + .eq(MemoryRecallEntity::getFilename, filename) + .eq(MemoryRecallEntity::getScope, scope) + .eq(MemoryRecallEntity::getOwnerKey, ownerKey) + .eq(MemoryRecallEntity::getDeleted, 0) + .last("LIMIT 1")); + if (current == null) return; + List hashes = parseQueryHashes(current.getQueryHashes()); + if (hashes.contains(userQueryHash) || hashes.size() >= MAX_QUERY_HASHES) return; + hashes.add(userQueryHash); + String previous = current.getQueryHashes(); + LambdaUpdateWrapper cas = new LambdaUpdateWrapper() + .eq(MemoryRecallEntity::getId, current.getId()) + .eq(MemoryRecallEntity::getDeleted, 0) + .set(MemoryRecallEntity::getQueryHashes, toJson(hashes)); + if (previous == null) cas.isNull(MemoryRecallEntity::getQueryHashes); + else cas.eq(MemoryRecallEntity::getQueryHashes, previous); + if (recallMapper.update(null, cas) > 0) return; + } + log.debug("[MemoryRecall] Query-hash merge contended for agent={}, file={}, owner={}", + agentId, filename, ownerKey); + } + /** * 重置所有记录的 dailyCount（在每轮 dreaming 开始时调用） */ @@ -404,16 +413,4 @@ public class MemoryRecallService { return MemoryScope.TEAM; } - private static void applyOwnerIdentity(LambdaQueryWrapper query, - String ownerKey, String scope) { - if (MemoryScope.PERSONAL.equals(scope)) { - query.eq(MemoryRecallEntity::getOwnerKey, ownerKey); - } else { - // V137 left legacy shared recall rows with NULL while newer rows may - // use the workspace-file empty-string sentinel. Treat both as shared. - query.and(w -> w.isNull(MemoryRecallEntity::getOwnerKey) - .or().eq(MemoryRecallEntity::getOwnerKey, "")); - } - } - } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/service/MemorySummarizationGate.java b/mateclaw-server/src/main/java/vip/mate/memory/service/MemorySummarizationGate.java index 3eb30d82..4b0a21f1 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/service/MemorySummarizationGate.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/service/MemorySummarizationGate.java @@ -11,7 +11,7 @@ import java.util.regex.Pattern; /** * Filters conversations that should not be promoted into long-term memory. */ -final class MemorySummarizationGate { +public final class MemorySummarizationGate { private static final Pattern FINISH_REASON = Pattern.compile( "\"(?:finishReason|finish_reason)\"\\s*:\\s*\"([^\"]+)\""); @@ -36,14 +36,14 @@ final class MemorySummarizationGate { } if (isExplicitRememberRequest(latestUser)) { - return Decision.analyze(); + return Decision.analyze(true); } if (looksLikeSourceAnalysis(latestUser)) { return Decision.skip("source-analysis conversations are one-off work, not long-term memory"); } - return Decision.analyze(); + return Decision.analyze(false); } private static boolean isNonDurableFinishReason(String finishReason) { @@ -56,7 +56,7 @@ final class MemorySummarizationGate { }; } - private static boolean isExplicitRememberRequest(String text) { + public static boolean isExplicitRememberRequest(String text) { String normalized = normalize(text); return normalized.contains("记住") || normalized.contains("remember") || normalized.contains("保存到记忆") || normalized.contains("写入记忆"); @@ -122,13 +122,13 @@ final class MemorySummarizationGate { return text == null ? "" : text.toLowerCase(Locale.ROOT); } - record Decision(boolean shouldAnalyze, String reason) { - static Decision analyze() { - return new Decision(true, "eligible"); + record Decision(boolean shouldAnalyze, boolean bypassCooldown, String reason) { + static Decision analyze(boolean bypassCooldown) { + return new Decision(true, bypassCooldown, "eligible"); } static Decision skip(String reason) { - return new Decision(false, reason); + return new Decision(false, false, reason); } } } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/service/MemorySummarizationService.java b/mateclaw-server/src/main/java/vip/mate/memory/service/MemorySummarizationService.java index 61817aee..91402bae 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/service/MemorySummarizationService.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/service/MemorySummarizationService.java @@ -81,30 +81,17 @@ public class MemorySummarizationService { // extraction never starves another owner sharing the same agent. String lockKey = agentId + ":" + (ownerKey == null ? "" : ownerKey); - // 冷却检查 - if (isInCooldown(lockKey)) { - log.debug("[Memory] Agent {} (owner {}) is in cooldown, skipping summarization", agentId, ownerKey); - return; - } - - ReentrantLock lock = agentLocks.computeIfAbsent(lockKey, k -> new ReentrantLock()); - if (!lock.tryLock()) { - log.debug("[Memory] Agent {} (owner {}) is already being summarized, skipping", agentId, ownerKey); - return; - } - - try { - doAnalyzeAndUpdate(agentId, conversationId, ownerKey); - lastRunTimes.put(lockKey, Instant.now()); - } finally { - lock.unlock(); - } - } - - private void doAnalyzeAndUpdate(Long agentId, String conversationId, String ownerKey) { - // 1. 加载对话消息 + // Load and classify before applying cooldown. An explicit user request + // to remember something must always get a chance to run, and skipped / + // unsupported conversations must not poison the next real request. List messages = conversationService.listMessages(conversationId); - if (messages.size() < properties.getMinMessagesForSummarize()) { + if (messages == null || messages.isEmpty()) { + log.debug("[Memory] Conversation {} has no messages, skipping", conversationId); + return; + } + String latestUser = latestMessageContent(messages, "user"); + boolean explicitRemember = MemorySummarizationGate.isExplicitRememberRequest(latestUser); + if (!explicitRemember && messages.size() < properties.getMinMessagesForSummarize()) { log.debug("[Memory] Conversation {} has only {} messages, skipping", conversationId, messages.size()); return; @@ -116,7 +103,31 @@ public class MemorySummarizationService { return; } - // 2. 加载现有记忆文件内容（按 owner 隔离） + // 冷却检查 + if (!decision.bypassCooldown() && isInCooldown(lockKey)) { + log.debug("[Memory] Agent {} (owner {}) is in cooldown, skipping summarization", agentId, ownerKey); + return; + } + + ReentrantLock lock = agentLocks.computeIfAbsent(lockKey, k -> new ReentrantLock()); + if (!lock.tryLock()) { + log.debug("[Memory] Agent {} (owner {}) is already being summarized, skipping", agentId, ownerKey); + return; + } + + try { + AnalysisOutcome outcome = doAnalyzeAndUpdate(agentId, conversationId, ownerKey, messages); + if (outcome == AnalysisOutcome.COMPLETED) { + lastRunTimes.put(lockKey, Instant.now()); + } + } finally { + lock.unlock(); + } + } + + private AnalysisOutcome doAnalyzeAndUpdate(Long agentId, String conversationId, String ownerKey, + List messages) { + // 1. 加载现有记忆文件内容（按 owner 隔离） String profileContent = readFileContentSafe(agentId, "PROFILE.md", ownerKey); String memoryContent = readFileContentSafe(agentId, "MEMORY.md", ownerKey); String dailyFilename = "memory/" + LocalDate.now() + ".md"; @@ -125,7 +136,7 @@ public class MemorySummarizationService { // 3. 构建对话 transcript String transcript = buildTranscript(messages); if (transcript.isBlank()) { - return; + return AnalysisOutcome.SKIPPED; } // 4. 调用 LLM 分析 @@ -150,22 +161,25 @@ public class MemorySummarizationService { llmResponse = callLlmWithRetry(chatModel, prompt, 2); if (llmResponse == null) { log.warn("[Memory] LLM returned null after retries for agent={}, conv={}", agentId, conversationId); - return; + return AnalysisOutcome.FAILED; } } catch (Exception e) { log.warn("[Memory] LLM call failed for agent={}, conv={}: {}", agentId, conversationId, e.getMessage()); - return; + return AnalysisOutcome.FAILED; } // 5. 解析 JSON 响应 try { JsonNode root = parseJsonResponse(llmResponse); - if (root == null || !root.path("should_update").asBoolean(false)) { - String reason = root != null ? root.path("reason").asText("") : "parse failed"; + if (root == null) { + return AnalysisOutcome.FAILED; + } + if (!root.path("should_update").asBoolean(false)) { + String reason = root.path("reason").asText(""); log.info("[Memory] No update needed for agent={}, conv={}: {}", agentId, conversationId, reason); - return; + return AnalysisOutcome.COMPLETED; } // 6. 应用更新 @@ -173,13 +187,32 @@ public class MemorySummarizationService { String reason = root.path("reason").asText(""); log.info("[Memory] Memory updated for agent={}, conv={}: {}", agentId, conversationId, reason); + return AnalysisOutcome.COMPLETED; } catch (Exception e) { log.warn("[Memory] Failed to parse/apply memory update for agent={}, conv={}: {}", agentId, conversationId, e.getMessage()); + return AnalysisOutcome.FAILED; } } + private static String latestMessageContent(List messages, String role) { + if (messages == null) return ""; + for (int i = messages.size() - 1; i >= 0; i--) { + MessageEntity message = messages.get(i); + if (role.equals(message.getRole()) && message.getContent() != null) { + return message.getContent(); + } + } + return ""; + } + + private enum AnalysisOutcome { + COMPLETED, + SKIPPED, + FAILED + } + private void applyUpdates(Long agentId, JsonNode root, String dailyFilename, String existingDailyContent, String ownerKey) { // Daily entry: 追加模式 diff --git a/mateclaw-server/src/main/java/vip/mate/memory/spi/MemoryManager.java b/mateclaw-server/src/main/java/vip/mate/memory/spi/MemoryManager.java index 4750feae..32f51a81 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/spi/MemoryManager.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/spi/MemoryManager.java @@ -1,6 +1,7 @@ package vip.mate.memory.spi; import io.micrometer.core.instrument.MeterRegistry; +import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import vip.mate.agent.context.TokenEstimator; @@ -11,7 +12,16 @@ import vip.mate.memory.spi.decorator.RetryableMemoryProvider; import java.util.ArrayList; import java.util.Comparator; import java.util.List; +import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -27,11 +37,17 @@ import java.util.stream.Collectors; */ @Slf4j @Component -public class MemoryManager { +public class MemoryManager implements AutoCloseable { private static final Pattern FENCE_TAG_RE = Pattern.compile("", Pattern.CASE_INSENSITIVE); private final List providers; + private final ExecutorService prefetchExecutor; + private final Map providerCircuits = new ConcurrentHashMap<>(); + private final long providerPrefetchTimeoutMs; + private final long providerPrefetchTotalBudgetMs; + private final int providerCircuitFailureThreshold; + private final long providerCircuitCooldownNanos; /** External plugin memory provider (single-select constraint) */ private volatile MemoryProvider externalPluginProvider = null; @@ -47,9 +63,15 @@ public class MemoryManager { .collect(Collectors.toList()); // Assemble decorator chain based on flags - this.providers = filtered.stream() + this.providers = new CopyOnWriteArrayList<>(filtered.stream() .map(p -> wrapWithDecorators(p, properties, meterRegistry)) - .collect(Collectors.toList()); + .collect(Collectors.toList())); + this.prefetchExecutor = Executors.newVirtualThreadPerTaskExecutor(); + this.providerPrefetchTimeoutMs = Math.max(0, properties.getProviderPrefetchTimeoutMs()); + this.providerPrefetchTotalBudgetMs = Math.max(0, properties.getProviderPrefetchTotalBudgetMs()); + this.providerCircuitFailureThreshold = Math.max(1, properties.getProviderCircuitFailureThreshold()); + this.providerCircuitCooldownNanos = TimeUnit.SECONDS.toNanos( + Math.max(0, properties.getProviderCircuitCooldownSeconds())); if (!disabled.isEmpty()) { log.info("[MemoryManager] Disabled providers: {}", disabled); @@ -164,15 +186,50 @@ public class MemoryManager { */ public String prefetchAll(Long agentId, String userQuery, String ownerKey) { List parts = new ArrayList<>(); + long startedAt = System.nanoTime(); + long totalBudgetNanos = providerPrefetchTotalBudgetMs == 0 + ? Long.MAX_VALUE : TimeUnit.MILLISECONDS.toNanos(providerPrefetchTotalBudgetMs); for (MemoryProvider provider : providers) { + long now = System.nanoTime(); + long remainingNanos = remainingBudget(totalBudgetNanos, startedAt, now); + if (remainingNanos <= 0) { + log.debug("[MemoryManager] Prefetch total budget exhausted before provider '{}'", provider.id()); + break; + } + ProviderCircuit circuit = providerCircuits.computeIfAbsent(provider.id(), ignored -> new ProviderCircuit()); + if (!circuit.tryAcquire(now, providerCircuitCooldownNanos)) { + log.debug("[MemoryManager] Provider '{}' prefetch skipped while circuit is open", provider.id()); + continue; + } + Future future = prefetchExecutor.submit(() -> provider.prefetch(agentId, userQuery, ownerKey)); try { - String result = provider.prefetch(agentId, userQuery, ownerKey); + long providerLimitNanos = providerPrefetchTimeoutMs == 0 + ? Long.MAX_VALUE : TimeUnit.MILLISECONDS.toNanos(providerPrefetchTimeoutMs); + long waitNanos = Math.min(providerLimitNanos, remainingNanos); + String result = waitNanos == Long.MAX_VALUE + ? future.get() : future.get(waitNanos, TimeUnit.NANOSECONDS); + circuit.onSuccess(); if (result != null && !result.isBlank()) { parts.add(sanitizeContext(result)); } - } catch (Exception e) { + } catch (TimeoutException e) { + future.cancel(true); + circuit.onFailure(providerCircuitFailureThreshold); + log.warn("[MemoryManager] Provider '{}' prefetch timed out after at most {} ms", + provider.id(), TimeUnit.NANOSECONDS.toMillis(Math.min( + providerPrefetchTimeoutMs == 0 ? remainingNanos + : TimeUnit.MILLISECONDS.toNanos(providerPrefetchTimeoutMs), remainingNanos))); + } catch (ExecutionException e) { + circuit.onFailure(providerCircuitFailureThreshold); + Throwable cause = e.getCause() != null ? e.getCause() : e; log.debug("[MemoryManager] Provider '{}' prefetch failed (non-fatal): {}", - provider.id(), e.getMessage()); + provider.id(), cause.getMessage()); + } catch (InterruptedException e) { + future.cancel(true); + Thread.currentThread().interrupt(); + circuit.onFailure(providerCircuitFailureThreshold); + log.debug("[MemoryManager] Provider '{}' prefetch interrupted", provider.id()); + break; } } if (parts.isEmpty()) { @@ -300,8 +357,13 @@ public class MemoryManager { throw new vip.mate.plugin.api.PluginException( "Only one external memory provider allowed. Current: " + externalPluginProvider.id()); } + if (providers.stream().anyMatch(existing -> existing.id().equals(provider.id()))) { + throw new vip.mate.plugin.api.PluginException( + "Memory provider ID already registered: " + provider.id()); + } if (!provider.isAvailable()) { log.warn("[MemoryManager] Plugin provider '{}' is not available, skipping", provider.id()); + closeProvider(provider); return; } externalPluginProvider = provider; @@ -315,8 +377,11 @@ public class MemoryManager { */ public synchronized void unregisterPluginProvider(String providerId) { if (externalPluginProvider != null && externalPluginProvider.id().equals(providerId)) { - providers.removeIf(p -> p.id().equals(providerId)); + MemoryProvider removed = externalPluginProvider; + providers.remove(removed); externalPluginProvider = null; + providerCircuits.remove(providerId); + closeProvider(removed); log.info("[MemoryManager] Plugin provider unregistered: {}", providerId); } } @@ -344,4 +409,59 @@ public class MemoryManager { public List getProviderIds() { return providers.stream().map(MemoryProvider::id).toList(); } + + private static long remainingBudget(long totalBudgetNanos, long startedAt, long now) { + if (totalBudgetNanos == Long.MAX_VALUE) { + return Long.MAX_VALUE; + } + return totalBudgetNanos - (now - startedAt); + } + + private void closeProvider(MemoryProvider provider) { + try { + provider.close(); + } catch (Exception e) { + log.warn("[MemoryManager] Provider '{}' close failed: {}", provider.id(), e.getMessage()); + } + } + + @Override + @PreDestroy + public void close() { + prefetchExecutor.shutdownNow(); + providers.forEach(this::closeProvider); + providers.clear(); + providerCircuits.clear(); + externalPluginProvider = null; + } + + private static final class ProviderCircuit { + private int consecutiveFailures; + private long openedAtNanos; + private boolean probeInFlight; + + synchronized boolean tryAcquire(long now, long cooldownNanos) { + if (openedAtNanos == 0) { + return true; + } + if (now - openedAtNanos < cooldownNanos || probeInFlight) { + return false; + } + probeInFlight = true; + return true; + } + + synchronized void onSuccess() { + consecutiveFailures = 0; + openedAtNanos = 0; + probeInFlight = false; + } + + synchronized void onFailure(int threshold) { + probeInFlight = false; + if (++consecutiveFailures >= threshold) { + openedAtNanos = System.nanoTime(); + } + } + } } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/spi/MemoryProvider.java b/mateclaw-server/src/main/java/vip/mate/memory/spi/MemoryProvider.java index 262c0c32..9b8c6b8b 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/spi/MemoryProvider.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/spi/MemoryProvider.java @@ -16,7 +16,7 @@ import java.util.List; * * @author MateClaw Team */ -public interface MemoryProvider { +public interface MemoryProvider extends AutoCloseable { /** * Unique provider identifier, e.g. "builtin", "structured", "session_search". @@ -160,4 +160,9 @@ public interface MemoryProvider { */ default void evict(Long agentId) { } + + /** Release provider-owned threads, clients, and other resources. */ + @Override + default void close() { + } } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/MemoryProviderDecorator.java b/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/MemoryProviderDecorator.java index d7480fa0..bbcb03ef 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/MemoryProviderDecorator.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/MemoryProviderDecorator.java @@ -39,4 +39,5 @@ public abstract class MemoryProviderDecorator implements MemoryProvider { } @Override public void warmup(Long agentId) { delegate.warmup(agentId); } @Override public void evict(Long agentId) { delegate.evict(agentId); } + @Override public void close() { delegate.close(); } } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/RetryableMemoryProvider.java b/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/RetryableMemoryProvider.java index ac584215..bb9ba2b1 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/RetryableMemoryProvider.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/RetryableMemoryProvider.java @@ -40,7 +40,7 @@ public class RetryableMemoryProvider extends MemoryProviderDecorator { } log.warn("[Retry] prefetch exhausted {} attempts for provider={}: {}", maxAttempts, delegate.id(), lastException != null ? lastException.getMessage() : ""); - return ""; + throw new IllegalStateException("Provider prefetch exhausted retries: " + delegate.id(), lastException); } @Override @@ -66,6 +66,7 @@ public class RetryableMemoryProvider extends MemoryProviderDecorator { } log.warn("[Retry] syncTurn exhausted {} attempts for provider={}: {}", maxAttempts, delegate.id(), lastException != null ? lastException.getMessage() : ""); + throw new IllegalStateException("Provider sync exhausted retries: " + delegate.id(), lastException); } private void sleep(int attempt) { @@ -73,6 +74,7 @@ public class RetryableMemoryProvider extends MemoryProviderDecorator { Thread.sleep((long) Math.pow(2, attempt - 1) * 100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); + throw new IllegalStateException("Provider retry interrupted: " + delegate.id(), e); } } } diff --git a/mateclaw-server/src/main/java/vip/mate/plugin/bridge/PluginMemoryBridge.java b/mateclaw-server/src/main/java/vip/mate/plugin/bridge/PluginMemoryBridge.java index b6c9d581..9d33f71d 100644 --- a/mateclaw-server/src/main/java/vip/mate/plugin/bridge/PluginMemoryBridge.java +++ b/mateclaw-server/src/main/java/vip/mate/plugin/bridge/PluginMemoryBridge.java @@ -76,4 +76,9 @@ public class PluginMemoryBridge implements MemoryProvider { public void onSessionEnd(Long agentId, String conversationId) { delegate.onSessionEnd(agentId, conversationId); } + + @Override + public void close() { + delegate.close(); + } } diff --git a/mateclaw-server/src/main/resources/application.yml b/mateclaw-server/src/main/resources/application.yml index 51fa3f7b..88935553 100644 --- a/mateclaw-server/src/main/resources/application.yml +++ b/mateclaw-server/src/main/resources/application.yml @@ -523,6 +523,10 @@ mate: soul-update-interval: 20 # 20 writes trigger one SOUL.md LLM update (0 = off) provider-retry-attempts: 1 # 1 = no retry (enable when external providers added) provider-metrics-enabled: false # actuator dependency now present; enable when external providers added + provider-prefetch-timeout-ms: 1500 # per-provider recall deadline; 0 = unlimited + provider-prefetch-total-budget-ms: 2500 # deadline for the complete recall chain; 0 = unlimited + provider-circuit-failure-threshold: 3 # consecutive recall failures before opening the circuit + provider-circuit-cooldown-seconds: 30 # open-circuit delay before one half-open probe # Phase 3: fact projection fact: projection-enabled: true diff --git a/mateclaw-server/src/main/resources/db/migration/h2/V192__memory_recall_unique_identity.sql b/mateclaw-server/src/main/resources/db/migration/h2/V192__memory_recall_unique_identity.sql new file mode 100644 index 00000000..b176ef05 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/h2/V192__memory_recall_unique_identity.sql @@ -0,0 +1,39 @@ +-- V192: make owner-aware recall writes race-safe. +-- Irreversible cleanup: soft-deleted rows are no longer useful to the recall +-- ledger, and duplicate active identities must collapse before uniqueness. +DELETE FROM mate_memory_recall WHERE deleted <> 0; +UPDATE mate_memory_recall SET owner_key = '' WHERE owner_key IS NULL; +UPDATE mate_memory_recall target +SET recall_count = (SELECT SUM(COALESCE(source.recall_count, 0)) + FROM mate_memory_recall source + WHERE source.agent_id = target.agent_id + AND source.filename = target.filename + AND source.scope = target.scope + AND source.owner_key = target.owner_key), + daily_count = (SELECT SUM(COALESCE(source.daily_count, 0)) + FROM mate_memory_recall source + WHERE source.agent_id = target.agent_id + AND source.filename = target.filename + AND source.scope = target.scope + AND source.owner_key = target.owner_key), + last_recalled_at = (SELECT MAX(source.last_recalled_at) + FROM mate_memory_recall source + WHERE source.agent_id = target.agent_id + AND source.filename = target.filename + AND source.scope = target.scope + AND source.owner_key = target.owner_key) +WHERE target.id IN ( + SELECT MAX(id) FROM mate_memory_recall + GROUP BY agent_id, filename, scope, owner_key + HAVING COUNT(*) > 1 +); +DELETE FROM mate_memory_recall +WHERE id NOT IN ( + SELECT MAX(id) + FROM mate_memory_recall + GROUP BY agent_id, filename, scope, owner_key +); +ALTER TABLE mate_memory_recall ALTER COLUMN owner_key SET DEFAULT ''; +ALTER TABLE mate_memory_recall ALTER COLUMN owner_key SET NOT NULL; +CREATE UNIQUE INDEX IF NOT EXISTS uk_memory_recall_identity + ON mate_memory_recall(agent_id, filename, scope, owner_key); diff --git a/mateclaw-server/src/main/resources/db/migration/kingbase/V192__memory_recall_unique_identity.sql b/mateclaw-server/src/main/resources/db/migration/kingbase/V192__memory_recall_unique_identity.sql new file mode 100644 index 00000000..a9fc4cfd --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/kingbase/V192__memory_recall_unique_identity.sql @@ -0,0 +1,39 @@ +-- V192: make owner-aware recall writes race-safe. +-- Irreversible cleanup: soft-deleted rows are no longer useful to the recall +-- ledger, and duplicate active identities must collapse before uniqueness. +DELETE FROM mate_memory_recall WHERE deleted <> 0; +UPDATE mate_memory_recall SET owner_key = '' WHERE owner_key IS NULL; +UPDATE mate_memory_recall AS target +SET recall_count = (SELECT SUM(COALESCE(source.recall_count, 0)) + FROM mate_memory_recall AS source + WHERE source.agent_id = target.agent_id + AND source.filename = target.filename + AND source.scope = target.scope + AND source.owner_key = target.owner_key), + daily_count = (SELECT SUM(COALESCE(source.daily_count, 0)) + FROM mate_memory_recall AS source + WHERE source.agent_id = target.agent_id + AND source.filename = target.filename + AND source.scope = target.scope + AND source.owner_key = target.owner_key), + last_recalled_at = (SELECT MAX(source.last_recalled_at) + FROM mate_memory_recall AS source + WHERE source.agent_id = target.agent_id + AND source.filename = target.filename + AND source.scope = target.scope + AND source.owner_key = target.owner_key) +WHERE target.id IN ( + SELECT MAX(id) FROM mate_memory_recall + GROUP BY agent_id, filename, scope, owner_key + HAVING COUNT(*) > 1 +); +DELETE FROM mate_memory_recall +WHERE id NOT IN ( + SELECT MAX(id) + FROM mate_memory_recall + GROUP BY agent_id, filename, scope, owner_key +); +ALTER TABLE mate_memory_recall ALTER COLUMN owner_key SET DEFAULT ''; +ALTER TABLE mate_memory_recall ALTER COLUMN owner_key SET NOT NULL; +CREATE UNIQUE INDEX IF NOT EXISTS uk_memory_recall_identity + ON mate_memory_recall(agent_id, filename, scope, owner_key); diff --git a/mateclaw-server/src/main/resources/db/migration/mysql/V192__memory_recall_unique_identity.sql b/mateclaw-server/src/main/resources/db/migration/mysql/V192__memory_recall_unique_identity.sql new file mode 100644 index 00000000..e32d5736 --- /dev/null +++ b/mateclaw-server/src/main/resources/db/migration/mysql/V192__memory_recall_unique_identity.sql @@ -0,0 +1,35 @@ +-- V192: make owner-aware recall writes race-safe. +-- Irreversible cleanup: soft-deleted rows are no longer useful to the recall +-- ledger, and duplicate active identities must collapse before uniqueness. +DELETE FROM mate_memory_recall WHERE deleted <> 0; +UPDATE mate_memory_recall SET owner_key = '' WHERE owner_key IS NULL; +DROP TABLE IF EXISTS tmp_memory_recall_merge; +CREATE TEMPORARY TABLE tmp_memory_recall_merge AS +SELECT MAX(id) AS keep_id, + SUM(COALESCE(recall_count, 0)) AS recall_total, + SUM(COALESCE(daily_count, 0)) AS daily_total, + MAX(last_recalled_at) AS last_recalled +FROM mate_memory_recall +GROUP BY agent_id, filename, scope, owner_key +HAVING COUNT(*) > 1; +UPDATE mate_memory_recall AS target +SET recall_count = (SELECT merged.recall_total FROM tmp_memory_recall_merge merged + WHERE merged.keep_id = target.id), + daily_count = (SELECT merged.daily_total FROM tmp_memory_recall_merge merged + WHERE merged.keep_id = target.id), + last_recalled_at = (SELECT merged.last_recalled FROM tmp_memory_recall_merge merged + WHERE merged.keep_id = target.id) +WHERE target.id IN (SELECT keep_id FROM tmp_memory_recall_merge); +DROP TABLE tmp_memory_recall_merge; +DELETE FROM mate_memory_recall +WHERE id NOT IN ( + SELECT keep_id FROM ( + SELECT MAX(id) AS keep_id + FROM mate_memory_recall + GROUP BY agent_id, filename, scope, owner_key + ) retained +); +ALTER TABLE mate_memory_recall + MODIFY COLUMN owner_key VARCHAR(128) NOT NULL DEFAULT ''; +CREATE UNIQUE INDEX uk_memory_recall_identity + ON mate_memory_recall(agent_id, filename, scope, owner_key); diff --git a/mateclaw-server/src/main/resources/docs/en/memory.md b/mateclaw-server/src/main/resources/docs/en/memory.md index ca93879b..ea7e656d 100644 --- a/mateclaw-server/src/main/resources/docs/en/memory.md +++ b/mateclaw-server/src/main/resources/docs/en/memory.md @@ -233,12 +233,12 @@ After a turn completes, the system handles extraction on a background thread. A - Message count meets the minimum (default 4) - The last user message is long enough (default at least 10 chars) -All pass — extraction begins. +An explicit request such as “remember this” bypasses the message-count, message-length, and cooldown gates. A failed or intentionally skipped analysis does not start the cooldown, so a later eligible request can retry immediately. ### Concurrency control -- **Cooldown** — same agent won't extract twice within 5 minutes (default) -- **Per-agent lock** — if an extraction is already running for this agent, the new request is skipped +- **Cooldown** — the same agent/owner bucket won't extract twice within 5 minutes (default) +- **Per-agent/owner lock** — if extraction is already running for this bucket, the new request is skipped ### What the LLM actually does @@ -586,7 +586,7 @@ Mem0 integration is an **optional community contribution** — it is NOT part of | `syncTurn(agentId, conversationId, userMessage, assistantReply, ownerKey)` | When `syncEnabled=true` and `ownerKey` is non-blank, **asynchronously** pushes this turn's user/assistant messages to `POST {baseUrl}/memories/` under `user_id = ownerKey` — the same identifier recall queries by. Failures are logged only, never block the response | | `getToolBeans` | Empty list — v1 exposes no agent-callable tools | -**Fault isolation**: any exception in recall or sync is swallowed and logged by the plugin itself; the platform keeps going with the other providers. Mem0 being down does not affect MateClaw's local memory. +**Fault isolation**: sync failures are logged inside the plugin. Recall failures propagate to the platform's provider boundary, where they are isolated from other providers and counted by the circuit breaker. Each provider has a deadline, the full recall chain has a total latency budget, and repeatedly failing providers are temporarily skipped. Mem0 being down therefore does not block MateClaw's local memory. ### Per-owner isolation mapping @@ -616,9 +616,12 @@ Both `prefetch` and `syncTurn` receive `ownerKey` from the platform, so writes a | `syncEnabled` | boolean | no | `true` | Whether syncTurn should push each turn to `/memories/` | | `maxResults` | integer | no | `5` | Cap on memories returned per recall | | `timeoutMs` | integer | no | `3000` | HTTP timeout in milliseconds, shared by recall and sync | +| `syncQueueCapacity` | integer | no | `256` | Maximum pending asynchronous sync turns; new writes are dropped with a warning when the queue is full | Config is read once at plugin load — changes require a plugin reload to take effect. +The platform-level recall guards are configured under `mate.memory`: `provider-prefetch-timeout-ms` (default `1500`), `provider-prefetch-total-budget-ms` (default `2500`), `provider-circuit-failure-threshold` (default `3`), and `provider-circuit-cooldown-seconds` (default `30`). Set either timeout/budget to `0` only when an unlimited wait is explicitly desired. + ### Known limitations (v1) - **Turns without a resolved owner are not synced**: `syncTurn` requires `ownerKey`; turns where the platform cannot resolve one (e.g. system-triggered runs) are skipped rather than written under a fallback identifier that recall could never surface. diff --git a/mateclaw-server/src/main/resources/docs/zh/memory.md b/mateclaw-server/src/main/resources/docs/zh/memory.md index 8a152b29..982a86c1 100644 --- a/mateclaw-server/src/main/resources/docs/zh/memory.md +++ b/mateclaw-server/src/main/resources/docs/zh/memory.md @@ -232,12 +232,12 @@ mate: - 消息数达到下限（默认 4 条） - 最后一条用户消息够长（默认至少 10 字符） -全部通过，开始提取。 +“记住这个”一类显式记忆请求会绕过消息数、消息长度和冷却门控。分析失败或被规则跳过时不会启动冷却，后续符合条件的请求可以立即重试。 ### 并发控制 -- **冷却**——同一个 Agent 在默认 5 分钟内不会重复提取 -- **按 Agent 加锁**——同一个 Agent 已经有一个提取任务在跑，新任务直接跳过 +- **冷却**——同一个 Agent/owner 记忆桶在默认 5 分钟内不会重复提取 +- **按 Agent/owner 加锁**——同一个记忆桶已有提取任务在跑时，新任务直接跳过 ### LLM 实际在做什么 @@ -580,7 +580,7 @@ Mem0 集成是**可选的社区贡献项**，不在 MateClaw 的默认安装里 | `syncTurn(agentId, conversationId, userMessage, assistantReply, ownerKey)` | 当 `syncEnabled=true` 且 `ownerKey` 非空时，**异步**把这一轮的 user/assistant 消息以 `user_id = ownerKey` 推到 `POST {baseUrl}/memories/` —— 与召回查询用同一个标识。失败只记日志、不阻塞响应 | | `getToolBeans` | 空列表——v1 不暴露 Agent 可调用的工具 | -**故障隔离**：recall 或 sync 任何一边抛异常，插件自己吞掉、写日志，平台继续走其他 provider。Mem0 挂了不会影响 MateClaw 的本地记忆。 +**故障隔离**：sync 异常由插件内部记录；recall 异常会上抛到平台的 provider 边界，由平台隔离并计入熔断器。每个 provider 有独立超时，整条召回链还有总时延预算；连续失败的 provider 会暂时跳过。因此 Mem0 挂了不会阻塞 MateClaw 的本地记忆。 ### per-owner 隔离的映射 @@ -610,9 +610,12 @@ Mem0 用 `user_id` + `agent_id` 做隔离。MateClaw 的映射： | `syncEnabled` | boolean | 否 | `true` | 是否在 syncTurn 时把每轮对话推到 `/memories/` | | `maxResults` | integer | 否 | `5` | 每次召回返回的记忆条数上限 | | `timeoutMs` | integer | 否 | `3000` | HTTP 超时（毫秒），recall 和 sync 共用 | +| `syncQueueCapacity` | integer | 否 | `256` | 异步同步队列最大待处理轮次；队列满时丢弃新写入并记录警告 | 配置只在插件加载时读一次——改了要重载插件才会生效。 +平台层召回保护位于 `mate.memory`：`provider-prefetch-timeout-ms`（默认 `1500`）、`provider-prefetch-total-budget-ms`（默认 `2500`）、`provider-circuit-failure-threshold`（默认 `3`）、`provider-circuit-cooldown-seconds`（默认 `30`）。只有明确需要无限等待时，才把单项超时或总预算设为 `0`。 + ### 已知限制（v1） - **没有解析出 owner 的轮次不会同步**：`syncTurn` 要求 `ownerKey`；平台解析不出 owner 的轮次（如系统触发的运行）会直接跳过，而不是用一个召回永远查不到的降级标识写入。 diff --git a/mateclaw-server/src/test/java/vip/mate/memory/MemoryManagerResilienceTest.java b/mateclaw-server/src/test/java/vip/mate/memory/MemoryManagerResilienceTest.java new file mode 100644 index 00000000..14a6da76 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/memory/MemoryManagerResilienceTest.java @@ -0,0 +1,116 @@ +package vip.mate.memory; + +import io.micrometer.core.instrument.MeterRegistry; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.ObjectProvider; +import vip.mate.memory.spi.MemoryManager; +import vip.mate.memory.spi.MemoryProvider; + +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.assertj.core.api.Assertions.assertThat; + +class MemoryManagerResilienceTest { + + @Test + void providerTimeoutDoesNotBlockLaterProviders() { + MemoryProperties properties = new MemoryProperties(); + properties.setProviderPrefetchTimeoutMs(50); + properties.setProviderPrefetchTotalBudgetMs(200); + MemoryProvider slow = provider("slow", () -> { + try { + Thread.sleep(5_000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return "late"; + }); + MemoryProvider fast = provider("fast", () -> "useful"); + + long started = System.nanoTime(); + try (MemoryManager manager = manager(properties, slow, fast)) { + String result = manager.prefetchAll(1L, "query", "user:1"); + assertThat(result).contains("useful").doesNotContain("late"); + } + assertThat((System.nanoTime() - started) / 1_000_000).isLessThan(1_000); + } + + @Test + void totalBudgetStopsDispatchingRemainingProviders() { + MemoryProperties properties = new MemoryProperties(); + properties.setProviderPrefetchTimeoutMs(500); + properties.setProviderPrefetchTotalBudgetMs(60); + AtomicInteger laterCalls = new AtomicInteger(); + MemoryProvider slow = provider("slow", () -> { + try { + Thread.sleep(500); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return "late"; + }); + MemoryProvider later = provider("later", () -> { + laterCalls.incrementAndGet(); + return "later"; + }); + + try (MemoryManager manager = manager(properties, slow, later)) { + assertThat(manager.prefetchAll(1L, "query")).isEmpty(); + } + assertThat(laterCalls).hasValue(0); + } + + @Test + void openCircuitSkipsRepeatedFailures() { + MemoryProperties properties = new MemoryProperties(); + properties.setProviderCircuitFailureThreshold(1); + properties.setProviderCircuitCooldownSeconds(60); + AtomicInteger calls = new AtomicInteger(); + MemoryProvider broken = provider("broken", () -> { + calls.incrementAndGet(); + throw new IllegalStateException("offline"); + }); + + try (MemoryManager manager = manager(properties, broken)) { + assertThat(manager.prefetchAll(1L, "one")).isEmpty(); + assertThat(manager.prefetchAll(1L, "two")).isEmpty(); + } + assertThat(calls).hasValue(1); + } + + @Test + void unregisterClosesOnlyTheExternalProvider() { + MemoryProvider builtin = provider("builtin", () -> "builtin"); + AtomicBoolean pluginClosed = new AtomicBoolean(); + MemoryProvider plugin = new MemoryProvider() { + @Override public String id() { return "plugin"; } + @Override public void close() { pluginClosed.set(true); } + }; + + try (MemoryManager manager = manager(new MemoryProperties(), builtin)) { + manager.registerPluginProvider(plugin); + manager.unregisterPluginProvider("plugin"); + assertThat(pluginClosed).isTrue(); + assertThat(manager.getProviders()).containsExactly(builtin); + } + } + + private static MemoryProvider provider(String id, java.util.function.Supplier prefetch) { + return new MemoryProvider() { + @Override public String id() { return id; } + @Override public String prefetch(Long agentId, String query, String ownerKey) { + return prefetch.get(); + } + }; + } + + private static MemoryManager manager(MemoryProperties properties, MemoryProvider... providers) { + ObjectProvider noRegistry = new ObjectProvider<>() { + @Override public MeterRegistry getObject(Object... args) { throw new UnsupportedOperationException(); } + @Override public MeterRegistry getIfAvailable() { return null; } + }; + return new MemoryManager(List.of(providers), properties, noRegistry); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/memory/nudge/MemoryNudgeCooldownTest.java b/mateclaw-server/src/test/java/vip/mate/memory/nudge/MemoryNudgeCooldownTest.java new file mode 100644 index 00000000..72edace6 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/memory/nudge/MemoryNudgeCooldownTest.java @@ -0,0 +1,64 @@ +package vip.mate.memory.nudge; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.model.ChatModel; +import org.springframework.ai.chat.model.ChatResponse; +import org.springframework.ai.chat.model.Generation; +import org.springframework.ai.chat.prompt.Prompt; +import vip.mate.agent.AgentGraphBuilder; +import vip.mate.llm.model.ModelConfigEntity; +import vip.mate.llm.service.ModelConfigService; +import vip.mate.memory.MemoryProperties; +import vip.mate.memory.service.StructuredMemoryService; +import vip.mate.workspace.conversation.ConversationService; +import vip.mate.workspace.conversation.model.MessageEntity; + +import java.util.List; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +class MemoryNudgeCooldownTest { + + @Test + void failedParseDoesNotStartCooldown() { + ConversationService conversations = mock(ConversationService.class); + StructuredMemoryService structured = mock(StructuredMemoryService.class); + ModelConfigService models = mock(ModelConfigService.class); + AgentGraphBuilder graphBuilder = mock(AgentGraphBuilder.class); + ChatModel chatModel = mock(ChatModel.class); + MemoryProperties properties = new MemoryProperties(); + properties.setNudgeEnabled(true); + properties.setNudgeTurnInterval(1); + properties.setNudgeCooldownMinutes(60); + when(conversations.listMessages("conversation")).thenReturn(List.of( + message("user", "one"), message("assistant", "two"), + message("user", "three"), message("assistant", "four"))); + when(structured.buildMemoryBlock(1L, null)).thenReturn(""); + when(models.getDefaultModel()).thenReturn(new ModelConfigEntity()); + when(graphBuilder.buildRuntimeChatModel(any())).thenReturn(chatModel); + when(chatModel.call(any(Prompt.class))) + .thenReturn(response("not-json")) + .thenReturn(response("[]")); + MemoryNudgeService service = new MemoryNudgeService(conversations, structured, models, + graphBuilder, properties, new ObjectMapper()); + + service.maybeNudge(1L, "conversation", 4); + service.maybeNudge(1L, "conversation", 4); + + verify(chatModel, times(2)).call(any(Prompt.class)); + } + + private static ChatResponse response(String body) { + return new ChatResponse(List.of(new Generation(new AssistantMessage(body)))); + } + + private static MessageEntity message(String role, String content) { + MessageEntity message = new MessageEntity(); + message.setRole(role); + message.setContent(content); + return message; + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/memory/service/MemoryRecallMigrationTest.java b/mateclaw-server/src/test/java/vip/mate/memory/service/MemoryRecallMigrationTest.java new file mode 100644 index 00000000..a290df30 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/memory/service/MemoryRecallMigrationTest.java @@ -0,0 +1,89 @@ +package vip.mate.memory.service; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.h2.jdbcx.JdbcDataSource; +import org.springframework.core.io.ClassPathResource; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator; + +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class MemoryRecallMigrationTest { + + @ParameterizedTest + @ValueSource(strings = {"mysql", "kingbase"}) + void dialectMigrationAppliesInCompatibleMode(String dialect) { + JdbcDataSource database = new JdbcDataSource(); + String mode = "kingbase".equals(dialect) ? "PostgreSQL" : "MySQL"; + database.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=" + mode + ";DB_CLOSE_DELAY=-1"); + JdbcTemplate jdbc = new JdbcTemplate(database); + createTable(jdbc); + insertDuplicates(jdbc); + + new ResourceDatabasePopulator(new ClassPathResource( + "db/migration/" + dialect + "/V192__memory_recall_unique_identity.sql")).execute(database); + + assertMerged(jdbc); + } + + @Test + void migrationMergesDuplicateCountersAndEnforcesOwnerAwareIdentity() { + EmbeddedDatabase database = new EmbeddedDatabaseBuilder() + .setType(EmbeddedDatabaseType.H2) + .generateUniqueName(true) + .build(); + try { + JdbcTemplate jdbc = new JdbcTemplate(database); + createTable(jdbc); + insertDuplicates(jdbc); + + new ResourceDatabasePopulator(new ClassPathResource( + "db/migration/h2/V192__memory_recall_unique_identity.sql")).execute(database); + + assertMerged(jdbc); + } finally { + database.shutdown(); + } + } + + private static void createTable(JdbcTemplate jdbc) { + jdbc.execute(""" + CREATE TABLE mate_memory_recall ( + id BIGINT PRIMARY KEY, + agent_id BIGINT NOT NULL, + filename VARCHAR(256) NOT NULL, + recall_count INT, + daily_count INT, + last_recalled_at TIMESTAMP, + owner_key VARCHAR(128), + scope VARCHAR(16) NOT NULL, + deleted INT NOT NULL + ) + """); + } + + private static void insertDuplicates(JdbcTemplate jdbc) { + jdbc.update("INSERT INTO mate_memory_recall VALUES (1,7,'MEMORY.md',2,1,TIMESTAMP '2026-01-01 00:00:00',NULL,'TEAM',0)"); + jdbc.update("INSERT INTO mate_memory_recall VALUES (2,7,'MEMORY.md',3,2,TIMESTAMP '2026-02-01 00:00:00','','TEAM',0)"); + jdbc.update("INSERT INTO mate_memory_recall VALUES (3,7,'old.md',9,9,NULL,'','TEAM',1)"); + } + + private static void assertMerged(JdbcTemplate jdbc) { + assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM mate_memory_recall", Integer.class)).isEqualTo(1); + assertThat(jdbc.queryForObject("SELECT recall_count FROM mate_memory_recall", Integer.class)).isEqualTo(5); + assertThat(jdbc.queryForObject("SELECT daily_count FROM mate_memory_recall", Integer.class)).isEqualTo(3); + assertThat(jdbc.queryForObject("SELECT owner_key FROM mate_memory_recall", String.class)).isEmpty(); + assertThatThrownBy(() -> jdbc.update( + "INSERT INTO mate_memory_recall VALUES (4,7,'MEMORY.md',1,1,NULL,'','TEAM',0)")) + .isInstanceOf(DataIntegrityViolationException.class); + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/memory/service/MemoryRecallOwnerIsolationTest.java b/mateclaw-server/src/test/java/vip/mate/memory/service/MemoryRecallOwnerIsolationTest.java index 3dad1752..84568292 100644 --- a/mateclaw-server/src/test/java/vip/mate/memory/service/MemoryRecallOwnerIsolationTest.java +++ b/mateclaw-server/src/test/java/vip/mate/memory/service/MemoryRecallOwnerIsolationTest.java @@ -9,6 +9,7 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import org.springframework.dao.DuplicateKeyException; import vip.mate.memory.MemoryProperties; import vip.mate.memory.identity.MemoryScope; import vip.mate.memory.model.MemoryRecallEntity; @@ -20,13 +21,13 @@ 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.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -77,7 +78,7 @@ class MemoryRecallOwnerIsolationTest { } @Test - @DisplayName("legacy recordRecall overload remains shared and ownerless") + @DisplayName("legacy recordRecall overload remains shared with canonical empty owner") void legacyRecordRecallRemainsShared() { MemoryRecallMapper mapper = mock(MemoryRecallMapper.class); when(mapper.selectOne(any())).thenReturn(null); @@ -87,10 +88,42 @@ class MemoryRecallOwnerIsolationTest { ArgumentCaptor inserted = ArgumentCaptor.forClass(MemoryRecallEntity.class); verify(mapper).insert(inserted.capture()); - assertNull(inserted.getValue().getOwnerKey()); + assertEquals("", inserted.getValue().getOwnerKey()); assertEquals(MemoryScope.TEAM, inserted.getValue().getScope()); } + @Test + @DisplayName("existing recall uses an atomic SQL increment without inserting") + @SuppressWarnings({"rawtypes", "unchecked"}) + void existingRecallIsIncrementedAtomically() { + MemoryRecallMapper mapper = mock(MemoryRecallMapper.class); + when(mapper.update(any(), any())).thenReturn(1); + MemoryRecallService service = new MemoryRecallService(mapper, new MemoryProperties(), new ObjectMapper()); + + service.recordRecall(7L, "MEMORY.md", "shared memory", null); + + ArgumentCaptor> wrapper = + ArgumentCaptor.forClass(com.baomidou.mybatisplus.core.conditions.Wrapper.class); + verify(mapper).update(eq(null), wrapper.capture()); + assertTrue(wrapper.getValue().getSqlSet().contains("recall_count = COALESCE(recall_count, 0) + 1")); + assertTrue(wrapper.getValue().getSqlSet().contains("daily_count = COALESCE(daily_count, 0) + 1")); + verify(mapper, never()).insert(any(MemoryRecallEntity.class)); + } + + @Test + @DisplayName("duplicate insert race retries the atomic update") + void duplicateInsertRetriesIncrement() { + MemoryRecallMapper mapper = mock(MemoryRecallMapper.class); + when(mapper.update(any(), any())).thenReturn(0, 1); + when(mapper.insert(any(MemoryRecallEntity.class))).thenThrow(new DuplicateKeyException("raced")); + MemoryRecallService service = new MemoryRecallService(mapper, new MemoryProperties(), new ObjectMapper()); + + service.recordRecall(7L, "MEMORY.md", "shared memory", null); + + verify(mapper, times(2)).update(any(), any()); + verify(mapper).insert(any(MemoryRecallEntity.class)); + } + @Test @DisplayName("shared Dream candidate query excludes PERSONAL scope") @SuppressWarnings({"rawtypes", "unchecked"}) diff --git a/mateclaw-server/src/test/java/vip/mate/memory/service/MemorySummarizationCooldownTest.java b/mateclaw-server/src/test/java/vip/mate/memory/service/MemorySummarizationCooldownTest.java new file mode 100644 index 00000000..f34a56c1 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/memory/service/MemorySummarizationCooldownTest.java @@ -0,0 +1,88 @@ +package vip.mate.memory.service; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.model.ChatModel; +import org.springframework.ai.chat.model.ChatResponse; +import org.springframework.ai.chat.model.Generation; +import org.springframework.ai.chat.prompt.Prompt; +import vip.mate.agent.AgentGraphBuilder; +import vip.mate.llm.model.ModelConfigEntity; +import vip.mate.llm.service.ModelConfigService; +import vip.mate.memory.MemoryProperties; +import vip.mate.workspace.conversation.ConversationService; +import vip.mate.workspace.conversation.model.MessageEntity; +import vip.mate.workspace.document.WorkspaceFileService; + +import java.util.List; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +class MemorySummarizationCooldownTest { + + private ConversationService conversations; + private AgentGraphBuilder graphBuilder; + private ChatModel chatModel; + private MemorySummarizationService service; + + @BeforeEach + void setUp() { + conversations = mock(ConversationService.class); + WorkspaceFileService files = mock(WorkspaceFileService.class); + ModelConfigService models = mock(ModelConfigService.class); + graphBuilder = mock(AgentGraphBuilder.class); + chatModel = mock(ChatModel.class); + MemoryProperties properties = new MemoryProperties(); + properties.setCooldownMinutes(60); + properties.setMinMessagesForSummarize(4); + when(models.getDefaultModel()).thenReturn(new ModelConfigEntity()); + when(graphBuilder.buildRuntimeChatModel(any())).thenReturn(chatModel); + service = new MemorySummarizationService(conversations, files, models, graphBuilder, + properties, new ObjectMapper(), mock(StructuredMemoryService.class)); + } + + @Test + void explicitRememberBypassesMessageMinimumAndCooldown() { + when(conversations.listMessages("explicit")).thenReturn(List.of( + message("user", "请记住：这个项目默认使用 PostgreSQL"), + message("assistant", "已记录。"))); + when(chatModel.call(any(Prompt.class))).thenReturn(noUpdate()); + + service.analyzeAndUpdateMemory(1L, "explicit"); + service.analyzeAndUpdateMemory(1L, "explicit"); + + verify(chatModel, times(2)).call(any(Prompt.class)); + } + + @Test + void failedAnalysisDoesNotStartCooldown() { + when(conversations.listMessages("normal")).thenReturn(List.of( + message("user", "我长期偏好简洁回答"), + message("assistant", "了解。"), + message("user", "今后都请保持这个风格"), + message("assistant", "好的。"))); + when(chatModel.call(any(Prompt.class))) + .thenThrow(new IllegalStateException("temporary outage")) + .thenReturn(noUpdate()); + + service.analyzeAndUpdateMemory(1L, "normal"); + service.analyzeAndUpdateMemory(1L, "normal"); + + verify(chatModel, times(2)).call(any(Prompt.class)); + } + + private static ChatResponse noUpdate() { + return new ChatResponse(List.of(new Generation( + new AssistantMessage("{\"should_update\":false,\"reason\":\"none\"}")))); + } + + private static MessageEntity message(String role, String content) { + MessageEntity message = new MessageEntity(); + message.setRole(role); + message.setContent(content); + return message; + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/memory/service/MemorySummarizationGateTest.java b/mateclaw-server/src/test/java/vip/mate/memory/service/MemorySummarizationGateTest.java index a0bab816..68bb3720 100644 --- a/mateclaw-server/src/test/java/vip/mate/memory/service/MemorySummarizationGateTest.java +++ b/mateclaw-server/src/test/java/vip/mate/memory/service/MemorySummarizationGateTest.java @@ -63,6 +63,7 @@ class MemorySummarizationGateTest { MemorySummarizationGate.evaluate(List.of(user, assistant)); assertTrue(decision.shouldAnalyze()); + assertTrue(decision.bypassCooldown()); } @Test @@ -127,6 +128,7 @@ class MemorySummarizationGateTest { assertTrue(decision.shouldAnalyze(), "return_direct represents a successful tool-driven answer; should reach analysis"); + assertFalse(decision.bypassCooldown()); } private static MessageEntity message(String role, String content, String metadata) { diff --git a/mateclaw-server/src/test/java/vip/mate/plugin/bridge/PluginMemoryBridgeTest.java b/mateclaw-server/src/test/java/vip/mate/plugin/bridge/PluginMemoryBridgeTest.java index f88b68e1..4d714718 100644 --- a/mateclaw-server/src/test/java/vip/mate/plugin/bridge/PluginMemoryBridgeTest.java +++ b/mateclaw-server/src/test/java/vip/mate/plugin/bridge/PluginMemoryBridgeTest.java @@ -6,6 +6,7 @@ import vip.mate.memory.spi.MemoryProvider; import vip.mate.plugin.api.memory.PluginMemoryProvider; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -205,6 +206,19 @@ class PluginMemoryBridgeTest { assertSame(toolBean, tools.get(0)); } + @Test + @DisplayName("close is forwarded so plugin-owned resources are released on unload") + void closeForwardsToPlugin() { + AtomicBoolean closed = new AtomicBoolean(); + PluginMemoryProvider delegate = new ForwardingPluginProvider(stub()) { + @Override public void close() { closed.set(true); } + }; + + new PluginMemoryBridge(delegate).close(); + + assertTrue(closed.get()); + } + // ---- helpers ---- private static PluginMemoryProvider stub() { @@ -251,5 +265,6 @@ class PluginMemoryBridgeTest { @Override public void onSessionEnd(Long agentId, String conversationId) { delegate.onSessionEnd(agentId, conversationId); } + @Override public void close() { delegate.close(); } } }