Merge pull request #6061 from MervinPraison/claude/issue-6024-20261026

fix: distinct session ids must never share a session file

src/praisonai-agents/praisonaiagents/session/hierarchy.py

@@ -306,6 +306,8 @@ class HierarchicalSessionStore(DefaultSessionStore):
                 raw = f.read()
             data = json.loads(raw.decode("utf-8"))
             session = ExtendedSessionData.from_dict(data)
+            if not self._owns_file(session_id, filepath, session):
+                return ExtendedSessionData(session_id=session_id)
             # Bind the fingerprint to this parsed object and these exact bytes.
             # A later writer must not certify an older payload via its stats.
             with self._lock:

src/praisonai-agents/praisonaiagents/session/sqlite_store.py

@@ -35,7 +35,13 @@ from typing import Any, Dict, List, Optional
 
 from praisonaiagents._logging import get_logger
 
-from .store import DefaultSessionStore, FileLock, SessionData
+from .store import (
+    DefaultSessionStore,
+    FileLock,
+    SessionData,
+    _legacy_session_filename,
+    _safe_session_filename,
+)
 
 logger = get_logger(__name__)
 
@@ -390,6 +396,22 @@ class SqliteSessionStore(DefaultSessionStore):
         except Exception:
             return set()
 
+    def _stored_session_id(self, filename: str) -> Optional[str]:
+        """Return the id stored in ``filename`` if it maps back to that file."""
+        import json
+
+        try:
+            with open(os.path.join(self.session_dir, filename), "r", encoding="utf-8") as f:
+                sid = json.load(f).get("session_id")
+        except (OSError, ValueError, AttributeError):
+            return None
+        if isinstance(sid, str) and filename in (
+            f"{_safe_session_filename(sid)}.json",
+            f"{_legacy_session_filename(sid)}.json",
+        ):
+            return sid
+        return None
+
     def _reindex_all(self) -> bool:
         """Backfill the index from existing JSON transcripts.
 
@@ -433,6 +455,10 @@ class SqliteSessionStore(DefaultSessionStore):
             if not filename.endswith(".json"):
                 continue
             sid = filename[:-5]
+            if "~" in sid or sid not in already:
+                # Hashed or pre-#6024 legacy name (Issue #6024): the real id
+                # lives in the file; loading it migrates a legacy file.
+                sid = self._stored_session_id(filename) or sid
             if sid in pending:
                 # Failed restore retries must not be bypassed by a synthetic
                 # empty transcript from the initial/upgrade backfill path.

src/praisonai-agents/praisonaiagents/session/store.py

@@ -6,6 +6,7 @@ Zero dependencies beyond stdlib.
 """
 
 import copy
+import hashlib
 import json
 from praisonaiagents._logging import get_logger
 import os
@@ -30,6 +31,27 @@ from ..paths import get_sessions_dir
 
 logger = get_logger(__name__)
 
+
+
+def _legacy_session_filename(session_id: str) -> str:
+    """Pre-#6024 lossy filename (many ids -> one name); kept only for migration."""
+    return "".join(c if c.isalnum() or c in "-_" else "_" for c in session_id)
+
+
+def _safe_session_filename(session_id: str) -> str:
+    """Injective, filesystem-safe filename stem for a session id (Issue #6024).
+
+    Ids already made of ``[A-Za-z0-9_-]`` keep their exact legacy name; any
+    other id gets a readable prefix plus a digest of the full id, so ids that
+    differ only in ``.``/``@``/``:``/… never share a file.
+    """
+    readable = _legacy_session_filename(session_id)
+    if readable == session_id:
+        return readable
+    digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest()[:16]
+    return f"{readable[:64]}~{digest}"
+
+
 # Module-level sentinel to track if we've warned about degraded locking
 _WARNED_NO_FCNTL = False
 
@@ -884,10 +906,56 @@ class DefaultSessionStore:
         )
 
     def _get_session_path(self, session_id: str) -> str:
-        """Get the file path for a session."""
-        # Sanitize session_id for filesystem
-        safe_id = "".join(c if c.isalnum() or c in "-_" else "_" for c in session_id)
-        return os.path.join(self.session_dir, f"{safe_id}.json")
+        """Get the file path for a session (injective mapping, Issue #6024)."""
+        path = self._session_file_path(session_id)
+        legacy = os.path.join(self.session_dir, f"{_legacy_session_filename(session_id)}.json")
+        # Migrate a pre-#6024 file only when it is provably this id's own.
+        # Recheck under the destination lock so a racing worker that already
+        # migrated and wrote here is never clobbered by a stale move.
+        if legacy != path and not os.path.exists(path) and os.path.exists(legacy):
+            try:
+                with FileLock(path, self.lock_timeout):
+                    if not os.path.exists(path):
+                        with open(legacy, "r", encoding="utf-8") as f:
+                            owner = json.load(f).get("session_id")
+                        if owner == session_id:
+                            os.replace(legacy, path)
+            except (OSError, ValueError, AttributeError):
+                pass
+        return path
+
+    def _session_file_path(self, session_id: str) -> str:
+        """Session file path without legacy migration (takes no lock)."""
+        return os.path.join(self.session_dir, f"{_safe_session_filename(session_id)}.json")
+
+    def _owns_file(self, session_id: str, filepath: str, session: SessionData) -> bool:
+        """Return False if the loaded file belongs to another session id.
+
+        A foreign file (e.g. a pre-#6024 legacy collision) is never served to
+        the caller; it is moved to its owner's own path (or set aside when that
+        is taken) so a later write here cannot destroy the owner's history.
+        Caller must hold the ``filepath`` lock. Raises ``OSError`` if the file
+        cannot be moved, so writers abort instead of overwriting it.
+        """
+        if not session.session_id:
+            session.session_id = session_id
+            return True
+        if session.session_id == session_id:
+            return True
+        logger.warning(
+            "Session file %s belongs to session %r, not %r; starting fresh",
+            filepath, session.session_id, session_id,
+        )
+        owner_path = self._session_file_path(session.session_id)
+        moved = False
+        if owner_path != filepath:
+            with FileLock(owner_path, self.lock_timeout):
+                if not os.path.exists(owner_path):
+                    os.replace(filepath, owner_path)
+                    moved = True
+        if not moved and self._quarantine_corrupt(filepath) is None:
+            raise OSError(f"cannot move foreign session file {filepath}")
+        return False
     
     def _load_session(self, session_id: str) -> SessionData:
         """Load session from disk with file locking."""
@@ -952,7 +1020,10 @@ class DefaultSessionStore:
         try:
             with open(filepath, "r", encoding="utf-8") as f:
                 data = json.load(f)
-            return SessionData.from_dict(data)
+            session = SessionData.from_dict(data)
+            if not self._owns_file(session_id, filepath, session):
+                return SessionData(session_id=session_id)
+            return session
         except (json.JSONDecodeError, UnicodeDecodeError) as e:
             # Malformed content — the parsed data is unusable, but the raw file
             # may still hold recoverable bytes. Quarantine it aside so the next
@@ -1122,9 +1193,7 @@ class DefaultSessionStore:
         try:
             spill_dir = self._spill_dir()
             os.makedirs(spill_dir, exist_ok=True)
-            safe_id = "".join(
-                c if c.isalnum() or c in "-_" else "_" for c in session_id
-            )
+            safe_id = _safe_session_filename(session_id)
             # A monotonic ms timestamp keeps files sortable/attributable, but a
             # short random token guarantees uniqueness so consecutive failures
             # within the same PID+millisecond never overwrite an earlier spill
@@ -1235,13 +1304,15 @@ class DefaultSessionStore:
             spill_dir = self._spill_dir()
             if not os.path.isdir(spill_dir):
                 return
-            safe_id = "".join(
-                c if c.isalnum() or c in "-_" else "_" for c in session_id
+            # Also match pre-#6024 spill names; the session_id check below
+            # rejects any spill that belongs to a colliding id.
+            prefixes = (
+                f"{_safe_session_filename(session_id)}.",
+                f"{_legacy_session_filename(session_id)}.",
             )
-            prefix = f"{safe_id}."
             candidates = sorted(
                 f for f in os.listdir(spill_dir)
-                if f.startswith(prefix) and f.endswith(".json")
+                if f.startswith(prefixes) and f.endswith(".json")
             )
         except (IOError, OSError):
             return
@@ -1331,7 +1402,9 @@ class DefaultSessionStore:
         new_messages = [m for _, msgs in recovered for m in msgs]
         if new_messages:
             session.messages.extend(new_messages)
-            filepath = self._get_session_path(session_id)
+            # Callers already hold this path's lock; the migrating lookup
+            # would take it again and stall for ``lock_timeout``.
+            filepath = self._session_file_path(session_id)
             session.updated_at = datetime.now(timezone.utc).isoformat()
             # Recovered turns extend the active window like any other append, so
             # they must go through the same retention policy (compact/truncate)

src/praisonai-agents/tests/unit/session/test_session_path_collisions.py

@@ -0,0 +1,119 @@
+"""Distinct session ids must never share a session file (Issue #6024)."""
+
+import json
+import os
+
+import pytest
+
+from praisonaiagents.session.store import DefaultSessionStore
+
+
+@pytest.fixture
+def store(tmp_path):
+    return DefaultSessionStore(session_dir=str(tmp_path))
+
+
+@pytest.mark.parametrize("a,b", [
+    ("alice.smith", "alice_smith"),
+    ("a:b", "a_b"),
+    ("[email protected]", "[email protected]"),
+    ("[email protected]", "u_tag_x_com"),
+    ("a/b", "a b"),
+    ("telegram:acct:default:chat:-1001:", "telegram_acct_default_chat_-1001_"),
+])
+def test_colliding_ids_get_distinct_paths(store, a, b):
+    assert store._get_session_path(a) != store._get_session_path(b)
+
+
+def test_safe_ids_keep_legacy_filename(store):
+    sid = "user_123-abc" + "x" * 100
+    assert os.path.basename(store._get_session_path(sid)) == f"{sid}.json"
+
+
+def test_colliding_users_have_separate_histories(store):
+    store.add_user_message("[email protected]", "My PIN is 4471.")
+    assert store.get_chat_history("[email protected]") == []
+    store.add_user_message("[email protected]", "hello")
+    assert [m["content"] for m in store.get_chat_history("[email protected]")] == ["My PIN is 4471."]
+    assert [m["content"] for m in store.get_chat_history("[email protected]")] == ["hello"]
+
+
+def _write_legacy(store, stem, owner, content):
+    path = os.path.join(store.session_dir, f"{stem}.json")
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump({"session_id": owner, "messages": [
+            {"role": "user", "content": content, "timestamp": 1.0}]}, f)
+    return path
+
+
+def test_legacy_file_migrated_only_for_owner(store):
+    legacy = _write_legacy(store, "alice_smith", "alice.smith", "secret")
+    # A colliding unsafe id never adopts it.
+    assert store.get_chat_history("alice:smith") == []
+    assert os.path.exists(legacy)
+    # The real owner migrates it on first read.
+    assert [m["content"] for m in store.get_chat_history("alice.smith")] == ["secret"]
+    assert not os.path.exists(legacy)
+
+
+def test_foreign_file_never_served_and_not_destroyed(store):
+    # Safe id whose filename holds another id's unmigrated legacy transcript.
+    _write_legacy(store, "alice_smith", "alice.smith", "secret")
+    assert store.get_chat_history("alice_smith") == []
+    store.add_user_message("alice_smith", "mine")
+    assert [m["content"] for m in store.get_chat_history("alice_smith")] == ["mine"]
+    # The owner's history was relocated rather than overwritten.
+    assert [m["content"] for m in store.get_chat_history("alice.smith")] == ["secret"]
+
+
+def test_failed_foreign_move_aborts_write(store, monkeypatch, tmp_path):
+    monkeypatch.setattr(store, "_spill_dir", lambda: str(tmp_path / "spill"))
+    store.add_user_message("alice.smith", "new")  # owner path taken
+    legacy = _write_legacy(store, "alice_smith", "alice.smith", "secret")
+    monkeypatch.setattr(store, "_quarantine_corrupt", lambda p: None)
+    assert store.add_user_message("alice_smith", "mine") is False
+    with open(legacy, encoding="utf-8") as f:
+        assert json.load(f)["session_id"] == "alice.smith"
+
+
+def test_foreign_file_set_aside_when_owner_path_taken(store):
+    store.add_user_message("alice.smith", "new")
+    legacy = _write_legacy(store, "alice_smith", "alice.smith", "secret")
+    store.add_user_message("alice_smith", "mine")
+    assert [m["content"] for m in store.get_chat_history("alice_smith")] == ["mine"]
+    assert [m["content"] for m in store.get_chat_history("alice.smith")] == ["new"]
+    aside = [p for p in os.listdir(store.session_dir) if p.startswith("alice_smith.json.corrupt-")]
+    assert len(aside) == 1 and os.path.exists(legacy)
+
+
+def test_sqlite_backfill_indexes_hashed_filenames(tmp_path):
+    from praisonaiagents.session.sqlite_store import SqliteSessionStore
+
+    DefaultSessionStore(session_dir=str(tmp_path)).add_user_message(
+        "[email protected]", "narwhal record")
+    store = SqliteSessionStore(session_dir=str(tmp_path), db_path=":memory:")
+    assert [h.session_id for h in store.search("narwhal")] == ["[email protected]"]
+
+
+def test_sqlite_backfill_indexes_legacy_collision_files(tmp_path):
+    from praisonaiagents.session.sqlite_store import SqliteSessionStore
+
+    legacy = DefaultSessionStore(session_dir=str(tmp_path))
+    _write_legacy(legacy, "alice_smith", "alice.smith", "walrus record")
+    store = SqliteSessionStore(session_dir=str(tmp_path), db_path=":memory:")
+    assert [h.session_id for h in store.search("walrus")] == ["alice.smith"]
+
+
+def test_spill_recovery_does_not_wait_on_own_lock(store, monkeypatch, tmp_path):
+    import time
+
+    spill = tmp_path / "spill"
+    spill.mkdir()
+    monkeypatch.setattr(store, "_spill_dir", lambda: str(spill))
+    # Legacy name of "alice.smith" belongs to another id, so it is never migrated.
+    _write_legacy(store, "alice_smith", "alice:smith", "theirs")
+    (spill / "alice_smith.1.json").write_text(json.dumps({"session_id": "alice.smith", "messages": [
+        {"role": "user", "content": "spilled", "timestamp": 2.0}]}), encoding="utf-8")
+    start = time.monotonic()
+    assert [m["content"] for m in store.get_chat_history("alice.smith")] == ["spilled"]
+    assert time.monotonic() - start < store.lock_timeout

src/praisonai-bot/tests/unit/test_bot_session_key_isolation.py

@@ -0,0 +1,32 @@
+"""Colliding bot user ids must get separate persisted histories (Issue #6024)."""
+
+import asyncio
+
+from praisonaiagents.session.store import DefaultSessionStore
+from praisonai_bot.bots._session import BotSessionManager
+
+
+class _EchoAgent:
+    def __init__(self):
+        self.name = "support"
+        self.chat_history = []
+        self.seen = {}
+
+    def chat(self, prompt):
+        self.seen[prompt] = [m["content"] for m in self.chat_history if m["role"] == "user"]
+        self.chat_history.append({"role": "user", "content": prompt})
+        self.chat_history.append({"role": "assistant", "content": "ok"})
+        return "ok"
+
+
+def test_colliding_email_senders_are_isolated(tmp_path):
+    mgr = BotSessionManager(store=DefaultSessionStore(session_dir=str(tmp_path)), platform="email")
+    agent = _EchoAgent()
+
+    async def main():
+        await mgr.chat(agent, "[email protected]", "My account PIN is 4471.")
+        await mgr.chat(agent, "[email protected]", "What do you know about me?")
+
+    asyncio.run(main())
+    assert "My account PIN is 4471." not in agent.seen["What do you know about me?"]
+    assert len([p for p in tmp_path.iterdir() if p.suffix == ".json"]) == 2

src/praisonai-code/praisonai_code/cli/commands/session.py

@@ -552,18 +552,15 @@ def session_fork(
     # searched (project first, then global) so the fork source and the
     # existence check stay consistent.
     #
-    # The store persists each session under a *sanitized* filename
-    # (``DefaultSessionStore._get_session_path`` replaces any char that is not
-    # alphanumeric/``-``/``_`` with ``_``). Sanitize identically here so a
-    # global-only id containing e.g. ``.`` or ``:`` still matches its real file
-    # instead of falling through to an empty project-scoped fork.
-    safe_id = "".join(c if c.isalnum() or c in "-_" else "_" for c in session_id)
+    # Ask each store directly: it resolves the injective filename and only
+    # migrates a pre-#6024 legacy file that really belongs to this id
+    # (Issue #6024), so a colliding id's file never selects the wrong store.
     store_dir = str(get_project_sessions_dir())
     for candidate in canonical_cli_stores():
         candidate_dir = getattr(candidate, "session_dir", None)
         if not candidate_dir:
             continue
-        if (Path(candidate_dir) / f"{safe_id}.json").exists():
+        if candidate.session_exists(session_id):
             store_dir = str(candidate_dir)
             break
 

src/praisonai-code/tests/unit/test_session_fork.py

@@ -163,3 +163,23 @@ def test_cli_session_fork_resolves_sanitized_global_id(project):
     # Real history copied, proving the fork resolved the global store (not an
     # empty project-scoped fallback) despite the special characters in the id.
     assert len(forked.messages) == 2
+
+
+def test_cli_session_fork_ignores_colliding_project_file(project):
+    # Issue #6024: a project session ``sess_global`` shares the pre-#6024 lossy
+    # filename with global ``sess.global``; fork must still pick the global one.
+    from praisonai_code.cli.state.project_sessions import canonical_cli_stores
+
+    _create_project_session("sess_global")
+    global_store = canonical_cli_stores()[-1]
+    global_store.add_message("sess.global", "user", "g-first")
+    global_store.add_message("sess.global", "assistant", "g-reply")
+
+    session_fork("sess.global", at_message=None, title=None)
+
+    from praisonaiagents.session.hierarchy import HierarchicalSessionStore
+
+    store = HierarchicalSessionStore(global_store.session_dir)
+    new_id = _forked_id(store, "sess.global")
+    assert new_id is not None
+    assert len(store._load_extended_session(new_id, force_reload=True).messages) == 2