From 97ccb96e41dea05dc0a4b74645808b058baa114a Mon Sep 17 00:00:00 2001 From: saddddboi <99187163+saddddboi@users.noreply.github.com> Date: Mon, 12 Oct 2026 09:21:01 +0800 Subject: [PATCH 1/2] fix(harness): isolate memory consolidation watermarks --- .../agent/memory/MemoryConsolidator.java | 29 ++++- .../MemoryConsolidatorFilesystemTest.java | 113 ++++++++++++++++++ docs/v2/en/docs/harness/memory.md | 2 + docs/v2/zh/docs/harness/memory.md | 2 + 4 files changed, 140 insertions(+), 6 deletions(-) diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java index e776323f32..ee57b0bb87 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java @@ -309,7 +309,7 @@ Instant readWatermark(RuntimeContext rc) { BaseStore store = this.baseStore; if (store != null) { try { - StoreItem item = store.get(WATERMARK_NAMESPACE, WATERMARK_KEY); + StoreItem item = store.get(watermarkNamespace(rc), WATERMARK_KEY); if (item != null && item.value() != null) { Object ts = item.value().get("ts"); if (ts instanceof Number n) { @@ -344,7 +344,7 @@ private Instant readWatermarkFromFile(RuntimeContext rc) { private void writeWatermark(RuntimeContext rc, Instant ts) { if (baseStore != null) { - writeWatermarkToStore(ts); + writeWatermarkToStore(rc, ts); } try { workspaceManager.writeUtf8WorkspaceRelative(rc, STATE_REL_PATH, ts.toString()); @@ -356,21 +356,38 @@ private void writeWatermark(RuntimeContext rc, Instant ts) { } } - private void writeWatermarkToStore(Instant ts) { + private List watermarkNamespace(RuntimeContext rc) { + var factory = workspaceManager.getNamespaceFactory(); + List scope = + factory == null + ? List.of() + : factory.getNamespace(rc != null ? rc : RuntimeContext.empty()); + if (scope == null || scope.isEmpty()) { + return WATERMARK_NAMESPACE; + } + // Keep intentionally unscoped cursors compatible, but never import the old global + // cursor into an isolated scope: it may describe another user's processed ledgers. + List namespace = new ArrayList<>(WATERMARK_NAMESPACE); + namespace.add("scopes"); + namespace.addAll(scope); + return List.copyOf(namespace); + } + + private void writeWatermarkToStore(RuntimeContext rc, Instant ts) { + List namespace = watermarkNamespace(rc); Map value = new HashMap<>(); value.put("ts", ts.toEpochMilli()); for (int attempt = 0; attempt < MAX_CAS_RETRIES; attempt++) { StoreItem item; try { - item = baseStore.get(WATERMARK_NAMESPACE, WATERMARK_KEY); + item = baseStore.get(namespace, WATERMARK_KEY); } catch (RuntimeException e) { log.warn("Failed to read consolidation watermark for CAS: {}", e.getMessage()); return; } long expectedVersion = item != null ? item.version() : 0L; try { - if (baseStore.putIfVersion( - WATERMARK_NAMESPACE, WATERMARK_KEY, value, expectedVersion)) { + if (baseStore.putIfVersion(namespace, WATERMARK_KEY, value, expectedVersion)) { return; } } catch (RuntimeException e) { diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java index 3fdc7b0186..0b671bb545 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java @@ -18,18 +18,34 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import io.agentscope.core.agent.RuntimeContext; +import io.agentscope.core.message.TextBlock; +import io.agentscope.core.model.ChatResponse; +import io.agentscope.core.model.Model; +import io.agentscope.harness.agent.IsolationScope; import io.agentscope.harness.agent.filesystem.remote.RemoteFilesystem; import io.agentscope.harness.agent.filesystem.remote.store.InMemoryStore; +import io.agentscope.harness.agent.filesystem.remote.store.NamespaceFactory; import io.agentscope.harness.agent.workspace.WorkspaceManager; import java.nio.file.Files; import java.nio.file.Path; import java.time.Instant; +import java.util.ArrayList; import java.util.List; import java.util.Map; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.EnumSource; +import reactor.core.publisher.Flux; /** * Verifies that {@link MemoryConsolidator} reads daily ledgers and writes watermark / MEMORY.md @@ -38,6 +54,103 @@ */ class MemoryConsolidatorFilesystemTest { + @ParameterizedTest + @CsvSource({"false, USER", "true, USER", "false, SESSION", "true, SESSION"}) + void consolidationOfOneScopeDoesNotSkipAnotherScopesLedger( + boolean sharedWatermarkStore, IsolationScope scope, @TempDir Path tmp) { + InMemoryStore store = new InMemoryStore(); + NamespaceFactory namespaces = scope.toNamespaceFactory(); + RemoteFilesystem fs = new RemoteFilesystem(store, namespaces); + RuntimeContext alice = RuntimeContext.builder().userId("alice").sessionId("a").build(); + RuntimeContext bob = + RuntimeContext.builder() + .userId(scope == IsolationScope.SESSION ? "alice" : "bob") + .sessionId("b") + .build(); + Instant ledgerTime = Instant.now().minusSeconds(3600); + seedStoreFile( + store, + namespaces.getNamespace(alice), + "memory/2026-10-11.md", + "Alice fact", + ledgerTime); + seedStoreFile( + store, + namespaces.getNamespace(bob), + "memory/2026-10-11.md", + "Bob fact", + ledgerTime); + seedStoreFile( + store, namespaces.getNamespace(bob), "MEMORY.md", "previous Bob fact", ledgerTime); + Model model = mock(Model.class); + when(model.stream(anyList(), isNull(), isNull())) + .thenReturn( + Flux.just( + ChatResponse.builder() + .content( + List.of( + TextBlock.builder() + .text("curated") + .build())) + .build())); + + try (WorkspaceManager wsm = new WorkspaceManager(tmp, fs, null, namespaces)) { + MemoryConsolidator consolidator = + new MemoryConsolidator( + wsm, model, null, 4000, sharedWatermarkStore ? store : null); + consolidator.consolidate(alice).block(); + assertEquals("curated", wsm.readMemoryMd(alice)); + assertEquals("previous Bob fact", wsm.readMemoryMd(bob)); + + consolidator.consolidate(bob).block(); + assertEquals("curated", wsm.readMemoryMd(bob)); + consolidator.consolidate(alice).block(); + consolidator.consolidate(bob).block(); + verify(model, times(2)).stream(anyList(), isNull(), isNull()); + if (sharedWatermarkStore) { + for (RuntimeContext rc : List.of(alice, bob)) { + var watermarkNamespace = + new ArrayList(List.of("memory", "consolidation", "scopes")); + watermarkNamespace.addAll(namespaces.getNamespace(rc)); + assertTrue(store.get(watermarkNamespace, "watermark") != null); + } + } + } + } + + @ParameterizedTest + @EnumSource(IsolationScope.class) + void watermarkUsesConfiguredScopeAndPreservesFileFallback( + IsolationScope scope, @TempDir Path tmp) { + InMemoryStore store = new InMemoryStore(); + NamespaceFactory namespaces = scope.toNamespaceFactory(); + RemoteFilesystem fs = + new RemoteFilesystem( + store, + rc -> { + var ns = new ArrayList(List.of("workspace")); + ns.addAll(namespaces.getNamespace(rc)); + return ns; + }); + RuntimeContext alice = RuntimeContext.builder().userId("alice").sessionId("a").build(); + RuntimeContext bob = RuntimeContext.builder().userId("bob").sessionId("b").build(); + Instant legacy = Instant.parse("2026-10-10T12:00:00Z"); + Instant scopedFile = legacy.minusSeconds(3600); + store.put( + List.of("memory", "consolidation"), + "watermark", + Map.of("ts", legacy.toEpochMilli())); + try (WorkspaceManager wsm = new WorkspaceManager(tmp, fs, null, namespaces)) { + MemoryConsolidator consolidator = new MemoryConsolidator(wsm, null, null, 4000, store); + boolean isolated = scope == IsolationScope.USER || scope == IsolationScope.SESSION; + assertEquals(isolated ? Instant.EPOCH : legacy, consolidator.readWatermark(alice)); + wsm.writeUtf8WorkspaceRelative( + alice, MemoryConsolidator.STATE_REL_PATH, scopedFile.toString()); + assertEquals(isolated ? scopedFile : legacy, consolidator.readWatermark(alice)); + assertEquals(isolated ? Instant.EPOCH : legacy, consolidator.readWatermark(bob)); + } + } + private static void seedStoreFile( InMemoryStore store, List ns, String path, String content, Instant modifiedAt) { Map value = diff --git a/docs/v2/en/docs/harness/memory.md b/docs/v2/en/docs/harness/memory.md index eadc165595..e108bbfa41 100644 --- a/docs/v2/en/docs/harness/memory.md +++ b/docs/v2/en/docs/harness/memory.md @@ -266,6 +266,8 @@ When memory is enabled, a throttled background job also runs. The first eligible Entering maintenance does not necessarily call the model: consolidation skips the LLM request when there are no new daily ledger entries since the last successful consolidation. `FlushTrigger.never()` does not disable this maintenance path. +When using a distributed store, consolidation watermarks follow the workspace's configured namespace mapping. Isolated users or sessions do not advance each other's cursors; intentionally unscoped workspaces retain a shared cursor. On upgrade, an isolated scope ignores the legacy global cursor and falls back to its own `memory/.consolidation_state` file (or processes all retained ledgers if no file exists). Custom `WorkspaceManager` instances should receive the same `NamespaceFactory` as their filesystem. + All thresholds are tunable via `.memory(MemoryConfig.builder()...)`, though most projects don't need to touch them. ## Turn it off entirely diff --git a/docs/v2/zh/docs/harness/memory.md b/docs/v2/zh/docs/harness/memory.md index 23a94852ce..66712f1aea 100644 --- a/docs/v2/zh/docs/harness/memory.md +++ b/docs/v2/zh/docs/harness/memory.md @@ -265,6 +265,8 @@ Memory 的记录边界为一行，Session 的记录边界为一条 entry，不 进入维护流程不一定会调用模型：如果自上次成功合并以来没有新增日流水账内容，consolidation 会跳过 LLM 请求。`FlushTrigger.never()` 不会关闭这条维护路径。 +使用分布式存储时，记忆合并水位遵循工作区配置的 namespace 映射。隔离的用户或会话不会推进彼此的水位；未分作用域的工作区仍共用水位。升级后，隔离作用域会忽略历史全局水位，回退到自己的 `memory/.consolidation_state` 文件；文件不存在时，会处理全部保留的日流水账。自行构造 `WorkspaceManager` 时，应传入与文件系统相同的 `NamespaceFactory`。 + 所有阈值都可以通过 `.memory(MemoryConfig.builder()...)` 调，绝大多数项目不需要碰。 ## 完全关掉 From fcc9f0bf10c3d22b393cb14a03f577f12a10ff93 Mon Sep 17 00:00:00 2001 From: saddddboi <99187163+saddddboi@users.noreply.github.com> Date: Mon, 12 Oct 2026 09:47:12 +0800 Subject: [PATCH 2/2] fix(harness): cover watermark fallback and CAS retries --- .../agent/memory/MemoryConsolidator.java | 2 +- .../MemoryConsolidatorFilesystemTest.java | 109 ++++++++++++++++++ 2 files changed, 110 insertions(+), 1 deletion(-) diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java index ee57b0bb87..0c7d374c3a 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java @@ -362,7 +362,7 @@ private List watermarkNamespace(RuntimeContext rc) { factory == null ? List.of() : factory.getNamespace(rc != null ? rc : RuntimeContext.empty()); - if (scope == null || scope.isEmpty()) { + if (scope.isEmpty()) { return WATERMARK_NAMESPACE; } // Keep intentionally unscoped cursors compatible, but never import the old global diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java index 0b671bb545..e85ef0550f 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java @@ -17,6 +17,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.isNull; @@ -151,6 +153,113 @@ void watermarkUsesConfiguredScopeAndPreservesFileFallback( } } + @Test + void watermarkWithoutNamespaceFactoryReadsLegacyStoreSlot(@TempDir Path tmp) { + InMemoryStore store = new InMemoryStore(); + Instant legacy = Instant.parse("2026-10-10T12:00:00Z"); + store.put( + List.of("memory", "consolidation"), + "watermark", + Map.of("ts", legacy.toEpochMilli())); + RemoteFilesystem fs = new RemoteFilesystem(store, List.of("workspace")); + try (WorkspaceManager wsm = new WorkspaceManager(tmp, fs, null, null)) { + MemoryConsolidator consolidator = new MemoryConsolidator(wsm, null, null, 4000, store); + assertEquals(legacy, consolidator.readWatermark(RuntimeContext.empty())); + } + } + + @Test + void watermarkNormalizesNullRuntimeContextBeforeResolvingNamespace(@TempDir Path tmp) { + InMemoryStore store = new InMemoryStore(); + Instant scoped = Instant.parse("2026-10-10T12:00:00Z"); + NamespaceFactory namespaces = + rc -> { + assertNotNull(rc); + assertNull(rc.getUserId()); + assertNull(rc.getSessionId()); + return List.of("anonymous"); + }; + store.put( + List.of("memory", "consolidation", "scopes", "anonymous"), + "watermark", + Map.of("ts", scoped.toEpochMilli())); + RemoteFilesystem fs = new RemoteFilesystem(store, namespaces); + try (WorkspaceManager wsm = new WorkspaceManager(tmp, fs, null, namespaces)) { + MemoryConsolidator consolidator = new MemoryConsolidator(wsm, null, null, 4000, store); + assertEquals(scoped, consolidator.readWatermark(null)); + } + } + + @Test + void watermarkRetriesCasConflictWithinTheSameScope(@TempDir Path tmp) { + List watermarkNamespace = List.of("memory", "consolidation", "scopes", "alice"); + Instant old = Instant.now().minusSeconds(7200); + List expectedVersions = new ArrayList<>(); + InMemoryStore store = + new InMemoryStore() { + @Override + public boolean putIfVersion( + List namespace, + String key, + Map value, + long expectedVersion) { + assertEquals(watermarkNamespace, namespace); + assertEquals("watermark", key); + expectedVersions.add(expectedVersion); + if (expectedVersions.size() == 1) { + // Another replica advances the version between the read and CAS. + super.put( + namespace, + key, + Map.of("ts", old.plusSeconds(1).toEpochMilli())); + } + return super.putIfVersion(namespace, key, value, expectedVersion); + } + }; + store.put(watermarkNamespace, "watermark", Map.of("ts", old.toEpochMilli())); + NamespaceFactory namespaces = IsolationScope.USER.toNamespaceFactory(); + RuntimeContext alice = RuntimeContext.builder().userId("alice").build(); + RemoteFilesystem fs = new RemoteFilesystem(store, namespaces); + seedStoreFile( + store, + namespaces.getNamespace(alice), + "memory/2026-10-11.md", + "Alice fact", + old.plusSeconds(3600)); + Model model = mock(Model.class); + when(model.stream(anyList(), isNull(), isNull())) + .thenReturn( + Flux.just( + ChatResponse.builder() + .content( + List.of( + TextBlock.builder() + .text("curated") + .build())) + .build())); + try (WorkspaceManager wsm = new WorkspaceManager(tmp, fs, null, namespaces)) { + MemoryConsolidator consolidator = new MemoryConsolidator(wsm, model, null, 4000, store); + Instant before = Instant.now(); + consolidator.consolidate(alice).block(); + + assertEquals("curated", wsm.readMemoryMd(alice)); + assertEquals(List.of(1L, 2L), expectedVersions); + assertEquals(3L, store.get(watermarkNamespace, "watermark").version()); + Instant fileWatermark = + Instant.parse( + wsm.readManagedWorkspaceFileUtf8( + alice, MemoryConsolidator.STATE_REL_PATH)); + assertFalse(fileWatermark.isBefore(before)); + assertEquals( + fileWatermark.toEpochMilli(), consolidator.readWatermark(alice).toEpochMilli()); + assertNull(store.get(List.of("memory", "consolidation"), "watermark")); + assertEquals( + Instant.EPOCH, + consolidator.readWatermark(RuntimeContext.builder().userId("bob").build())); + verify(model, times(1)).stream(anyList(), isNull(), isNull()); + } + } + private static void seedStoreFile( InMemoryStore store, List ns, String path, String content, Instant modifiedAt) { Map value =