Skip to content

fix(harness): isolate memory consolidation watermarks #3513

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
saddddboi wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(harness): isolate memory consolidation watermarks #3513

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
97ccb96
fix(harness): isolate memory consolidation watermarks
saddddboi Oct 12, 2026
fcc9f0b
fix(harness): cover watermark fallback and CAS retries
saddddboi Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 23 additions & 6 deletions agentscope-harness/src/main/java/io/agentscope/harness/agent/memory/MemoryConsolidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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());
Expand All @@ -356,21 +356,38 @@ private void writeWatermark(RuntimeContext rc, Instant ts) {
}
}

private void writeWatermarkToStore(Instant ts) {
private List<String> watermarkNamespace(RuntimeContext rc) {
var factory = workspaceManager.getNamespaceFactory();
List<String> scope =
factory == null
? List.of()
: factory.getNamespace(rc != null ? rc : RuntimeContext.empty());
if (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<String> namespace = new ArrayList<>(WATERMARK_NAMESPACE);
namespace.add("scopes");
namespace.addAll(scope);
return List.copyOf(namespace);
}

private void writeWatermarkToStore(RuntimeContext rc, Instant ts) {
List<String> namespace = watermarkNamespace(rc);
Map<String, Object> 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) {
Expand Down
222 changes: 222 additions & 0 deletions ...ss/src/test/java/io/agentscope/harness/agent/memory/MemoryConsolidatorFilesystemTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,19 +17,37 @@

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;
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
Expand All @@ -38,6 +56,210 @@
*/
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<String>(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<String>(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));
}
}

@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<String> watermarkNamespace = List.of("memory", "consolidation", "scopes", "alice");
Instant old = Instant.now().minusSeconds(7200);
List<Long> expectedVersions = new ArrayList<>();
InMemoryStore store =
new InMemoryStore() {
@Override
public boolean putIfVersion(
List<String> namespace,
String key,
Map<String, Object> 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<String> ns, String path, String content, Instant modifiedAt) {
Map<String, Object> value =
Expand Down
2 changes: 2 additions & 0 deletions docs/v2/en/docs/harness/memory.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions docs/v2/zh/docs/harness/memory.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,6 +265,8 @@ Memory 的记录边界为一行，Session 的记录边界为一条 entry，不

进入维护流程不一定会调用模型：如果自上次成功合并以来没有新增日流水账内容，consolidation 会跳过 LLM 请求。`FlushTrigger.never()` 不会关闭这条维护路径。

使用分布式存储时，记忆合并水位遵循工作区配置的 namespace 映射。隔离的用户或会话不会推进彼此的水位；未分作用域的工作区仍共用水位。升级后，隔离作用域会忽略历史全局水位，回退到自己的 `memory/.consolidation_state` 文件；文件不存在时，会处理全部保留的日流水账。自行构造 `WorkspaceManager` 时，应传入与文件系统相同的 `NamespaceFactory`。

所有阈值都可以通过 `.memory(MemoryConfig.builder()...)` 调，绝大多数项目不需要碰。

## 完全关掉
Expand Down
Loading