refactor(session): extract _safe_json_loads to DRY corruption recovery

- Extract duplicated raw_decode fallback into _safe_json_loads() helper
- Add total-corruption fallback: returns empty dict instead of raising
- Add 3 test cases for completely garbled files
- Addresses gemini-code-assist review feedback on PR agentscope-ai#3278

Author: zealonexp

src/copaw/app/runner/session.py

@@ -21,6 +21,45 @@
 logger = logging.getLogger(__name__)
 
 
+def _safe_json_loads(content: str, filepath: str = "") -> dict:
+    """Parse JSON with corruption recovery.
+
+    Attempts standard ``json.loads`` first.  If that fails due to
+    trailing garbage (a common symptom of concurrent-write race
+    conditions), falls back to ``raw_decode`` to extract the first
+    valid JSON object.  If the file is completely unparseable, returns
+    an empty dict and logs a warning so callers never crash.
+
+    Args:
+        content: Raw file content.
+        filepath: Used only for log messages.
+
+    Returns:
+        Parsed dict, or ``{}`` when the content is beyond recovery.
+    """
+    try:
+        return json.loads(content)
+    except json.JSONDecodeError:
+        pass
+
+    # Try to extract the first valid JSON object.
+    try:
+        result, _ = json.JSONDecoder().raw_decode(content)
+        logger.warning(
+            "Session file %s had corrupted JSON. "
+            "Recovered first valid object via raw_decode.",
+            filepath,
+        )
+        return result
+    except json.JSONDecodeError:
+        logger.warning(
+            "Session file %s is completely corrupted and could not "
+            "be recovered. Returning empty dict.",
+            filepath,
+        )
+        return {}
+
+
 # Characters forbidden in Windows filenames
 _UNSAFE_FILENAME_RE = re.compile(r'[\\/:*?"<>|]')
 
@@ -111,18 +150,7 @@ async def load_session_state(
                 errors="surrogatepass",
             ) as f:
                 content = await f.read()
-                try:
-                    states = json.loads(content)
-                except json.JSONDecodeError:
-                    # Fallback: extract first valid JSON object (handles
-                    # race-condition corruption where two writes overlap)
-                    decoder = json.JSONDecoder()
-                    states, _ = decoder.raw_decode(content)
-                    logger.warning(
-                        "Session file %s had corrupted JSON (Extra data). "
-                        "Recovered first valid object.",
-                        session_save_path,
-                    )
+                states = _safe_json_loads(content, session_save_path)
 
             for name, state_module in state_modules_mapping.items():
                 if name in states:
@@ -165,16 +193,7 @@ async def update_session_state(
                 errors="surrogatepass",
             ) as f:
                 content = await f.read()
-                try:
-                    states = json.loads(content)
-                except json.JSONDecodeError:
-                    decoder = json.JSONDecoder()
-                    states, _ = decoder.raw_decode(content)
-                    logger.warning(
-                        "Session file %s had corrupted JSON during update. "
-                        "Recovered first valid object.",
-                        session_save_path,
-                    )
+                states = _safe_json_loads(content, session_save_path)
 
         else:
             if not create_if_not_exist:
@@ -243,16 +262,7 @@ async def get_session_state_dict(
                 errors="surrogatepass",
             ) as file:
                 content = await file.read()
-                try:
-                    states = json.loads(content)
-                except json.JSONDecodeError:
-                    decoder = json.JSONDecoder()
-                    states, _ = decoder.raw_decode(content)
-                    logger.warning(
-                        "Session file %s had corrupted JSON during get. "
-                        "Recovered first valid object.",
-                        session_save_path,
-                    )
+                states = _safe_json_loads(content, session_save_path)
 
             logger.info(
                 "Get session state dict from %s successfully.",

tests/unit/agents/test_session.py

@@ -180,3 +180,57 @@ async def test_get_nonexistent(sess):
         user_id="",
     )
     assert result == {}
+
+
+# ── completely corrupted file ──────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_load_completely_corrupted(sess, tmp_session_dir):
+    """File with no valid JSON at all should not crash (returns empty)."""
+    path = os.path.join(tmp_session_dir, "test--session.json")
+    with open(path, "w", encoding="utf-8") as f:
+        f.write("{{{THIS IS NOT JSON AT ALL!!!")
+
+    mod = FakeModule()
+    await sess.load_session_state(
+        "test:session",
+        user_id="",
+        memory=mod,
+    )
+    # memory key not in recovered (empty) dict → data stays None
+    assert mod.data is None
+
+
+@pytest.mark.asyncio
+async def test_get_completely_corrupted(sess, tmp_session_dir):
+    """get_session_state_dict returns empty dict for totally garbled file."""
+    path = os.path.join(tmp_session_dir, "test--session.json")
+    with open(path, "w", encoding="utf-8") as f:
+        f.write("NOT JSON {{{{")
+
+    result = await sess.get_session_state_dict(
+        "test:session",
+        user_id="",
+    )
+    assert result == {}
+
+
+@pytest.mark.asyncio
+async def test_update_completely_corrupted(sess, tmp_session_dir):
+    """update_session_state recovers from total corruption
+    by starting fresh."""
+    path = os.path.join(tmp_session_dir, "test--session.json")
+    with open(path, "w", encoding="utf-8") as f:
+        f.write("GARBAGE DATA !!!")
+
+    await sess.update_session_state(
+        "test:session",
+        key="memory.content",
+        value=["recovered"],
+        user_id="",
+    )
+
+    with open(path, encoding="utf-8") as f:
+        result = json.load(f)
+    assert result["memory"]["content"] == ["recovered"]