Skip to content

fix: reload session file on read to prevent stale chat history #1764

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
cursor wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

fix: reload session file on read to prevent stale chat history #1764

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 11 additions & 21 deletions src/praisonai-agents/praisonaiagents/session/store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,32 +265,22 @@ def _get_session_path(self, session_id: str) -> str:
def _load_session(self, session_id: str) -> SessionData:
"""Load session from disk with file locking."""
filepath = self._get_session_path(session_id)

# Check cache first
with self._lock:
if session_id in self._cache:
return self._cache[session_id]

# Load from disk
if not os.path.exists(filepath):
session = SessionData(session_id=session_id)

# When a session file exists, always reload under FileLock so reads
# from another DefaultSessionStore instance (or process) are visible.
if os.path.exists(filepath):
with FileLock(filepath, self.lock_timeout):
session = self._load_session_from_disk(session_id, filepath)
with self._lock:
self._cache[session_id] = session
return session

with FileLock(filepath, self.lock_timeout):
try:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
session = SessionData.from_dict(data)
except (json.JSONDecodeError, IOError) as e:
logger.warning(f"Failed to load session {session_id}: {e}")
session = SessionData(session_id=session_id)


with self._lock:
if session_id in self._cache:
return self._cache[session_id]
session = SessionData(session_id=session_id)
self._cache[session_id] = session

return session
return session
Comment on lines +271 to +283
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot May 28, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Avoid TOCTOU race between existence check and cache return.

Line 271 checks file existence before locking, but Lines 279-280 can still return cached data if another instance creates/writes the file in between. That can reintroduce a stale-read edge case on this path.

Suggested fix 
     def _load_session(self, session_id: str) -> SessionData:
         """Load session from disk with file locking."""
         filepath = self._get_session_path(session_id)
-
-        # When a session file exists, always reload under FileLock so reads
-        # from another DefaultSessionStore instance (or process) are visible.
-        if os.path.exists(filepath):
-            with FileLock(filepath, self.lock_timeout):
-                session = self._load_session_from_disk(session_id, filepath)
-            with self._lock:
-                self._cache[session_id] = session
-            return session
-
-        with self._lock:
-            if session_id in self._cache:
-                return self._cache[session_id]
-            session = SessionData(session_id=session_id)
-            self._cache[session_id] = session
-            return session
+        # Serialize existence/read decision to avoid stale cache races.
+        with FileLock(filepath, self.lock_timeout):
+            if os.path.exists(filepath):
+                session = self._load_session_from_disk(session_id, filepath)
+            else:
+                with self._lock:
+                    session = self._cache.get(session_id) or SessionData(session_id=session_id)
+
+        with self._lock:
+            self._cache[session_id] = session
+        return session
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/praisonai-agents/praisonaiagents/session/store.py` around lines 271 -
283, Race between the os.path.exists check and returning cached data can cause a
stale-read; fix by checking the in-memory cache first, then if the file exists
acquire the FileLock and re-check the cache before loading from disk.
Concretely: in the method that uses filepath/session_id, do "with self._lock: if
session_id in self._cache: return self._cache[session_id]" then if
os.path.exists(filepath) acquire FileLock(filepath, self.lock_timeout) and,
while holding self._lock again, re-check self._cache (return if present) else
call _load_session_from_disk(session_id, filepath), set
self._cache[session_id]=session and return; if file does not exist create
SessionData(session_id=...) under self._lock, store and return. Ensure you
reference _load_session_from_disk, self._lock, self._cache, FileLock,
self.lock_timeout, session_id and filepath.


def _load_session_from_disk(self, session_id: str, filepath: str) -> SessionData:
"""Load session JSON from disk (caller must hold FileLock)."""
Expand Down
22 changes: 18 additions & 4 deletions src/praisonai-agents/tests/unit/session/test_session_store.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -405,14 +405,28 @@ def test_invalidate_cache(self, temp_store):
with open(filepath, "w") as f:
json.dump(data, f)

# Without invalidation, cache returns old data
# Reads reload from disk when the session file exists
history = temp_store.get_chat_history("session-1")
assert len(history) == 1 # Cached
# After invalidation, new data is loaded
assert len(history) == 2

# invalidate_cache still clears in-memory state for non-existent files
temp_store.invalidate_cache("session-1")
history = temp_store.get_chat_history("session-1")
assert len(history) == 2

def test_get_chat_history_sees_other_store_instance(self, temp_store):
"""Another store on the same session_dir must see new messages without invalidate_cache."""
with tempfile.TemporaryDirectory() as tmpdir:
writer = DefaultSessionStore(session_dir=tmpdir)
reader = DefaultSessionStore(session_dir=tmpdir)

writer.add_user_message("session-1", "first")
reader._load_session("session-1")
writer.add_user_message("session-1", "second")

history = reader.get_chat_history("session-1")
assert len(history) == 2
assert history[1]["content"] == "second"
Comment on lines +417 to +429
Copy link
Copy Markdown
Contributor

@coderabbitai coderabbitai Bot May 28, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

Add agent-level regression coverage for this feature path.

This new regression test is useful, but it only validates store internals. Per repo policy, this feature also needs a real agentic test (agent.start() with a real prompt and text response) to verify end-to-end behavior.

As per coding guidelines, "Real agentic tests are MANDATORY for every feature: Agent must call agent.start() with a real prompt, call the LLM, and produce actual text response—not just smoke tests of object construction".

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/praisonai-agents/tests/unit/session/test_session_store.py` around lines
417 - 429, The unit test test_get_chat_history_sees_other_store_instance only
exercises DefaultSessionStore internals; add an end-to-end agentic regression
that uses Agent.start() to exercise the same cross-process session visibility:
create a temporary session_dir, instantiate an Agent (or two Agent instances)
configured to use DefaultSessionStore(session_dir=tmpdir), call agent.start()
with a real prompt to produce an actual text response (hitting the LLM), then
instantiate a separate Agent or DefaultSessionStore pointing at the same
session_dir and assert get_chat_history(session_id) contains the messages
produced by the first agent (use DefaultSessionStore._load_session /
get_chat_history to verify the new message is visible). Ensure the test uses
real agent.start() invocation and validates text response content and session
persistence.


def test_list_sessions_with_none_updated_at(self, temp_store):
"""Test list_sessions handles None updated_at values without crashing.
Expand Down
Loading