mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(memory): thread ownerKey through post-turn sync for per-owner writes
This commit is contained in:
mateaix
parent
f9fcf35dc7
commit
2ed0d7d04f
@ -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.
|
||||
* <p>
|
||||
* 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.
|
||||
*/
|
||||
|
||||
@ -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.</li>
|
||||
* <li>{@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.</li>
|
||||
* <li>{@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.</li>
|
||||
* <li>{@code getToolBeans} — empty (no agent-facing tools in v1)</li>
|
||||
* </ul>
|
||||
*
|
||||
@ -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);
|
||||
}
|
||||
|
||||
@ -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<String> 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();
|
||||
}
|
||||
|
||||
@ -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);
|
||||
|
||||
@ -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());
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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<Object> 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); }
|
||||
|
||||
@ -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();
|
||||
|
||||
@ -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;
|
||||
|
||||
@ -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<Object> getToolBeans() {
|
||||
List<Object> beans = delegate.getToolBeans();
|
||||
|
||||
@ -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.
|
||||
|
||||
|
||||
@ -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 的结果。
|
||||
|
||||
|
||||
@ -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<String> receivedOwner = new AtomicReference<>("sentinel");
|
||||
AtomicReference<String> 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<String> 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() {
|
||||
|
||||
@ -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");
|
||||
|
||||
@ -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<Object> 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());
|
||||
}
|
||||
}
|
||||
|
||||
@ -111,6 +111,50 @@ class PluginMemoryBridgeTest {
|
||||
"not crash with AbstractMethodError");
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("five-arg syncTurn forwards ownerKey verbatim to the delegate")
|
||||
void fiveArgSyncTurnForwardsOwnerKey() {
|
||||
AtomicReference<String> 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<String> 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<Object> getToolBeans() { return delegate.getToolBeans(); }
|
||||
@Override public void onSessionEnd(Long agentId, String conversationId) {
|
||||
delegate.onSessionEnd(agentId, conversationId);
|
||||
|
||||
Loading…
Reference in New Issue
Block a user