From 2dac2da41834077cab7ed6776c4462f4da5a289b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E7=90=B3=28HuLin=29?= Date: Wed, 9 Sep 2026 15:10:08 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix(statestore):=20=E4=BF=AE=E5=A4=8D=20sav?= =?UTF-8?q?eIfVersion=20=E4=B8=AD=E7=9A=84=E5=8F=8D=E5=BA=8F=E5=88=97?= =?UTF-8?q?=E5=8C=96=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 当 expectedVersion 为 UNVERSIONED 时,所有 StateStore 实现都试图通过 getVersioned(..., State.class) 获取版本号,但 State 是接口, Jackson 无法直接反序列化接口类型,导致 InvalidDefinitionException。 修复方案: - 为每个 StateStore 实现添加私有的 readVersionOnly 方法 - 该方法直接查询版本号字段,不反序列化 state_data - 避免 Jackson 尝试将 JSON 反序列化为 State 接口 修改的实现: - JdbcAgentStateStore - MysqlAgentStateStore - PostgresAgentStateStore - RedisAgentStateStore - JedisAgentStateStore - RedissonAgentStateStore Fixes #3047 --- .../jdbc/state/JdbcAgentStateStore.java | 22 +++++++++++++++- .../mysql/state/MysqlAgentStateStore.java | 25 ++++++++++++++++++- .../state/PostgresAgentStateStore.java | 25 ++++++++++++++++++- .../redis/state/RedisAgentStateStore.java | 22 ++++++++++++++-- 4 files changed, 89 insertions(+), 5 deletions(-) diff --git a/agentscope-extensions/agentscope-extensions-jdbc/src/main/java/io/agentscope/extensions/jdbc/state/JdbcAgentStateStore.java b/agentscope-extensions/agentscope-extensions-jdbc/src/main/java/io/agentscope/extensions/jdbc/state/JdbcAgentStateStore.java index daa94a4e37..a5b0c808bb 100644 --- a/agentscope-extensions/agentscope-extensions-jdbc/src/main/java/io/agentscope/extensions/jdbc/state/JdbcAgentStateStore.java +++ b/agentscope-extensions/agentscope-extensions-jdbc/src/main/java/io/agentscope/extensions/jdbc/state/JdbcAgentStateStore.java @@ -233,7 +233,7 @@ public long saveIfVersion( String userId, String sessionId, String key, State value, long expectedVersion) { if (expectedVersion == UNVERSIONED) { save(userId, sessionId, key, value); - return getVersioned(userId, sessionId, key, State.class).version(); + return readVersionOnly(userId, sessionId, key); } String slotId = slotId(userId, sessionId); validateSlotId(slotId); @@ -276,6 +276,26 @@ public long saveIfVersion( } } + private long readVersionOnly(String userId, String sessionId, String key) { + String slotId = slotId(userId, sessionId); + validateSlotId(slotId); + validateStateKey(key); + + BoundSql boundSql = dialect.sessionStateSelectVersioned(slotId, key, SINGLE_STATE_INDEX); + try (Connection conn = dataSource.getConnection(); + PreparedStatement stmt = conn.prepareStatement(boundSql.sql())) { + bindParams(stmt, boundSql.params()); + try (ResultSet rs = stmt.executeQuery()) { + if (!rs.next()) { + return 0L; + } + return rs.getLong("version"); + } + } catch (Exception e) { + throw new RuntimeException("Failed to read version for state: " + key, e); + } + } + @Override public Optional get( String userId, String sessionId, String key, Class type) { diff --git a/agentscope-extensions/agentscope-extensions-mysql/src/main/java/io/agentscope/extensions/mysql/state/MysqlAgentStateStore.java b/agentscope-extensions/agentscope-extensions-mysql/src/main/java/io/agentscope/extensions/mysql/state/MysqlAgentStateStore.java index 6580e077fe..25a0a8f3c8 100644 --- a/agentscope-extensions/agentscope-extensions-mysql/src/main/java/io/agentscope/extensions/mysql/state/MysqlAgentStateStore.java +++ b/agentscope-extensions/agentscope-extensions-mysql/src/main/java/io/agentscope/extensions/mysql/state/MysqlAgentStateStore.java @@ -425,7 +425,7 @@ public long saveIfVersion( String userId, String sessionId, String key, State value, long expectedVersion) { if (expectedVersion == UNVERSIONED) { save(userId, sessionId, key, value); - return getVersioned(userId, sessionId, key, State.class).version(); + return readVersionOnly(userId, sessionId, key); } String slotId = slotId(userId, sessionId); @@ -449,6 +449,29 @@ public long saveIfVersion( } } + private long readVersionOnly(String userId, String sessionId, String key) { + String slotId = slotId(userId, sessionId); + validateSessionId(slotId); + validateStateKey(key); + + String sql = "SELECT version FROM " + getFullTableName() + + " WHERE session_id = ? AND state_key = ? AND item_index = 0"; + + try (Connection conn = dataSource.getConnection(); + PreparedStatement stmt = conn.prepareStatement(sql)) { + stmt.setString(1, slotId); + stmt.setString(2, key); + try (ResultSet rs = stmt.executeQuery()) { + if (!rs.next()) { + return 0L; + } + return rs.getLong("version"); + } + } catch (Exception e) { + throw new RuntimeException("Failed to read version for state: " + key, e); + } + } + private long insertIfAbsent(Connection conn, String slotId, String key, State value) throws Exception { String insertSql = diff --git a/agentscope-extensions/agentscope-extensions-postgresql/src/main/java/io/agentscope/extensions/postgresql/state/PostgresAgentStateStore.java b/agentscope-extensions/agentscope-extensions-postgresql/src/main/java/io/agentscope/extensions/postgresql/state/PostgresAgentStateStore.java index 24323f130b..cf32c9c397 100644 --- a/agentscope-extensions/agentscope-extensions-postgresql/src/main/java/io/agentscope/extensions/postgresql/state/PostgresAgentStateStore.java +++ b/agentscope-extensions/agentscope-extensions-postgresql/src/main/java/io/agentscope/extensions/postgresql/state/PostgresAgentStateStore.java @@ -330,7 +330,7 @@ public long saveIfVersion( String userId, String sessionId, String key, State value, long expectedVersion) { if (expectedVersion == UNVERSIONED) { save(userId, sessionId, key, value); - return getVersioned(userId, sessionId, key, State.class).version(); + return readVersionOnly(userId, sessionId, key); } String slotId = slotId(userId, sessionId); @@ -354,6 +354,29 @@ public long saveIfVersion( } } + private long readVersionOnly(String userId, String sessionId, String key) { + String slotId = slotId(userId, sessionId); + validateSessionId(slotId); + validateStateKey(key); + + String sql = "SELECT version FROM " + getFullTableName() + + " WHERE session_id = ? AND state_key = ? AND item_index = 0"; + + try (Connection conn = dataSource.getConnection(); + PreparedStatement stmt = conn.prepareStatement(sql)) { + stmt.setString(1, slotId); + stmt.setString(2, key); + try (ResultSet rs = stmt.executeQuery()) { + if (!rs.next()) { + return 0L; + } + return rs.getLong("version"); + } + } catch (Exception e) { + throw new RuntimeException("Failed to read version for state: " + key, e); + } + } + private long insertIfAbsent(Connection conn, String slotId, String key, State value) throws Exception { String insertSql = diff --git a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/RedisAgentStateStore.java b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/RedisAgentStateStore.java index 7679fd8b5e..a7c43303b8 100644 --- a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/RedisAgentStateStore.java +++ b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/RedisAgentStateStore.java @@ -270,8 +270,7 @@ public long saveIfVersion( String userId, String sessionId, String key, State value, long expectedVersion) { if (expectedVersion == UNVERSIONED) { save(userId, sessionId, key, value); - VersionedState after = getVersioned(userId, sessionId, key, State.class); - return after.version(); + return readVersionOnly(userId, sessionId, key); } String slotId = slotId(userId, sessionId); String redisKey = getStateKey(slotId, key); @@ -290,6 +289,25 @@ public long saveIfVersion( } } + /** + * Read only the version number without deserializing the payload. + * This avoids Jackson's inability to deserialize the State marker interface. + */ + private long readVersionOnly(String userId, String sessionId, String key) { + String slotId = slotId(userId, sessionId); + String redisKey = getStateKey(slotId, key); + String versionKey = RedisStateVersionSupport.versionKey(redisKey); + try { + String json = client.get(redisKey); + if (json == null) { + return 0L; + } + return RedisStateVersionSupport.parseVersion(json, client.get(versionKey)); + } catch (Exception e) { + throw new RuntimeException("Failed to read version for state: " + key, e); + } + } + @Override public void save(String userId, String sessionId, String key, List values) { String slotId = slotId(userId, sessionId); From 073a372db73bd29bee5085594dfe5957b0e15905 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=83=A1=E7=90=B3=28HuLin=29?= Date: Wed, 9 Sep 2026 15:11:21 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(statestore):=20=E4=BF=AE=E5=A4=8D=20Jed?= =?UTF-8?q?is/Redisson=20StateStore=20=E4=B8=AD=E7=9A=84=E5=8F=8D=E5=BA=8F?= =?UTF-8?q?=E5=88=97=E5=8C=96=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 补充修复 JedisAgentStateStore 和 RedissonAgentStateStore 中 saveIfVersion 方法在 expectedVersion 为 UNVERSIONED 时的反序列化问题。 Fixes #3047 --- .../state/jedis/JedisAgentStateStore.java | 17 ++++++++++++++++- .../redisson/RedissonAgentStateStore.java | 19 ++++++++++++++++++- 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/jedis/JedisAgentStateStore.java b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/jedis/JedisAgentStateStore.java index 0513656db5..458f326124 100644 --- a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/jedis/JedisAgentStateStore.java +++ b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/jedis/JedisAgentStateStore.java @@ -119,11 +119,26 @@ public long saveIfVersion( String userId, String sessionId, String key, State value, long expectedVersion) { if (expectedVersion == UNVERSIONED) { save(userId, sessionId, key, value); - return getVersioned(userId, sessionId, key, State.class).version(); + return readVersionOnly(userId, sessionId, key); } return evalSave(userId, sessionId, key, value, Long.toString(expectedVersion)); } + private long readVersionOnly(String userId, String sessionId, String key) { + String slotId = slotId(userId, sessionId); + String redisKey = getStateKey(slotId, key); + String versionKey = RedisStateVersionSupport.versionKey(redisKey); + try (Jedis jedis = jedisPool.getResource()) { + String json = jedis.get(redisKey); + if (json == null) { + return 0L; + } + return RedisStateVersionSupport.parseVersion(json, jedis.get(versionKey)); + } catch (Exception e) { + throw new RuntimeException("Failed to read version for state: " + key, e); + } + } + private long evalSave( String userId, String sessionId, String key, State value, String expectedVersionArg) { String slotId = slotId(userId, sessionId); diff --git a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/redisson/RedissonAgentStateStore.java b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/redisson/RedissonAgentStateStore.java index f497da3818..b123cc75b5 100644 --- a/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/redisson/RedissonAgentStateStore.java +++ b/agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/redisson/RedissonAgentStateStore.java @@ -125,11 +125,28 @@ public long saveIfVersion( String userId, String sessionId, String key, State value, long expectedVersion) { if (expectedVersion == UNVERSIONED) { save(userId, sessionId, key, value); - return getVersioned(userId, sessionId, key, State.class).version(); + return readVersionOnly(userId, sessionId, key); } return evalSave(userId, sessionId, key, value, Long.toString(expectedVersion)); } + private long readVersionOnly(String userId, String sessionId, String key) { + String slotId = slotId(userId, sessionId); + String redisKey = getStateKey(slotId, key); + String versionKey = RedisStateVersionSupport.versionKey(redisKey); + try { + RBucket bucket = redissonClient.getBucket(redisKey, StringCodec.INSTANCE); + String json = bucket.get(); + if (json == null) { + return 0L; + } + RBucket versionBucket = redissonClient.getBucket(versionKey, StringCodec.INSTANCE); + return RedisStateVersionSupport.parseVersion(json, versionBucket.get()); + } catch (Exception e) { + throw new RuntimeException("Failed to read version for state: " + key, e); + } + } + private long evalSave( String userId, String sessionId, String key, State value, String expectedVersionArg) { String slotId = slotId(userId, sessionId);