fix(session): add _decode_storage_key for corrupt-file repair in list_sessions

This commit is contained in:
axelray-dev
2026-06-27 16:52:54 +08:00
committed by Xubin Ren
parent 00a907c493
commit 3ce77633c0
2 changed files with 16 additions and 16 deletions
+15 -2
View File
@@ -409,6 +409,18 @@ class SessionManager:
"""Collision-resistant encoding for internal session storage filenames."""
return base64.urlsafe_b64encode(key.encode()).decode().rstrip("=")
@staticmethod
def _decode_storage_key(stem: str) -> str | None:
"""Reverse _storage_key(): decode a base64url (no-padding) stem back to the original key."""
try:
# Restore padding stripped by rstrip("=")
padding = 4 - len(stem) % 4
if padding != 4:
stem += "=" * padding
return base64.urlsafe_b64decode(stem).decode("utf-8")
except Exception:
return None
def _get_session_path(self, key: str) -> Path:
"""Get the collision-resistant workspace path for a session."""
return self.sessions_dir / f"{self._storage_key(key)}.jsonl"
@@ -804,7 +816,8 @@ class SessionManager:
sessions = []
for path in self.sessions_dir.glob("*.jsonl"):
fallback_key = path.stem.replace("_", ":", 1)
decoded = self._decode_storage_key(path.stem)
fallback_key = decoded or path.stem.replace("_", ":", 1)
try:
# Read the metadata line and a small preview for session lists.
with open(path, encoding="utf-8") as f:
@@ -812,7 +825,7 @@ class SessionManager:
if first_line:
data = json.loads(first_line)
if data.get("_type") == "metadata":
key = data.get("key") or path.stem.replace("_", ":", 1)
key = data.get("key") or fallback_key
metadata = data.get("metadata", {})
title = _metadata_title(metadata)
preview = ""
+1 -14
View File
@@ -231,21 +231,8 @@ def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]:
}
def _try_decode_storage_stem(stem: str) -> str | None:
"""Try to decode a base64url (no-padding) session storage stem back to the original key."""
import base64
try:
padding = 4 - len(stem) % 4
if padding != 4:
stem += "=" * padding
return base64.urlsafe_b64decode(stem).decode("utf-8")
except Exception:
return None
def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, Any] | None:
storage_key = _try_decode_storage_stem(path.stem)
storage_key = SessionManager._decode_storage_key(path.stem)
fallback_key = storage_key or path.stem.replace("_", ":", 1)
try:
with open(path, encoding="utf-8") as f: