fix runtime error

Author: zhilingluo

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "agentscope-runtime"
-version = "1.0.3a3"
+version = "1.0.3a4"
 description = "A production-ready runtime framework for agent applications, providing secure sandboxed execution environments and scalable deployment solutions with multi-framework support."
 readme = "README.md"
 requires-python = ">=3.10"

src/agentscope_runtime/engine/services/agent_state/redis_state_service.py

@@ -159,4 +159,8 @@ async def export_state(
         if self._ttl_seconds is not None:
             await self._redis.expire(key, self._ttl_seconds)
 
-        return json.loads(state_json)
+        try:
+            return json.loads(state_json)
+        except (json.JSONDecodeError, ValueError):
+            # Return None for corrupted state data instead of raising exception
+            return None

src/agentscope_runtime/engine/services/memory/redis_memory_service.py

@@ -133,12 +133,14 @@ async def add_memory(
         if self._ttl_seconds is not None:
             await self._redis.expire(key, self._ttl_seconds)
 
-    async def search_memory(
+    async def search_memory(  # pylint: disable=too-many-branches
         self,
         user_id: str,
         messages: list,
         filters: Optional[Dict[str, Any]] = None,
     ) -> list:
+        if not self._redis:
+            raise RuntimeError("Redis connection is not available")
         key = self._user_key(user_id)
         if (
             not messages
@@ -172,7 +174,11 @@ async def search_memory(
             msgs_json = await self._redis.hget(key, session_id)
             if not msgs_json:
                 continue
-            msgs = self._deserialize(msgs_json)
+            try:
+                msgs = self._deserialize(msgs_json)
+            except Exception:
+                # Skip corrupted message data
+                continue
 
             # Match messages in this session
             for msg in msgs:
@@ -192,9 +198,8 @@ async def search_memory(
 
         # Refresh TTL on read to extend lifetime of actively used data,
         # if a TTL is configured and there is existing data for this key.
-        ttl_seconds = getattr(self, "_ttl", None)
-        if ttl_seconds and hash_keys:
-            await self._redis.expire(key, ttl_seconds)
+        if self._ttl_seconds is not None and hash_keys:
+            await self._redis.expire(key, self._ttl_seconds)
 
         return result
 
@@ -211,6 +216,8 @@ async def list_memory(
         user_id: str,
         filters: Optional[Dict[str, Any]] = None,
     ) -> list:
+        if not self._redis:
+            raise RuntimeError("Redis connection is not available")
         key = self._user_key(user_id)
         page_num = filters.get("page_num", 1) if filters else 1
         page_size = filters.get("page_size", 10) if filters else 10
@@ -226,8 +233,12 @@ async def list_memory(
         for session_id in sorted(hash_keys):
             msgs_json = await self._redis.hget(key, session_id)
             if msgs_json:
-                msgs = self._deserialize(msgs_json)
-                all_msgs.extend(msgs)
+                try:
+                    msgs = self._deserialize(msgs_json)
+                    all_msgs.extend(msgs)
+                except Exception:
+                    # Skip corrupted message data
+                    continue
 
                 # Early exit optimization: if we've loaded enough messages
                 # to cover the requested page, we can stop (but this assumes
@@ -236,7 +247,7 @@ async def list_memory(
 
         # Refresh TTL on active use to keep memory alive,
         # mirroring get_session behavior
-        if getattr(self, "_ttl_seconds", None):
+        if self._ttl_seconds is not None and hash_keys:
             await self._redis.expire(key, self._ttl_seconds)
         return all_msgs[start_index:end_index]
 
@@ -245,6 +256,8 @@ async def delete_memory(
         user_id: str,
         session_id: Optional[str] = None,
     ) -> None:
+        if not self._redis:
+            raise RuntimeError("Redis connection is not available")
         key = self._user_key(user_id)
         if session_id:
             await self._redis.hdel(key, session_id)

src/agentscope_runtime/engine/services/session_history/redis_session_history_service.py

@@ -106,6 +106,8 @@ async def create_session(
         user_id: str,
         session_id: Optional[str] = None,
     ) -> Session:
+        if not self._redis:
+            raise RuntimeError("Redis connection is not available")
         if session_id and session_id.strip():
             sid = session_id.strip()
         else:
@@ -127,12 +129,18 @@ async def get_session(
         user_id: str,
         session_id: str,
     ) -> Optional[Session]:
+        if not self._redis:
+            raise RuntimeError("Redis connection is not available")
         key = self._session_key(user_id, session_id)
         session_json = await self._redis.get(key)
         if session_json is None:
             return None
 
-        session = self._session_from_json(session_json)
+        try:
+            session = self._session_from_json(session_json)
+        except Exception:
+            # Return None for corrupted session data
+            return None
 
         # Refresh TTL when accessing the session
         if self._ttl_seconds is not None:
@@ -141,6 +149,8 @@ async def get_session(
         return session
 
     async def delete_session(self, user_id: str, session_id: str):
+        if not self._redis:
+            raise RuntimeError("Redis connection is not available")
         key = self._session_key(user_id, session_id)
         await self._redis.delete(key)
 
@@ -150,6 +160,8 @@ async def list_sessions(self, user_id: str) -> list[Session]:
         Uses SCAN to find all session:{user_id}:* keys. Expired sessions
         naturally disappear as their keys expire, avoiding stale entries.
         """
+        if not self._redis:
+            raise RuntimeError("Redis connection is not available")
         pattern = self._session_pattern(user_id)
         sessions = []
         cursor = 0
@@ -163,9 +175,13 @@ async def list_sessions(self, user_id: str) -> list[Session]:
             for key in keys:
                 session_json = await self._redis.get(key)
                 if session_json:
-                    session = self._session_from_json(session_json)
-                    session.messages = []
-                    sessions.append(session)
+                    try:
+                        session = self._session_from_json(session_json)
+                        session.messages = []
+                        sessions.append(session)
+                    except Exception:
+                        # Skip corrupted session data
+                        continue
 
             if cursor == 0:
                 break
@@ -182,6 +198,8 @@ async def append_message(
             List[Dict[str, Any]],
         ],
     ):
+        if not self._redis:
+            raise RuntimeError("Redis connection is not available")
         if not isinstance(message, list):
             message = [message]
         norm_message = []
@@ -199,14 +217,24 @@ async def append_message(
 
         session_json = await self._redis.get(key)
         if session_json is None:
-            raise RuntimeError(
-                f"Session {session_id} not found or has expired for user "
-                f"{user_id}. Previous memory/state has been lost. "
-                f"Please create a new session.",
+            # Session expired or not found, treat as a new session
+            # Create a new session with the current messages
+            stored_session = Session(
+                id=session_id,
+                user_id=user_id,
+                messages=norm_message.copy(),
             )
-
-        stored_session = self._session_from_json(session_json)
-        stored_session.messages.extend(norm_message)
+        else:
+            try:
+                stored_session = self._session_from_json(session_json)
+                stored_session.messages.extend(norm_message)
+            except Exception:
+                # Session data corrupted, treat as a new session
+                stored_session = Session(
+                    id=session_id,
+                    user_id=user_id,
+                    messages=norm_message.copy(),
+                )
 
         # Limit the number of messages per session to prevent memory issues
         if self._max_messages_per_session is not None:

src/agentscope_runtime/version.py

@@ -1,2 +1,2 @@
 # -*- coding: utf-8 -*-
-__version__ = "v1.0.3a3"
+__version__ = "v1.0.3a4"

tests/unit/test_redis_session_history_service.py

@@ -264,19 +264,25 @@ async def test_append_message(
     for i, msg in enumerate(stored_session.messages[2:]):
         assert msg.content == messages3[i].get("content")
 
-    # Test appending to a non-existent session (should raise RuntimeError)
+    # Test appending to a non-existent session (should create new session)
     non_existent_session = Session(
         id="non_existent",
         user_id=user_id,
         messages=[],
     )
-    # This should raise a RuntimeError indicating
-    # the session is missing/expired
-    with pytest.raises(RuntimeError, match="not found or has expired"):
-        await session_history_service.append_message(
-            non_existent_session,
-            message1,
-        )
+    # This should not raise an error, but create a new session
+    await session_history_service.append_message(
+        non_existent_session,
+        message1,
+    )
+    # Verify the session was created with the message
+    retrieved_session = await session_history_service.get_session(
+        user_id,
+        "non_existent",
+    )
+    assert retrieved_session is not None
+    assert len(retrieved_session.messages) == 1
+    assert retrieved_session.messages[0].content == message1.get("content")
 
 
 @pytest.mark.asyncio