From 51791be0cf9f9ee3a3e3f3ebf57973aee32f66a2 Mon Sep 17 00:00:00 2001 From: guslegend0510 Date: Mon, 13 Jul 2026 20:24:07 +0800 Subject: [PATCH 1/2] fix(state): bound persisted session context --- .../java/io/agentscope/core/ReActAgent.java | 107 ++++++++++++++++-- .../agent/ReActAgentPerSessionStateTest.java | 76 +++++++++++++ .../harness/agent/HarnessAgent.java | 26 ++++- .../harness/agent/HarnessAgentTest.java | 3 + docs/v2/en/integration/distributed/mysql.md | 5 + docs/v2/en/integration/session/overview.md | 30 +++++ docs/v2/zh/integration/distributed/mysql.md | 3 + docs/v2/zh/integration/session/overview.md | 24 ++++ 8 files changed, 265 insertions(+), 9 deletions(-) diff --git a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java index 0bb6b49af7..cd74299b91 100644 --- a/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java +++ b/agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java @@ -245,6 +245,9 @@ public class ReActAgent extends AgentBase implements AutoCloseable { private final AgentStateStore stateStore; + /** Maximum number of conversation messages included in a persisted AgentState snapshot. */ + private final int maxPersistedContextMessages; + /** * Builder-time fallback {@code sessionId}, used only when a call does not supply a * {@code sessionId} via its {@link RuntimeContext}. Each call still picks its own active slot @@ -308,6 +311,7 @@ private ReActAgent(Builder builder, Toolkit agentToolkit) { this.middlewares = List.copyOf(mws); this.stateStore = builder.stateStore; + this.maxPersistedContextMessages = builder.maxPersistedContextMessages; this.defaultSessionId = builder.defaultSessionId != null && !builder.defaultSessionId.isBlank() ? builder.defaultSessionId @@ -326,11 +330,8 @@ private ReActAgent(Builder builder, Toolkit agentToolkit) { // interrupted request, so persist that session directly rather than the // instance "last-active" CallExecution (which is wrong under concurrency). agentState -> - stateStore.save( - agentState.getUserId(), - agentState.getSessionId(), - "agent_state", - agentState)); + persistAgentState( + agentState.getUserId(), agentState.getSessionId(), agentState)); } } @@ -421,11 +422,43 @@ private Mono saveStateToSession(CallExecution scope) { syncToolkitToState(scope.state); SlotRef ref = SlotRef.parse(scope.slotKey); AgentState toSave = scope.state; - return Mono.fromRunnable( - () -> stateStore.save(ref.userId, ref.sessionId, "agent_state", toSave)) + return Mono.fromRunnable(() -> persistAgentState(ref.userId, ref.sessionId, toSave)) .subscribeOn(Schedulers.boundedElastic()); } + private void persistAgentState(String userId, String sessionId, AgentState state) { + stateStore.save(userId, sessionId, "agent_state", createPersistenceSnapshot(state)); + } + + /** + * Creates a state snapshot for persistence without mutating the live per-call state. + * + *

The full state is returned when no trimming is necessary. When the configured limit is + * exceeded, all non-conversation state is preserved while only the most recent context + * messages are copied into the persisted snapshot. + */ + private AgentState createPersistenceSnapshot(AgentState state) { + List context = state.getContext(); + if (context.size() <= maxPersistedContextMessages) { + return state; + } + + int fromIndex = context.size() - maxPersistedContextMessages; + return AgentState.builder() + .sessionId(state.getSessionId()) + .userId(state.getUserId()) + .summary(state.getSummary()) + .context(context.subList(fromIndex, context.size())) + .replyId(state.getReplyId()) + .curIter(state.getCurIter()) + .shutdownInterrupted(state.isShutdownInterrupted()) + .permissionContext(state.getPermissionContext()) + .toolContext(state.getToolContext()) + .tasksContext(state.getTasksContext()) + .planModeContext(state.getPlanModeContext()) + .build(); + } + /** * Per-call slot activation. Reads {@code (userId, sessionId)} from the given RuntimeContext * (falling back to {@link #defaultSessionId} when absent), and atomically swaps the active @@ -3733,8 +3766,40 @@ public void saveAgentState(String userId, String sessionId) { String slot = slotKey(userId, sessionId); AgentState s = stateCache.get(slot); if (s != null) { - stateStore.save(userId, sessionId, "agent_state", s); + persistAgentState(userId, sessionId, s); + } + } + + /** + * Deletes all persisted state for the session identified by the given runtime context and + * evicts the corresponding local caches. + * + *

If a call for the same session is still in flight, that call may persist state again when + * it completes. Callers should invoke this method after active work for the session has stopped. + * + * @param ctx runtime context identifying the session + */ + public void deleteSessionState(RuntimeContext ctx) { + String uid = ctx != null ? ctx.getUserId() : null; + String sid = ctx != null ? ctx.getSessionId() : null; + deleteSessionState(uid, sid); + } + + /** + * Deletes all persisted state for a {@code (userId, sessionId)} session and evicts its local + * state and permission caches. + * + * @param userId nullable user identifier + * @param sessionId session identifier; uses the default session id when null or blank + */ + public void deleteSessionState(String userId, String sessionId) { + String sid = (sessionId == null || sessionId.isBlank()) ? defaultSessionId : sessionId; + if (stateStore != null) { + stateStore.delete(userId, sid); } + String slot = slotKey(userId, sid); + stateCache.remove(slot); + permissionEngineCache.remove(slot); } /** Returns the {@link AgentStateStore} configured for state persistence, or {@code null}. */ @@ -3742,6 +3807,11 @@ public AgentStateStore getStateStore() { return stateStore; } + /** Returns the maximum number of context messages included in persisted state snapshots. */ + public int getMaxPersistedContextMessages() { + return maxPersistedContextMessages; + } + /** * Returns the builder-time fallback {@code sessionId}, used when a call's * {@link RuntimeContext} carries no {@code sessionId}. @@ -3854,6 +3924,7 @@ public static class Builder { private Boolean flatStopOnReject; private AgentStateStore stateStore; private String defaultSessionId; + private int maxPersistedContextMessages = Integer.MAX_VALUE; // ==================== 1.x legacy compatibility fields ==================== // Below fields back the deprecated `longTermMemory(...)`, `knowledge(...)`, @@ -4194,6 +4265,26 @@ public Builder stateStore(AgentStateStore stateStore) { return this; } + /** + * Limits the number of recent conversation messages stored inside each persisted + * {@code agent_state} snapshot. + * + *

The default is unlimited for backward compatibility. A value of {@code 0} persists + * the rest of the agent state without conversation messages. The live state used by the + * current call is not modified when a snapshot is trimmed. + * + * @param maxMessages maximum number of recent context messages to persist + * @return this builder instance + */ + public Builder maxPersistedContextMessages(int maxMessages) { + if (maxMessages < 0) { + throw new IllegalArgumentException( + "maxPersistedContextMessages must be >= 0: " + maxMessages); + } + this.maxPersistedContextMessages = maxMessages; + return this; + } + /** * Sets the builder-time fallback {@code sessionId} used to persist {@code agent_state} * when a call does not supply a {@code sessionId} on its {@link RuntimeContext}. Defaults diff --git a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentPerSessionStateTest.java b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentPerSessionStateTest.java index 5d88b1febc..ff1784acee 100644 --- a/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentPerSessionStateTest.java +++ b/agentscope-core/src/test/java/io/agentscope/core/agent/ReActAgentPerSessionStateTest.java @@ -19,6 +19,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import io.agentscope.core.ReActAgent; @@ -119,6 +120,81 @@ void savePersistsPerSlot() { assertEquals("", other.getSummary()); } + @Test + @DisplayName("persisted context keeps only the configured number of recent messages") + void persistedContextIsBoundedWithoutMutatingLiveState() { + InMemoryAgentStateStore store = new InMemoryAgentStateStore(); + ReActAgent agent = + ReActAgent.builder() + .name("asst") + .sysPrompt("hi") + .model(new NoopModel()) + .stateStore(store) + .maxPersistedContextMessages(2) + .build(); + AgentState live = agent.getAgentState("u1", "sessA"); + live.contextMutable() + .addAll(List.of(userMsg("one"), userMsg("two"), userMsg("three"), userMsg("four"))); + + agent.saveAgentState("u1", "sessA"); + + assertEquals(4, live.getContext().size(), "saving must not trim the live state"); + AgentState persisted = + store.get("u1", "sessA", "agent_state", AgentState.class).orElseThrow(); + assertEquals(List.of("three", "four"), allText(persisted)); + } + + @Test + @DisplayName("automatic call persistence applies the configured context limit") + void automaticCallPersistenceIsBounded() { + InMemoryAgentStateStore store = new InMemoryAgentStateStore(); + ReActAgent agent = + ReActAgent.builder() + .name("asst") + .sysPrompt("hi") + .model(new NoopModel()) + .stateStore(store) + .maxPersistedContextMessages(3) + .build(); + RuntimeContext ctx = RuntimeContext.builder().userId("u1").sessionId("sessA").build(); + + agent.call(List.of(userMsg("first")), ctx).block(Duration.ofSeconds(5)); + agent.call(List.of(userMsg("second")), ctx).block(Duration.ofSeconds(5)); + + AgentState persisted = + store.get("u1", "sessA", "agent_state", AgentState.class).orElseThrow(); + assertEquals(3, persisted.getContext().size()); + assertEquals(List.of("ok", "second", "ok"), allText(persisted)); + } + + @Test + @DisplayName("deleteSessionState removes persisted state and evicts local caches") + void deleteSessionStateRemovesStoreAndCache() { + InMemoryAgentStateStore store = new InMemoryAgentStateStore(); + ReActAgent agent = agent(store); + AgentState original = agent.getAgentState("u1", "sessA"); + original.setSummary("stale"); + agent.saveAgentState("u1", "sessA"); + + agent.deleteSessionState("u1", "sessA"); + + assertFalse(store.exists("u1", "sessA")); + AgentState fresh = agent.getAgentState("u1", "sessA"); + assertNotSame(original, fresh); + assertEquals("", fresh.getSummary()); + } + + @Test + @DisplayName("negative persisted context limit is rejected") + void negativePersistedContextLimitIsRejected() { + IllegalArgumentException error = + assertThrows( + IllegalArgumentException.class, + () -> ReActAgent.builder().maxPersistedContextMessages(-1)); + + assertTrue(error.getMessage().contains("must be >= 0")); + } + @Test @DisplayName("user interrupt persists recovery state to the store") void userInterruptPersistsRecoveryState() throws Exception { diff --git a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java index d669184cb9..2ea119bea0 100644 --- a/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java +++ b/agentscope-harness/src/main/java/io/agentscope/harness/agent/HarnessAgent.java @@ -422,6 +422,16 @@ public AgentStateStore getStateStore() { return delegate.getStateStore(); } + /** Deletes persisted and locally cached state for the session identified by {@code ctx}. */ + public void deleteSessionState(RuntimeContext ctx) { + delegate.deleteSessionState(ctx); + } + + /** Deletes persisted and locally cached state for a {@code (userId, sessionId)} session. */ + public void deleteSessionState(String userId, String sessionId) { + delegate.deleteSessionState(userId, sessionId); + } + /** * The distributed store configured on this agent, or {@code null} for local * deployments. Exposed so {@link io.agentscope.harness.agent.gateway.GatewayBootstrap} can build @@ -1120,9 +1130,11 @@ private Builder() {} * {@code maxIters}{@code agent.getMaxIters()} * {@code generateOptions}{@code agent.getGenerateOptions()} * {@code toolkit}defensive copy via {@code agent.getToolkit().copy()} - * Persistence + * Persistence * {@code session}{@code agent.getStateStore()} if non-null * {@code defaultSessionId}{@code agent.getDefaultSessionId()} if non-null + * {@code maxPersistedContextMessages} + * {@code agent.getMaxPersistedContextMessages()} * Model resilience (from {@code agent.getModelConfig()}) * {@code maxRetries}{@link ModelConfig#maxRetries()} * {@code fallbackModel}{@link ModelConfig#fallbackModel()} if non-null @@ -1213,6 +1225,7 @@ public static Builder fromAgent(ReActAgent agent) { if (srcDefaultSessionId != null) { b.defaultSessionId(srcDefaultSessionId); } + b.maxPersistedContextMessages(agent.getMaxPersistedContextMessages()); // Model resilience. ModelConfig mc = agent.getModelConfig(); @@ -1390,6 +1403,17 @@ public Builder stateStore(AgentStateStore stateStore) { return this; } + /** + * Limits the number of recent conversation messages stored in persisted agent state. + * + * @param maxMessages maximum number of recent context messages to persist + * @return this builder + */ + public Builder maxPersistedContextMessages(int maxMessages) { + inner.maxPersistedContextMessages(maxMessages); + return this; + } + /** * Configures a distributed store that provides all storage components at once: * {@link AgentStateStore}, {@link io.agentscope.harness.agent.filesystem.remote.store.BaseStore}, diff --git a/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentTest.java b/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentTest.java index 8618448cec..dee5a4e352 100644 --- a/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentTest.java +++ b/agentscope-harness/src/test/java/io/agentscope/harness/agent/HarnessAgentTest.java @@ -621,6 +621,7 @@ public Flux onAgent( .name("source") .model(stubModel("done")) .toolkit(new Toolkit()) + .maxPersistedContextMessages(7) .middlewares(List.of(userMiddleware, new AgentTraceMiddleware())) .build(); @@ -632,6 +633,8 @@ public Flux onAgent( List entries = builder.buildSubagentEntries(workspace); HarnessAgent child = builder.build(); + assertEquals(7, child.getDelegate().getMaxPersistedContextMessages()); + long copiedUserMiddlewareCount = child.getDelegate().getMiddlewares().stream() .filter(m -> m == userMiddleware) diff --git a/docs/v2/en/integration/distributed/mysql.md b/docs/v2/en/integration/distributed/mysql.md index 3ff58d5927..913f24178d 100644 --- a/docs/v2/en/integration/distributed/mysql.md +++ b/docs/v2/en/integration/distributed/mysql.md @@ -24,11 +24,16 @@ DistributedStore store = MysqlDistributedStore.create(dataSource); HarnessAgent agent = HarnessAgent.builder() .distributedStore(store) + .maxPersistedContextMessages(200) // optional: bound session state growth .filesystem(new RemoteFilesystemSpec() .isolationScope(IsolationScope.USER)) .build(); ``` +`agent_state` contains the serialized conversation context, so long-running sessions grow unless a +retention limit is configured. See [Agent State Store](../session/overview.md) for persisted context +limits and application-controlled session deletion. + ## Components Provided ### 1. MysqlAgentStateStore diff --git a/docs/v2/en/integration/session/overview.md b/docs/v2/en/integration/session/overview.md index a6350f414e..017861a92e 100644 --- a/docs/v2/en/integration/session/overview.md +++ b/docs/v2/en/integration/session/overview.md @@ -31,6 +31,36 @@ ReActAgent agent = ReActAgent.builder() .build(); ``` +## Limit Persisted Conversation History + +By default, the complete conversation context is included in every persisted `agent_state` +snapshot. For long-running sessions, set a maximum number of recent messages to keep the stored +record bounded. This works with both a standalone `AgentStateStore` and a `DistributedStore`: + +```java +HarnessAgent agent = HarnessAgent.builder() + .name("assistant") + .model(model) + .distributedStore(distributedStore) + .maxPersistedContextMessages(200) + .build(); +``` + +The default is unlimited for backward compatibility. A value of `0` persists the rest of the agent +state without conversation messages. Trimming creates a persistence snapshot and does not mutate +the live state of the current call. + +Applications can also choose when to remove an entire session: + +```java +agent.deleteSessionState(userId, sessionId); +// or: agent.deleteSessionState(runtimeContext); +``` + +This removes the session from the configured store and evicts its local state caches. Call it only +after active work for that session has stopped; an in-flight call can persist the session again when +it completes. + For detailed usage and code examples, see each store's documentation: - [Redis](../distributed/redis.md#1-redisagentstatestore) diff --git a/docs/v2/zh/integration/distributed/mysql.md b/docs/v2/zh/integration/distributed/mysql.md index b62742a94f..b147f8fd02 100644 --- a/docs/v2/zh/integration/distributed/mysql.md +++ b/docs/v2/zh/integration/distributed/mysql.md @@ -24,11 +24,14 @@ DistributedStore store = MysqlDistributedStore.create(dataSource); HarnessAgent agent = HarnessAgent.builder() .distributedStore(store) + .maxPersistedContextMessages(200) // 可选：限制会话状态增长 .filesystem(new RemoteFilesystemSpec() .isolationScope(IsolationScope.USER)) .build(); ``` +`agent_state` 会序列化完整的会话上下文，因此长期运行的会话会持续增长。请参阅 [Agent 状态存储](../session/overview.md)，了解持久化消息上限和由接入系统主动删除会话的方式。 + ## 提供的组件 ### 1. MysqlAgentStateStore diff --git a/docs/v2/zh/integration/session/overview.md b/docs/v2/zh/integration/session/overview.md index b99ed0ba89..95c33565a9 100644 --- a/docs/v2/zh/integration/session/overview.md +++ b/docs/v2/zh/integration/session/overview.md @@ -31,6 +31,30 @@ ReActAgent agent = ReActAgent.builder() .build(); ``` +## 限制持久化的会话历史 + +默认情况下，每次保存 `agent_state` 都会包含完整的会话上下文。对于长期运行的会话，可以设置仅持久化最近多少条消息，避免存储记录无限增长。该配置同时适用于单独使用 `AgentStateStore` 和通过 `DistributedStore` 一键配置的场景： + +```java +HarnessAgent agent = HarnessAgent.builder() + .name("assistant") + .model(model) + .distributedStore(distributedStore) + .maxPersistedContextMessages(200) + .build(); +``` + +为保持向后兼容，默认不限制。设置为 `0` 时仍会保存其他 Agent 状态，但不会保存会话消息。裁剪只作用于持久化快照，不会修改当前调用正在使用的运行态。 + +接入系统也可以自行决定何时删除整个会话： + +```java +agent.deleteSessionState(userId, sessionId); +// 或：agent.deleteSessionState(runtimeContext); +``` + +该方法会删除存储中的会话并清理本地状态缓存。请在该会话没有正在执行的调用时使用；否则进行中的调用仍可能在结束时再次保存会话。 + 详细用法和代码示例请参阅各后端的文档： - [Redis](../distributed/redis.md#1-redisagentstatestore) From c633f850d4f0d0531343323b4b199e14c65c2414 Mon Sep 17 00:00:00 2001 From: guslegend0510 Date: Mon, 13 Jul 2026 21:45:09 +0800 Subject: [PATCH 2/2] ci: retry codecov upload