From 2ed0d7d04f45bec92f49aa4c3c0d0023630d65b7 Mon Sep 17 00:00:00 2001 From: mateaix <57164338+mateaix@users.noreply.github.com> Date: Sun, 26 Jul 2026 11:07:04 +0800 Subject: [PATCH] feat(memory): thread ownerKey through post-turn sync for per-owner writes --- .../api/memory/PluginMemoryProvider.java | 22 +++++++ .../vip/mate/plugin/mem0/Mem0Provider.java | 36 +++++++----- .../mate/plugin/mem0/Mem0ProviderTest.java | 37 ++++++++++-- .../lifecycle/MemoryLifecycleMediator.java | 2 +- .../vip/mate/memory/spi/MemoryManager.java | 12 +++- .../vip/mate/memory/spi/MemoryProvider.java | 18 ++++++ .../decorator/MemoryProviderDecorator.java | 3 + .../spi/decorator/MetricsMemoryProvider.java | 7 ++- .../decorator/RetryableMemoryProvider.java | 7 ++- .../plugin/bridge/PluginMemoryBridge.java | 8 +++ .../src/main/resources/docs/en/memory.md | 6 +- .../src/main/resources/docs/zh/memory.md | 6 +- .../MemoryManagerPluginPrefetchTest.java | 58 +++++++++++++++++++ .../lifecycle/LifecycleFlagGuardTest.java | 6 +- .../MemoryLifecycleMediatorTest.java | 6 +- .../plugin/bridge/PluginMemoryBridgeTest.java | 48 +++++++++++++++ 16 files changed, 245 insertions(+), 37 deletions(-) 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 128c185a..b4b47833 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 @@ -80,6 +80,28 @@ public interface PluginMemoryProvider { String userMessage, String assistantReply) { } + /** + * Post-turn sync with per-owner isolation. Called by the platform with the + * same {@code ownerKey} that was resolved for this turn's prefetch, so + * providers can persist the turn under the same per-user identifier they + * recall by. + *

+ * Default implementation degrades to the four-arg variant, dropping the + * owner key. External providers that isolate memory per end-user should + * override this so that written memories stay reachable by owner-scoped + * recall. + * + * @param agentId the agent ID + * @param conversationId the conversation ID + * @param userMessage user's message text + * @param assistantReply assistant's reply text + * @param ownerKey memory owner key (e.g. {@code "user:42"}), or null if unknown + */ + default void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { + syncTurn(agentId, conversationId, userMessage, assistantReply); + } + /** * Tool beans this provider wants to expose to the agent. */ 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 95ea8df7..fc42efd6 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 @@ -19,9 +19,13 @@ import java.util.concurrent.Executors; * 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. - *

  • {@code syncTurn} — when {@code syncEnabled} and {@code ownerKey} is - * non-blank, asynchronously pushes the turn to {@code POST /memories/}. - * Failures are logged and swallowed; never blocks the response path.
    • + *
  • {@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 + * 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)
    • * * @@ -119,19 +123,20 @@ class Mem0Provider implements PluginMemoryProvider { @Override public void syncTurn(Long agentId, String conversationId, String userMessage, String assistantReply) { + // Four-arg variant: no owner key → skip. Mem0 keys memories by user_id; + // writing under any fallback identifier (e.g. agentId) would store + // memories that owner-scoped prefetch can never recall. + } + + @Override + public void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { if (!config.syncEnabled()) { return; } - // ownerKey is NOT available in the two-arg syncTurn signature - // (PlatformMemoryProvider only passes agentId + conversationId + messages). - // We push the turn using agentId as the user_id fallback — this is - // weaker isolation than prefetch (which has ownerKey), but better than - // dropping the turn. If users need strict per-owner sync, configure - // syncEnabled=false and rely on prefetch-only recall. - // NOTE: this is a known v1 limitation; a future SPI extension would - // pass ownerKey into syncTurn as well. - String userId = agentId == null ? null : agentId.toString(); - if (userId == null || userId.isBlank()) { + if (ownerKey == null || ownerKey.isBlank()) { + // Same guard as prefetch: Mem0 requires user_id; without the owner + // key the write would break per-owner isolation. return; } if ((userMessage == null || userMessage.isBlank()) @@ -140,10 +145,11 @@ class Mem0Provider implements PluginMemoryProvider { } CompletableFuture.runAsync(() -> { try { - client.addMemories(userId, agentId == null ? null : agentId.toString(), + client.addMemories(ownerKey, agentId == null ? null : agentId.toString(), conversationId, userMessage, assistantReply); } catch (Exception e) { - log.debug("[Mem0] syncTurn failed for agent={}: {}", agentId, e.getMessage()); + log.debug("[Mem0] syncTurn failed for agent={} owner={}: {}", + agentId, ownerKey, e.getMessage()); } }, async); } 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 dceedf42..0a84d40d 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 @@ -13,6 +13,7 @@ import java.io.InputStream; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import static org.assertj.core.api.Assertions.assertThat; @@ -22,11 +23,13 @@ class Mem0ProviderTest { private Mem0Provider provider; private final AtomicInteger addCount = new AtomicInteger(); private final AtomicInteger searchCount = new AtomicInteger(); + private final AtomicReference lastAddBody = new AtomicReference<>(); @BeforeEach void setUp() throws IOException { addCount.set(0); searchCount.set(0); + lastAddBody.set(null); HttpHandler handler = this::handle; server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); server.createContext("/", handler); @@ -44,13 +47,15 @@ class Mem0ProviderTest { } private void handle(HttpExchange exchange) throws IOException { + String body; try (InputStream in = exchange.getRequestBody()) { - in.readAllBytes(); // drain + body = new String(in.readAllBytes(), StandardCharsets.UTF_8); } String path = exchange.getRequestURI().getPath(); byte[] resp; if ("/memories/".equals(path)) { addCount.incrementAndGet(); + lastAddBody.set(body); resp = "{\"results\":[]}".getBytes(StandardCharsets.UTF_8); } else if ("/memories/search/".equals(path)) { searchCount.incrementAndGet(); @@ -137,8 +142,8 @@ class Mem0ProviderTest { } @Test - void syncTurn_pushesAsynchronouslyWithoutBlocking() throws Exception { - provider.syncTurn(1L, "conv-1", "hello", "world"); + void syncTurn_pushesAsynchronouslyWithOwnerKeyAsUserId() throws Exception { + provider.syncTurn(1L, "conv-1", "hello", "world", "user:42"); // Wait briefly for the async executor to fire the POST. long deadline = System.currentTimeMillis() + 2000; @@ -146,11 +151,31 @@ class Mem0ProviderTest { Thread.sleep(20); } assertThat(addCount.get()).isEqualTo(1); + // The write must land under the same user_id that prefetch recalls by. + assertThat(lastAddBody.get()).contains("\"user_id\":\"user:42\""); + assertThat(lastAddBody.get()).contains("\"agent_id\":\"1\""); + } + + @Test + void fourArgSyncTurn_skipsBecauseNoOwnerKey() throws Exception { + // Without ownerKey, a write would be keyed by an identifier that + // owner-scoped prefetch never queries; the provider must skip. + provider.syncTurn(1L, "conv-1", "hello", "world"); + Thread.sleep(200); // give async a chance to (not) fire + assertThat(addCount.get()).isZero(); + } + + @Test + void syncTurn_skipsWhenOwnerKeyBlank() throws Exception { + provider.syncTurn(1L, "conv-1", "hello", "world", ""); + provider.syncTurn(1L, "conv-1", "hello", "world", null); + Thread.sleep(200); + assertThat(addCount.get()).isZero(); } @Test void syncTurn_skipsWhenBothMessagesBlank() throws Exception { - provider.syncTurn(1L, "conv-1", " ", ""); + provider.syncTurn(1L, "conv-1", " ", "", "user:42"); Thread.sleep(200); // give async a chance to (not) fire assertThat(addCount.get()).isZero(); } @@ -165,7 +190,7 @@ class Mem0ProviderTest { server.createContext("/", ex -> { ex.sendResponseHeaders(200, 0); ex.close(); }); // Note: client still points at the old port → connection refused. - provider.syncTurn(1L, "conv-1", "hi", "there"); + provider.syncTurn(1L, "conv-1", "hi", "there", "user:42"); Thread.sleep(500); // No exception thrown; nothing to assert beyond "test didn't blow up". } @@ -177,7 +202,7 @@ class Mem0ProviderTest { "http://127.0.0.1:" + server.getAddress().getPort(), null, true, false, 3, 3000); Mem0Provider p = new Mem0Provider(cfg, new Mem0Client(cfg), LoggerFactory.getLogger("test")); - p.syncTurn(1L, "conv-1", "hi", "there"); + p.syncTurn(1L, "conv-1", "hi", "there", "user:42"); Thread.sleep(200); assertThat(addCount.get()).isZero(); } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/lifecycle/MemoryLifecycleMediator.java b/mateclaw-server/src/main/java/vip/mate/memory/lifecycle/MemoryLifecycleMediator.java index f52f28a6..c9625b09 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/lifecycle/MemoryLifecycleMediator.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/lifecycle/MemoryLifecycleMediator.java @@ -61,7 +61,7 @@ public class MemoryLifecycleMediator { public void afterLlmCall(TurnContext ctx, String assistantReply) { try { memoryManager.syncAll(ctx.agentId(), ctx.conversationId(), - ctx.userQuery(), assistantReply); + ctx.userQuery(), assistantReply, ctx.ownerKey()); events.publishEvent(new TurnCompletedEvent(ctx, assistantReply)); log.debug("[Memory] afterLlmCall: agent={}, conv={}, replyLen={}", ctx.agentId(), ctx.conversationId(), assistantReply != null ? assistantReply.length() : 0); 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 bbbcf0b7..c5c8e302 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 @@ -189,9 +189,19 @@ public class MemoryManager { */ public void syncAll(Long agentId, String conversationId, String userMessage, String assistantReply) { + syncAll(agentId, conversationId, userMessage, assistantReply, null); + } + + /** + * Owner-scoped post-turn sync. Passes the same resolved memory + * {@code ownerKey} that prefetch used, so owner-aware providers persist + * the turn under the identifier their recall path queries by. + */ + public void syncAll(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { for (MemoryProvider provider : providers) { try { - provider.syncTurn(agentId, conversationId, userMessage, assistantReply); + provider.syncTurn(agentId, conversationId, userMessage, assistantReply, ownerKey); } catch (Exception e) { log.warn("[MemoryManager] Provider '{}' syncTurn failed: {}", provider.id(), e.getMessage()); 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 00595074..262c0c32 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 @@ -87,6 +87,24 @@ public interface MemoryProvider { String userMessage, String assistantReply) { } + /** + * Owner-scoped post-turn sync. Providers that isolate memory per end-user + * override this to persist the turn under the same {@code ownerKey} that + * owner-scoped prefetch recalls by. Default delegates to + * {@link #syncTurn(Long, String, String, String)} for providers that are + * not owner-aware. + * + * @param agentId the agent ID + * @param conversationId the conversation ID + * @param userMessage user's message text + * @param assistantReply assistant's reply text + * @param ownerKey resolved memory owner key (e.g. "user:42"); may be null + */ + default void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { + syncTurn(agentId, conversationId, userMessage, assistantReply); + } + /** * Spring AI @Tool beans this provider wants to expose to the agent. * These are collected by MemoryManager and added to the tool set. 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 1617c0be..d7480fa0 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 @@ -28,6 +28,9 @@ public abstract class MemoryProviderDecorator implements MemoryProvider { @Override public void syncTurn(Long agentId, String conversationId, String userMessage, String assistantReply) { delegate.syncTurn(agentId, conversationId, userMessage, assistantReply); } + @Override public void syncTurn(Long agentId, String conversationId, String userMessage, String assistantReply, String ownerKey) { + delegate.syncTurn(agentId, conversationId, userMessage, assistantReply, ownerKey); + } @Override public List getToolBeans() { return delegate.getToolBeans(); } @Override public void onSessionEnd(Long agentId, String conversationId) { delegate.onSessionEnd(agentId, conversationId); } @Override public String onPreCompress(Long agentId, List messages) { return delegate.onPreCompress(agentId, messages); } diff --git a/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/MetricsMemoryProvider.java b/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/MetricsMemoryProvider.java index 04d76382..56ce7f98 100644 --- a/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/MetricsMemoryProvider.java +++ b/mateclaw-server/src/main/java/vip/mate/memory/spi/decorator/MetricsMemoryProvider.java @@ -58,9 +58,14 @@ public class MetricsMemoryProvider extends MemoryProviderDecorator { @Override public void syncTurn(Long agentId, String conversationId, String userMessage, String assistantReply) { + syncTurn(agentId, conversationId, userMessage, assistantReply, null); + } + + @Override + public void syncTurn(Long agentId, String conversationId, String userMessage, String assistantReply, String ownerKey) { syncTimer.record(() -> { try { - delegate.syncTurn(agentId, conversationId, userMessage, assistantReply); + delegate.syncTurn(agentId, conversationId, userMessage, assistantReply, ownerKey); } catch (Exception e) { meterRegistry.counter("memory.sync.failures", "provider", delegate.id()).increment(); 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 fd99dd1c..ac584215 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 @@ -45,10 +45,15 @@ public class RetryableMemoryProvider extends MemoryProviderDecorator { @Override public void syncTurn(Long agentId, String conversationId, String userMessage, String assistantReply) { + syncTurn(agentId, conversationId, userMessage, assistantReply, null); + } + + @Override + public void syncTurn(Long agentId, String conversationId, String userMessage, String assistantReply, String ownerKey) { Exception lastException = null; for (int attempt = 1; attempt <= maxAttempts; attempt++) { try { - delegate.syncTurn(agentId, conversationId, userMessage, assistantReply); + delegate.syncTurn(agentId, conversationId, userMessage, assistantReply, ownerKey); return; } catch (Exception e) { lastException = 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 b1e9f542..b6c9d581 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 @@ -58,6 +58,14 @@ public class PluginMemoryBridge implements MemoryProvider { delegate.syncTurn(agentId, conversationId, userMessage, assistantReply); } + @Override + public void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { + // Forward ownerKey to the plugin provider; plugins that don't override the + // five-arg variant fall back to the four-arg default (ownerKey dropped). + delegate.syncTurn(agentId, conversationId, userMessage, assistantReply, ownerKey); + } + @Override public List getToolBeans() { List beans = delegate.getToolBeans(); diff --git a/mateclaw-server/src/main/resources/docs/en/memory.md b/mateclaw-server/src/main/resources/docs/en/memory.md index 5eb1d9c9..ca93879b 100644 --- a/mateclaw-server/src/main/resources/docs/en/memory.md +++ b/mateclaw-server/src/main/resources/docs/en/memory.md @@ -583,7 +583,7 @@ Mem0 integration is an **optional community contribution** — it is NOT part of |------|----------| | `systemPromptBlock` | Returns empty — leaves the resident system prompt alone, avoids per-turn token bloat | | `prefetch(agentId, query, ownerKey)` | When `searchEnabled=true` and `ownerKey` is non-blank, calls `POST {baseUrl}/memories/search/` and returns a `[Mem0 Recall]` block concatenated into the current turn's context | -| `syncTurn(agentId, conversationId, userMessage, assistantReply)` | When `syncEnabled=true`, **asynchronously** pushes this turn's user/assistant messages to `POST {baseUrl}/memories/`. Failures are logged only, never block the response | +| `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. @@ -597,7 +597,7 @@ Mem0 isolates by `user_id` + `agent_id`. MateClaw maps them as: | `ownerKey` (e.g. `user:42` / `feishu:sender_abc`) | `user_id` | Passed through verbatim | | `agentId` | `agent_id` | The digital employee ID | -Only the three-arg `prefetch` variant receives `ownerKey`. The two-arg variant (no ownerKey) returns empty — Mem0 requires `user_id`, without it isolation is impossible. +Both `prefetch` and `syncTurn` receive `ownerKey` from the platform, so writes and recalls are keyed by the same `user_id`. The variants without `ownerKey` skip (empty recall / dropped write) — Mem0 requires `user_id`, without it isolation is impossible. ### Installation @@ -621,7 +621,7 @@ Config is read once at plugin load — changes require a plugin reload to take e ### Known limitations (v1) -- **`syncTurn` has no `ownerKey`**: the plugin SPI's `syncTurn` signature is only `(agentId, conversationId, userMessage, assistantReply)`, so when pushing to Mem0 the plugin falls back to using `agentId` as `user_id`. This is coarser isolation than prefetch (which has ownerKey). If you need strict per-owner sync, set `syncEnabled=false` and rely on prefetch-only recall, with writes handled by your own Mem0 client. +- **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. - **No token budget control**: the `[Mem0 Recall]` block returned by prefetch is concatenated into the context directly — it is NOT subject to the `system-block-max-chars` injection budget (that budget only governs `user`/`feedback` structured entries). `maxResults` is the only size knob. - **No agent tools**: v1 does not expose `mem0_search` / `mem0_add` style tools for the agent to call proactively. The agent only passively receives prefetch results. diff --git a/mateclaw-server/src/main/resources/docs/zh/memory.md b/mateclaw-server/src/main/resources/docs/zh/memory.md index d457c185..8a152b29 100644 --- a/mateclaw-server/src/main/resources/docs/zh/memory.md +++ b/mateclaw-server/src/main/resources/docs/zh/memory.md @@ -577,7 +577,7 @@ Mem0 集成是**可选的社区贡献项**，不在 MateClaw 的默认安装里 |------|------| | `systemPromptBlock` | 返回空——常驻 system prompt 不动，避免每轮 token 膨胀 | | `prefetch(agentId, query, ownerKey)` | 当 `searchEnabled=true` 且 `ownerKey` 非空时，调 `POST {baseUrl}/memories/search/`，返回一个 `[Mem0 Recall]` 块拼进本轮上下文 | -| `syncTurn(agentId, conversationId, userMessage, assistantReply)` | 当 `syncEnabled=true` 时，**异步**把这一轮的 user/assistant 消息推到 `POST {baseUrl}/memories/`，失败只记日志、不阻塞响应 | +| `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 的本地记忆。 @@ -591,7 +591,7 @@ Mem0 用 `user_id` + `agent_id` 做隔离。MateClaw 的映射： | `ownerKey`（如 `user:42` / `feishu:sender_abc`） | `user_id` | 透传，原样作为 user_id | | `agentId` | `agent_id` | 数字员工 ID | -只有 `prefetch` 的三参版能拿到 `ownerKey`。两参版（无 ownerKey）会直接返回空——Mem0 要求 `user_id`，没它无法隔离。 +`prefetch` 和 `syncTurn` 都能从平台拿到 `ownerKey`，写入和召回用同一个 `user_id`。拿不到 `ownerKey` 的变体会直接跳过（召回返回空 / 放弃写入）——Mem0 要求 `user_id`，没它无法隔离。 ### 安装步骤 @@ -615,7 +615,7 @@ Mem0 用 `user_id` + `agent_id` 做隔离。MateClaw 的映射： ### 已知限制（v1） -- **`syncTurn` 拿不到 `ownerKey`**：插件 SPI 的 `syncTurn` 签名只有 `(agentId, conversationId, userMessage, assistantReply)`，所以推送 Mem0 时只能用 `agentId` 作为 `user_id` 降级。这比 prefetch（有 ownerKey）的隔离粒度粗。如果你需要严格的 per-owner 同步，把 `syncEnabled=false`，只依赖 prefetch 做召回，由你自己的 Mem0 客户端负责写入。 +- **没有解析出 owner 的轮次不会同步**：`syncTurn` 要求 `ownerKey`；平台解析不出 owner 的轮次（如系统触发的运行）会直接跳过，而不是用一个召回永远查不到的降级标识写入。 - **没有 token 预算控制**：prefetch 返回的 `[Mem0 Recall]` 块直接拼进上下文，不受 `system-block-max-chars` 那套注入预算约束（那套只管 `user`/`feedback` 结构化条目）。`maxResults` 是唯一的尺寸闸门。 - **没有 Agent 工具**：v1 不暴露 `mem0_search` / `mem0_add` 之类的工具给 Agent 主动调用。Agent 只能被动接收 prefetch 的结果。 diff --git a/mateclaw-server/src/test/java/vip/mate/memory/MemoryManagerPluginPrefetchTest.java b/mateclaw-server/src/test/java/vip/mate/memory/MemoryManagerPluginPrefetchTest.java index d9eaecbb..bee33b59 100644 --- a/mateclaw-server/src/test/java/vip/mate/memory/MemoryManagerPluginPrefetchTest.java +++ b/mateclaw-server/src/test/java/vip/mate/memory/MemoryManagerPluginPrefetchTest.java @@ -189,6 +189,64 @@ class MemoryManagerPluginPrefetchTest { assertFalse(result.contains("[Plugin Recall]")); } + @Test + @DisplayName("syncAll with ownerKey reaches the plugin's five-arg syncTurn verbatim") + void syncAllForwardsOwnerKeyToPlugin() { + AtomicReference receivedOwner = new AtomicReference<>("sentinel"); + AtomicReference receivedConversation = new AtomicReference<>(); + + PluginMemoryProvider plugin = new PluginMemoryProvider() { + @Override + public String id() { return "test-plugin-mem"; } + + @Override + public boolean isAvailable() { return true; } + + @Override + public void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { + receivedOwner.set(ownerKey); + receivedConversation.set(conversationId); + } + }; + + MemoryManager manager = newManager(new PluginMemoryBridge(plugin)); + manager.syncAll(7L, "conv-1", "hello", "world", "user:42"); + + assertEquals("user:42", receivedOwner.get(), + "ownerKey must reach the plugin's five-arg syncTurn — synced memories " + + "must be keyed by the same identifier prefetch recalls by"); + assertEquals("conv-1", receivedConversation.get()); + } + + @Test + @DisplayName("syncAll without ownerKey delegates to the five-arg path with null — " + + "owner-aware plugins see null and can opt out of the write") + void syncAllWithoutOwnerKeyPassesNullToFiveArg() { + AtomicReference receivedOwner = new AtomicReference<>("sentinel"); + PluginMemoryProvider plugin = new PluginMemoryProvider() { + @Override + public String id() { return "test-plugin-mem"; } + + @Override + public boolean isAvailable() { return true; } + + @Override + public void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { + receivedOwner.set(ownerKey); + } + }; + + MemoryManager manager = newManager(new PluginMemoryBridge(plugin)); + manager.syncAll(7L, "conv-1", "hello", "world"); + + assertEquals(null, receivedOwner.get(), + "four-arg syncAll must surface as null ownerKey to the plugin's " + + "five-arg variant — the contract signal that lets owner-aware " + + "plugins skip writes they could never recall"); + } + // ---- helpers ---- private static MemoryProvider stubBuiltin() { diff --git a/mateclaw-server/src/test/java/vip/mate/memory/lifecycle/LifecycleFlagGuardTest.java b/mateclaw-server/src/test/java/vip/mate/memory/lifecycle/LifecycleFlagGuardTest.java index b35e115f..0ef1c6d3 100644 --- a/mateclaw-server/src/test/java/vip/mate/memory/lifecycle/LifecycleFlagGuardTest.java +++ b/mateclaw-server/src/test/java/vip/mate/memory/lifecycle/LifecycleFlagGuardTest.java @@ -57,7 +57,7 @@ class LifecycleFlagGuardTest { } verify(memoryManager, never()).prefetchAll(any(), any(), any()); - verify(memoryManager, never()).syncAll(any(), any(), any(), any()); + verify(memoryManager, never()).syncAll(any(), any(), any(), any(), any()); verify(memoryManager, never()).onSessionEnd(any(), any()); } @@ -99,7 +99,7 @@ class LifecycleFlagGuardTest { mediator.afterLlmCall(new TurnContext(1L, "c1", "s1", i, "hello"), "reply-" + i); } - verify(memoryManager, times(10)).syncAll(eq(1L), eq("c1"), eq("hello"), anyString()); + verify(memoryManager, times(10)).syncAll(eq(1L), eq("c1"), eq("hello"), anyString(), any()); } @Test @@ -143,7 +143,7 @@ class LifecycleFlagGuardTest { @DisplayName("Provider exception in syncAll degrades gracefully (no throw)") void syncException_graceful() { org.mockito.Mockito.doThrow(new RuntimeException("boom")) - .when(memoryManager).syncAll(any(), any(), any(), any()); + .when(memoryManager).syncAll(any(), any(), any(), any(), any()); // Should not throw mediator.afterLlmCall(new TurnContext(1L, "c1", "s1", 1, "q"), "reply"); diff --git a/mateclaw-server/src/test/java/vip/mate/memory/lifecycle/MemoryLifecycleMediatorTest.java b/mateclaw-server/src/test/java/vip/mate/memory/lifecycle/MemoryLifecycleMediatorTest.java index f74049f8..4318a69e 100644 --- a/mateclaw-server/src/test/java/vip/mate/memory/lifecycle/MemoryLifecycleMediatorTest.java +++ b/mateclaw-server/src/test/java/vip/mate/memory/lifecycle/MemoryLifecycleMediatorTest.java @@ -72,7 +72,7 @@ class MemoryLifecycleMediatorTest { TurnContext ctx = new TurnContext(1L, "c1", "s1", 1, "hello"); mediator.afterLlmCall(ctx, "reply text"); - verify(memoryManager).syncAll(1L, "c1", "hello", "reply text"); + verify(memoryManager).syncAll(1L, "c1", "hello", "reply text", null); ArgumentCaptor eventCaptor = ArgumentCaptor.forClass(Object.class); verify(eventPublisher).publishEvent(eventCaptor.capture()); @@ -107,7 +107,7 @@ class MemoryLifecycleMediatorTest { @DisplayName("afterLlmCall swallows syncAll exceptions") void afterLlmCall_exceptionSwallowed() { doThrow(new RuntimeException("sync failed")) - .when(memoryManager).syncAll(any(), any(), any(), any()); + .when(memoryManager).syncAll(any(), any(), any(), any(), any()); // Should not throw mediator.afterLlmCall(new TurnContext(1L, "c1", "s1", 1, "q"), "reply"); @@ -150,6 +150,6 @@ class MemoryLifecycleMediatorTest { } verify(memoryManager, times(5)).prefetchAll(eq(1L), any(), any()); - verify(memoryManager, times(5)).syncAll(eq(1L), eq("c1"), any(), any()); + verify(memoryManager, times(5)).syncAll(eq(1L), eq("c1"), any(), any(), any()); } } 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 30789253..f88b68e1 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 @@ -111,6 +111,50 @@ class PluginMemoryBridgeTest { "not crash with AbstractMethodError"); } + @Test + @DisplayName("five-arg syncTurn forwards ownerKey verbatim to the delegate") + void fiveArgSyncTurnForwardsOwnerKey() { + AtomicReference receivedOwner = new AtomicReference<>("not-called"); + PluginMemoryProvider delegate = new ForwardingPluginProvider(stub()) { + @Override + public void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { + receivedOwner.set(ownerKey); + } + }; + + PluginMemoryBridge bridge = new PluginMemoryBridge(delegate); + bridge.syncTurn(42L, "conv-1", "hello", "world", "user:42"); + + assertEquals("user:42", receivedOwner.get(), + "ownerKey must reach the delegate — synced memories must be keyed by " + + "the same identifier owner-scoped prefetch recalls by"); + } + + @Test + @DisplayName("when the plugin does not override the five-arg syncTurn, " + + "the SPI default degrades to four-arg (ownerKey dropped, not crashed)") + void fiveArgSyncTurnDegradesToFourArgWhenPluginDoesNotOverride() { + AtomicReference fourArgCalled = new AtomicReference<>("not-called"); + PluginMemoryProvider pluginOnlyFourArg = new PluginMemoryProvider() { + @Override + public String id() { return "four-arg-only"; } + + @Override + public void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply) { + fourArgCalled.set("called"); + } + }; + + PluginMemoryBridge bridge = new PluginMemoryBridge(pluginOnlyFourArg); + bridge.syncTurn(42L, "conv-1", "hello", "world", "user:42"); + + assertEquals("called", fourArgCalled.get(), + "SPI default should drop ownerKey and route to the four-arg impl, " + + "not crash with AbstractMethodError"); + } + @Test @DisplayName("metadata (id/order/isAvailable) and lifecycle hooks pass through") void metadataAndLifecyclePassthrough() { @@ -199,6 +243,10 @@ class PluginMemoryBridgeTest { String userMessage, String assistantReply) { delegate.syncTurn(agentId, conversationId, userMessage, assistantReply); } + @Override public void syncTurn(Long agentId, String conversationId, + String userMessage, String assistantReply, String ownerKey) { + delegate.syncTurn(agentId, conversationId, userMessage, assistantReply, ownerKey); + } @Override public List getToolBeans() { return delegate.getToolBeans(); } @Override public void onSessionEnd(Long agentId, String conversationId) { delegate.onSessionEnd(agentId, conversationId);