Repository navigation
feat(nodes): add persistent cross-session memory node with Redis and in-memory backends - #547
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a persistent cross-session memory node with pluggable backends (in-memory or Redis), session lifecycle and key/value APIs with history and auto-summarization, pipeline integration via IGlobal/IInstance, service metadata, and comprehensive tests. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant IInstance
participant IGlobal
participant Store as PersistentMemoryStore
participant Backend
Client->>IInstance: writeQuestions(question)
IInstance->>IGlobal: read store
alt store exists
IInstance->>Store: resume_session(session_id)
Store->>Backend: resume_session(session_id)
Backend-->>Store: session metadata / keys
Store-->>IInstance: key-values
IInstance->>IInstance: attach memory_context to question
end
IInstance-->>Client: forward (enriched) question
Client->>IInstance: writeAnswers(answer)
IInstance->>IGlobal: read store
alt store exists
IInstance->>Store: resume/create session
IInstance->>Store: put(session_id, "last_answer", text)
Store->>Backend: put(session_id,...)
Store-->>IInstance: ok
IInstance->>Store: increment(session_id, "answer_count")
Store->>Backend: increment(...)
alt history length > max_history
Store->>Store: summarize_if_needed(session_id)
Store->>Backend: replace_history(...)
end
end
IInstance-->>Client: forward answer
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@nodes/src/nodes/memory_persistent/IGlobal.py`:
- Line 34: Replace the string concatenation used to build the requirements path
(the assignment to the variable requirements) with os.path.join to mirror other
modules (e.g., agent_langchain/IGlobal.py); compute the base dir with
os.path.dirname(os.path.realpath(__file__)) and then call os.path.join(base_dir,
'requirements.txt') so path construction is consistent and portable across
platforms.
- Around line 54-57: The endGlobal method currently just nulls self.store and
self.config but when backend='redis' this doesn't close the Redis connection;
update cleanup so endGlobal calls a proper close method on the store/backend
(e.g., implement and call PersistentMemoryStore.close() or RedisBackend.close()
which invokes redis_client.close() or connection_pool.disconnect()), ensure
endGlobal detects if self.store has a close attribute and calls it (or
explicitly checks for RedisBackend) before setting self.store = None, and add
the corresponding close() implementation on RedisBackend/PersistentMemoryStore
to cleanly shutdown the redis.Redis client.
In `@nodes/src/nodes/memory_persistent/IInstance.py`:
- Around line 106-109: The current read-modify-write for answer_count
(store.get(session_id, 'answer_count') then store.put(... current_count + 1)) is
non-atomic and can lose increments under concurrent writes; update the
implementation in the writeAnswers flow to use an atomic increment API on the
store (e.g., call store.incr(session_id, 'answer_count') or
store.atomic_increment/INCR when using Redis) or add an atomic_increment method
to the in-memory store implementation and call that instead of get/put to ensure
increments are safe under concurrency.
In `@nodes/src/nodes/memory_persistent/memory_store.py`:
- Around line 600-611: The summarize_if_needed code is breaking encapsulation by
inspecting backend types and touching private attributes; add an abstract method
replace_history(session_id: str, entries: List[Dict[str, Any]]) -> None to the
MemoryBackend ABC, implement replace_history in InMemoryBackend (acquire its
lock and set _history[session_id] = entries) and in RedisBackend (build a
pipeline, delete and rpush the serialized entries, execute), then update
summarize_if_needed to call backend.replace_history(session_id, [summary_entry]
+ recent_entries) instead of using isinstance checks or accessing
_lock/_history/_client/_history_key directly.
- Line 602: Replace the list concatenation used to update the backend history
with list unpacking for clarity and performance: where the code assigns
self._backend._history[session_id] = [summary_entry] + recent_entries, change it
to construct the new list using the summary_entry and the recent_entries
iterable (using the unpacking form) so the update uses [summary_entry,
*recent_entries]; reference symbols: self._backend._history, session_id,
summary_entry, recent_entries.
- Around line 285-290: The loop that appends data keys one-by-one is
inefficient; in memory_store.py replace the manual loop that iterates over
members and appends to keys_to_expire with a single list.extend call (e.g.,
extend with a list comprehension that maps each member to
self._data_key(session_id, member)), then iterate keys_to_expire to call
self._client.expire(k, ttl_seconds); update references to members,
keys_to_expire, _data_key(session_id, ...), and _client.expire accordingly.
- Around line 390-394: When putting a new key in MemoryPersistentStore (in the
put() flow that calls _get_ttl_seconds and _client.expire on
_data_key(session_id, key)), also refresh the TTL on the session structural keys
(__keys__, __history__, __meta__) so they don't expire before newly added data;
implement this by calling _client.expire for each structural key (e.g. the Redis
keys you use to store the key set, history list and meta hash) with the same ttl
returned by _get_ttl_seconds(session_id) whenever ttl is truthy, and add a brief
comment/docstring noting that put() now "touches" session TTL on activity.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: e83f7be9-b861-4082-8d6d-7f95ebc0b1a6
📒 Files selected for processing (7)
nodes/src/nodes/memory_persistent/IGlobal.pynodes/src/nodes/memory_persistent/IInstance.pynodes/src/nodes/memory_persistent/__init__.pynodes/src/nodes/memory_persistent/memory_store.pynodes/src/nodes/memory_persistent/requirements.txtnodes/src/nodes/memory_persistent/services.jsontest/nodes/test_memory_persistent.py
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@nodes/src/nodes/memory_persistent/IInstance.py`:
- Around line 55-60: The node currently lets ValueError from
store.resume_session(session_id) and store.create_session(session_id) (triggered
by malformed question.metadata['session_id'] or answer.metadata['session_id'])
bubble up and fail the pipeline; wrap those calls (both the session handling
block around session_id and the similar block at lines ~100-104) in a try/except
that catches ValueError and simply skips memory/session operations, leaving the
payload unchanged, or alternatively validate session_id before calling
resume_session/create_session and skip on invalid values; ensure you reference
the same store, session_id, resume_session and create_session call sites so both
places behave the same.
In `@nodes/src/nodes/memory_persistent/memory_store.py`:
- Around line 116-123: Session expiry is only enforced in resume_session() and
list_sessions(), so create a small helper (e.g.,
_expire_session_if_needed(session_id: str) or _expire_sessions_check()) that,
while holding self._lock, checks a session's 'expires_at' against time.time()
and removes it from self._sessions when expired; call this helper at the start
of every public session method (create_session, resume_session, put, get,
list_keys, clear, get_history, list_sessions) so expired in-memory sessions are
pruned consistently and free up the _MAX_INMEMORY_SESSIONS slot; ensure the
helper uses the same session-id validation (_validate_session_id) and updates
any related state consistently under self._lock.
- Around line 238-241: The replace_history implementation for replace_history
currently assigns a shallow copy via list(entries), violating the backend's
deep-copy contract; change it to perform a deep copy of the entries (e.g., use
copy.deepcopy(entries)) before storing into self._history[session_id] under the
existing self._lock so subsequent mutations to caller-provided dicts don't
affect the stored history, and add the necessary import for copy if missing.
- Around line 383-408: The put() path can recreate session data for a stale
session ID because it only checks _REDIS_SESSIONS_KEY before writing; change the
guard in put() to validate the session metadata key via
self._meta_key(session_id) (ensure the meta key exists and that
self._client.pttl(self._meta_key(session_id)) returns > 0) before performing
SET/SADD/RPUSH so expired sessions are rejected; if the meta is missing or ttl
<= 0 return the same {'ok': False, 'error': ...} as used for missing sessions
and avoid creating _data_key, _keys_key, or _history_key, and apply the same
metadata-based check to other public Redis operations that currently only
consult _REDIS_SESSIONS_KEY.
In `@test/nodes/test_memory_persistent.py`:
- Around line 285-294: The helpers _make_backend and _ensure_engine_mocks are
directly mutating sys.modules and leaving MagicMocks in place; change them to
use pytest's monkeypatch (or unittest.mock.patch.dict) to temporarily set
sys.modules entries (e.g., monkeypatch.setitem(sys.modules, 'redis',
mock_redis_module)) so the mocked modules are automatically restored after each
test, and apply the same change to the other helper region referenced (lines
778-809) where 'engLib', 'depends', or 'ai.common.*' are injected; ensure you
remove direct assignments to sys.modules and instead use monkeypatch/patch.dict
for all mocked imports.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: cb42903f-f108-4363-a724-a81e67455b5b
📒 Files selected for processing (4)
nodes/src/nodes/memory_persistent/IGlobal.pynodes/src/nodes/memory_persistent/IInstance.pynodes/src/nodes/memory_persistent/memory_store.pytest/nodes/test_memory_persistent.py
|
Thanks for this contribution — the storage layer is solid, the backend abstraction is clean, and the test coverage is impressive. One question before I dig deeper into the review: how is continuity between runs achieved? Specifically, Could you clarify:
This is central to the feature working end-to-end, so I want to make sure I'm not missing something before completing the review. |
|
Great question — here's how session continuity works: How The if hasattr(question, 'metadata') and isinstance(question.metadata, dict):
session_id = question.metadata.get('session_id')You're correct that the core question.metadata = {'session_id': 'user-abc-session-1'}The continuity flow:
If no This design means the client controls session identity — the node is a stateful middleware that enriches the pipeline with memory context when a session key is provided. |
Response to asclearuc's Session ID Continuity QuestionGreat question — this is indeed central to the feature working correctly. Let me clarify the design: 1. How does session_id get into the question before reaching this node?The # Client/upstream code creates a Question with metadata
question = Question(
text='What is the capital of France?',
metadata={'session_id': 'user-session-abc123'} # ← Set by client
)The When this question reaches the
2. What ensures the same ID is used across separate pipeline runs?The client must provide the same Example: If the client sends a different session_id in Run 2, it starts a fresh session with no prior context. Important caveat: DESIGN GAPHowever, I found a critical issue during review: there's no guarantee the When the LLM node generates an answer, it doesn't automatically copy # Question arrives with session_id
question.metadata = {'session_id': 'user-session-abc123', 'memory_context': {...}}
# LLM node processes and returns answer
answer = llm.generate(question)
# answer.metadata is empty — session_id is LOST!
# writeAnswers() looks for session_id in answer.metadata
# Doesn't find it → answer never gets stored in the sessionThis causes silent data loss — answers are discarded if downstream nodes don't know to preserve RecommendationThis should be fixed before merge. Options:
Suggest implementing #1 + #2 to guarantee continuity even if downstream nodes don't cooperate. |
|
Hi team — friendly ping for a re-review when you get a chance. All 12 CodeRabbit suggestions have been addressed, 84 tests pass, and the session_id question has been answered in the thread. This one should be ready to go. Thanks! |
|
This enables stateful pipelines — users can build chatbots with cross-session memory. 84 tests pass, all 12 review items addressed. Ready for final review when you have a moment. |
|
Given the design gap around |
…ride-org#547) - Add atomic increment() method: InMemoryBackend uses lock-protected increment, RedisBackend uses native INCRBY for true atomicity - Simplify IInstance answer_count to use store.increment() instead of manual lock + get + put - Use list unpacking in RedisBackend.delete_session() for clarity - Remove unused threading import from IInstance Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Review Feedback AddressedAll Major CodeRabbit issues have been implemented: Major Fixes
Minor Fixes
Ready for review. |
|
No description provided. |
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (2)
nodes/src/nodes/memory_persistent/memory_store.py (2)
425-450:⚠️ Potential issue | 🟠 MajorUse the first
PTTLresult for the write path.The check at Line 425 only proves the meta key existed before the mutation. If it expires before
pipe.execute(), the second lookup at Line 446 returns-2and the freshly written data/history keys are left without any TTL. Queue thePEXPIREcalls in the same pipeline using the originalremaining_msvalue so the mutation and expiry stay consistent at the boundary.🛠️ Suggested change
remaining_ms = self._client.pttl(self._meta_key(session_id)) if remaining_ms == -2: self._client.srem(_REDIS_SESSIONS_KEY, session_id) return {'ok': False, 'error': f'Session {session_id!r} has expired'} serialized = json.dumps(value) pipe = self._client.pipeline() pipe.set(self._data_key(session_id, key), serialized) pipe.sadd(self._keys_key(session_id), key) pipe.rpush( self._history_key(session_id), json.dumps( { 'op': 'put', 'key': key, 'timestamp': time.time(), } ), ) - pipe.execute() - # Align new key's expiry with the session metadata key's - # remaining TTL so it doesn't outlive the session. - remaining_ms = self._client.pttl(self._meta_key(session_id)) - if remaining_ms and remaining_ms > 0: - self._client.pexpire(self._data_key(session_id, key), remaining_ms) - self._client.pexpire(self._keys_key(session_id), remaining_ms) - self._client.pexpire(self._history_key(session_id), remaining_ms) + if remaining_ms and remaining_ms > 0: + pipe.pexpire(self._data_key(session_id, key), remaining_ms) + pipe.pexpire(self._keys_key(session_id), remaining_ms) + pipe.pexpire(self._history_key(session_id), remaining_ms) + pipe.execute()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@nodes/src/nodes/memory_persistent/memory_store.py` around lines 425 - 450, The current code checks remaining_ms via self._client.pttl(self._meta_key(session_id)) before piping the writes but then re-reads TTL after pipe.execute(), allowing a race where the meta key could expire and leave the new keys without TTL; fix by capturing the first remaining_ms (from the initial pttl call), enqueue the PEXPIRE calls on the same pipeline (pipe.pexpire(..., remaining_ms)) for self._data_key(session_id, key), self._keys_key(session_id), and self._history_key(session_id) before pipe.execute(), and only run those pexpire pipeline entries when remaining_ms is positive (use that captured value instead of re-calling pttl or using self._client.pexpire after execute).
136-143:⚠️ Potential issue | 🟠 MajorPrune expired sessions before enforcing
_MAX_INMEMORY_SESSIONS.
create_session()checks duplicate/cap state without reclaiming expired sessions first. Once enough TTL'd sessions age out, new sessions can still fail with the cap error, and an expired ID can't be recreated until another code path happens to prune it.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@nodes/src/nodes/memory_persistent/memory_store.py` around lines 136 - 143, create_session currently checks for duplicates and the _MAX_INMEMORY_SESSIONS cap before reclaiming expired sessions; call the session-pruning logic at the start of create_session so expired entries are removed prior to duplicate/cap checks. Specifically, in create_session(session_id: str, ...) invoke the existing prune method (e.g. self._prune_expired_sessions() or equivalent cleanup routine) or inline the expiration cleanup before checking session_id in self._sessions and comparing len(self._sessions) to _MAX_INMEMORY_SESSIONS, so an expired session can be recreated and capacity is calculated after reclamation.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@nodes/src/nodes/memory_persistent/IInstance.py`:
- Around line 28-29: The node currently extracts session_id in writeQuestions
but re-derives it in writeAnswers from answer.metadata, causing continuity loss
if metadata isn't copied; modify the class (IInstance) to store a request-scoped
attribute (e.g., self._current_session_id) when writeQuestions extracts the
session (use the same extraction logic currently in writeQuestions), have
writeAnswers use self._current_session_id as a fallback when answer.metadata
lacks session info, and ensure open(self, _obj: Entry) clears this attribute
(set it to None) to reset per-request state; update references in writeQuestions
and writeAnswers so they read/write this single attribute instead of re-deriving
the session.
In `@nodes/src/nodes/memory_persistent/memory_store.py`:
- Around line 603-609: The TTL is being truncated by int(...) in
PersistentMemoryStore.__init__ (attribute _ttl_seconds), losing sub-second
precision and creating 0s; change the assignment to preserve fractional seconds
(e.g., set self._ttl_seconds: Optional[float] = session_ttl_hours * 3600 if
session_ttl_hours > 0 else None) and update any code paths in InMemoryBackend
and RedisBackend that read _ttl_seconds to accept a float (or explicitly convert
to the backend-required units like milliseconds/ceil to 1s where appropriate) so
small TTLs (e.g., ~0.36s) are handled correctly.
---
Duplicate comments:
In `@nodes/src/nodes/memory_persistent/memory_store.py`:
- Around line 425-450: The current code checks remaining_ms via
self._client.pttl(self._meta_key(session_id)) before piping the writes but then
re-reads TTL after pipe.execute(), allowing a race where the meta key could
expire and leave the new keys without TTL; fix by capturing the first
remaining_ms (from the initial pttl call), enqueue the PEXPIRE calls on the same
pipeline (pipe.pexpire(..., remaining_ms)) for self._data_key(session_id, key),
self._keys_key(session_id), and self._history_key(session_id) before
pipe.execute(), and only run those pexpire pipeline entries when remaining_ms is
positive (use that captured value instead of re-calling pttl or using
self._client.pexpire after execute).
- Around line 136-143: create_session currently checks for duplicates and the
_MAX_INMEMORY_SESSIONS cap before reclaiming expired sessions; call the
session-pruning logic at the start of create_session so expired entries are
removed prior to duplicate/cap checks. Specifically, in
create_session(session_id: str, ...) invoke the existing prune method (e.g.
self._prune_expired_sessions() or equivalent cleanup routine) or inline the
expiration cleanup before checking session_id in self._sessions and comparing
len(self._sessions) to _MAX_INMEMORY_SESSIONS, so an expired session can be
recreated and capacity is calculated after reclamation.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 26e22269-5e4a-4ea3-9265-73f70b6d405a
📒 Files selected for processing (3)
nodes/src/nodes/memory_persistent/IInstance.pynodes/src/nodes/memory_persistent/memory_store.pytest/nodes/test_memory_persistent.py
🚀 Merge RequestAll Major CodeRabbit issues addressed:
Ready for review and merge. |
|
@asclearuc @stepmikhaylov — This PR has all CodeRabbit feedback addressed (counter race condition, TTL drift, expired session enforcement, deep-copy fix, test isolation). 73/73 tests passing. Would appreciate a review when you get a chance. Thanks! 🙏 |
|
@asclearuc I addressed the concrete continuity hole inside the node in 101441b by carrying |
|
@asclearuc I pushed a narrow fix for the two concrete implementation bugs: preserving question-to-answer session continuity inside the node, and preserving fractional TTL precision. That said, I agree those changes do not fully settle the broader contract issue you called out around |
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@nodes/src/nodes/memory_persistent/memory_store.py`:
- Around line 453-462: The get method currently only checks membership in
_REDIS_SESSIONS_KEY and can return data for a session whose meta key expired;
update memory_persistent.memory_store.MemoryStore.get to mirror the put logic by
also fetching the meta key TTL (use PTTL on the meta key used in put, e.g.
self._meta_key(session_id) or the same meta key logic used in put) and returning
an error when PTTL indicates the meta key is missing/expired (<= -2) or has no
TTL (-1) as appropriate; keep the existing sismember check, then if meta TTL
shows expired return the same not-found structure as put, otherwise proceed to
read the data key and JSON-decode it as currently done.
- Around line 99-101: The abstract base's close() method currently has an empty
body which triggers Ruff B027; make the no-op intentional by either adding an
explicit pass statement inside close() or add a short comment like "Optional
override for backends that need cleanup; no-op by default" so readers and
linters understand it's deliberate — update the close() method on the ABC (the
close() implementation in the memory store class) accordingly.
- Around line 326-340: In _set_ttl replace the redundant int(round(...)) with a
math.ceil-based conversion to milliseconds to avoid precision loss for very
small TTLs: change ttl_ms calculation to use ttl_ms = max(1,
math.ceil(ttl_seconds * 1000)), add an import for math if missing, and keep the
rest of the logic (building keys_to_expire, fetching members via
_client.smembers, and calling _client.pexpire) unchanged so every session key
gets at least 1ms TTL.
- Around line 565-570: The close() method currently swallows all exceptions;
update MemoryStore.close (the close method in memory_store.py) to stop using a
bare except: pass — either wrap the close call with
contextlib.suppress(Exception) or catch Exception as e and call the module/class
logger (e.g., self.logger or a module-level logger) to log the exception at
debug level before returning; ensure you reference self._client.close() in the
change so cleanup still proceeds but failures are observable.
In `@test/nodes/test_memory_persistent.py`:
- Around line 810-824: The pytest fixture engine_mocks currently uses yield but
has no teardown logic, so replace the final "yield entries" with "return
entries" to satisfy Ruff PT022; locate the engine_mocks fixture and change its
return mechanism (which builds entries via _build_engine_mock_entries and
installs them into sys.modules using monkeypatch) to return the entries after
performing the monkeypatch setup, leaving the existing mock-install and
sys.modules clearing logic unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 73ba8b24-1763-4d5c-8cb9-15c44e7b71cf
📒 Files selected for processing (3)
nodes/src/nodes/memory_persistent/IInstance.pynodes/src/nodes/memory_persistent/memory_store.pytest/nodes/test_memory_persistent.py
…tride-org#547) - os.path.join for cross-platform path construction - Explicit Redis close()/disconnect() on teardown (try/finally) - Atomic counter: INCR for Redis, threading.Lock for in-memory - list.extend instead of loop-append in memory_store - TTL inheritance: late-added keys inherit remaining session TTL Tests added for TTL inheritance, concurrent increment, Redis teardown. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (1)
nodes/src/nodes/memory_persistent/memory_store.py (1)
450-522:⚠️ Potential issue | 🟠 MajorExpiry-check inconsistency across Redis read paths can feed orphaned history back through summarization.
put()(line 423-426) andincrement()(line 530-533) both guard againstpttl(meta_key) == -2and evict the stale session_id. The read paths here —get(450-459),list_keys(461-467),clear(469-512), and especiallyget_history(514-522) — only checksismember(_REDIS_SESSIONS_KEY, …). Because_REDIS_SESSIONS_KEYitself never has a TTL, a session whose meta/keys/history were evicted by Redis TTL can still:
- Return stale values from
get/list_keys(previously raised forgetin past review, still open).- Feed a non-empty entry list from
get_historyintoPersistentMemoryStore.summarize_if_needed→RedisBackend.replace_history(lines 549-560). At that pointpttl(meta_key)is-2, theremaining_ms > 0guard skipspexpire, and the pipeline recreateshistory_keywith no TTL — i.e., a persistent orphan on a logically-expired session.Reusing the same
pttlguard in all read paths (or factoring it into a_require_live_sessionhelper called from every public method) closes the loop and keeps the "session has expired" semantics uniform.🔧 Suggested consolidation
+ def _require_live_session(self, session_id: str) -> Optional[Dict[str, Any]]: + """Return an error dict if session is missing or its meta has expired; None if live.""" + if not self._client.sismember(_REDIS_SESSIONS_KEY, session_id): + return {'ok': False, 'error': f'Session {session_id!r} not found'} + if self._client.pttl(self._meta_key(session_id)) == -2: + self._client.srem(_REDIS_SESSIONS_KEY, session_id) + return {'ok': False, 'error': f'Session {session_id!r} has expired'} + return NoneThen replace the four
sismember-only guards inget/list_keys/clear/get_historywitherr = self._require_live_session(session_id); if err: return err, and add an early-return toreplace_historyas well so summarization can't resurrect an expired session's history list.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@nodes/src/nodes/memory_persistent/memory_store.py` around lines 450 - 522, The read paths get, list_keys, clear, and get_history currently only check sismember(_REDIS_SESSIONS_KEY, session_id) and can resurrect expired sessions; add a shared helper _require_live_session(session_id) that checks pttl(self._meta_key(session_id)) and treats pttl == -2 as expired (returning the same {'ok': False, 'error': ...} shape), call this helper at the start of get, list_keys, clear, and get_history (return early on error), and also add an early-return in replace_history (used by summarize_if_needed/RedisBackend.replace_history) so it does not recreate history keys for sessions whose meta_key pttl == -2. Ensure the helper mirrors the eviction logic used in put and increment (which currently check pttl(meta_key) == -2) and reference _REDIS_SESSIONS_KEY, _meta_key/_history_key, and the summarize_if_needed/replace_history call sites when applying the change.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@nodes/src/nodes/memory_persistent/memory_store.py`:
- Around line 92-97: The docstring for the base-class method increment is
misleading about atomicity: change it to state clearly that the default
implementation uses a non-atomic get+put sequence (calling get and put) and
therefore is not safe for concurrent updates, and instruct subclass authors to
override increment to provide an atomic implementation where needed (see
InMemoryBackend.increment and RedisBackend.increment as examples of atomic
overrides); mention the involved symbols increment, get, and put so maintainers
can find and update the docstring accordingly.
- Around line 549-560: replace_history currently skips session validation and
can create orphaned history lists; call self._validate_session_id(session_id) at
the start of replace_history and perform the same live-session existence check
used elsewhere (e.g., verify membership in the sessions set via
self._client.sismember(...) or check TTL via
self._client.pttl(self._meta_key(session_id))) and abort/raise if the session is
not live before executing the pipeline (so _history_key(session_id) is never
created for deleted/expired sessions); keep the existing TTL-preservation logic
using _meta_key and _history_key if the session is valid.
In `@nodes/test/memory_persistent/test_memory_store.py`:
- Around line 331-336: The test test_redis_backend_close_swallows_errors
currently enforces that backend.close() must not raise, which prevents changing
RedisBackend.close() to use contextlib.suppress + a debug log; update the test
to still ensure no exception is propagated but also assert that the error is
logged: keep the fake client raising (backend._client.close =
MagicMock(side_effect=RuntimeError('boom'))) then call backend.close() inside a
no-raise check and use pytest's caplog to assert a debug/error message was
emitted (matching "boom" or "RedisBackend.close") so the test allows refactoring
to contextlib.suppress with observability in RedisBackend.close().
---
Duplicate comments:
In `@nodes/src/nodes/memory_persistent/memory_store.py`:
- Around line 450-522: The read paths get, list_keys, clear, and get_history
currently only check sismember(_REDIS_SESSIONS_KEY, session_id) and can
resurrect expired sessions; add a shared helper
_require_live_session(session_id) that checks pttl(self._meta_key(session_id))
and treats pttl == -2 as expired (returning the same {'ok': False, 'error': ...}
shape), call this helper at the start of get, list_keys, clear, and get_history
(return early on error), and also add an early-return in replace_history (used
by summarize_if_needed/RedisBackend.replace_history) so it does not recreate
history keys for sessions whose meta_key pttl == -2. Ensure the helper mirrors
the eviction logic used in put and increment (which currently check
pttl(meta_key) == -2) and reference _REDIS_SESSIONS_KEY, _meta_key/_history_key,
and the summarize_if_needed/replace_history call sites when applying the change.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: ae07eeb0-0324-421f-959c-0f3b9f02d193
📒 Files selected for processing (4)
nodes/src/nodes/memory_persistent/IGlobal.pynodes/src/nodes/memory_persistent/memory_store.pynodes/test/memory_persistent/__init__.pynodes/test/memory_persistent/test_memory_store.py
|
@asclearuc @stepmikhaylov — marked ready for review. Applied the remaining CodeRabbit items:
12 new tests covering TTL alignment, concurrent increment, teardown race. 356/356 passing. Ready for initial review. |
Add a cross-session persistent memory node with pluggable backends (Redis for production, in-memory for testing). The node intercepts questions to enrich them with stored session context and persists answers for future retrieval. Key features: - Session management: create, resume, list, delete - Keyed storage: put, get, list_keys, clear per session - Auto-summarization when history exceeds configurable limit - TTL enforcement for session expiry - Thread-safe with deep copy on get/put to prevent mutation - Input validation to prevent Redis key injection/path traversal - InMemory backend bounded to 1000 sessions max Includes 79 tests covering session CRUD, memory operations, session isolation, mocked Redis backend, auto-summarization, history retrieval, TTL enforcement, thread safety, deep copy mutation prevention, and IGlobal/IInstance lifecycle. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Use os.path.join for requirements path instead of string concatenation for cross-platform path safety - Add proper Redis connection cleanup in endGlobal to prevent connection leaks on pipeline shutdown - Make answer_count increment atomic with a threading lock to prevent lost updates from concurrent threads - Replace manual loop with list.extend for keys_to_expire in Redis backend _set_ttl method Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Fix Redis TTL drift: use PTTL on metadata key to align new key expiry with session remaining TTL instead of reapplying full original TTL - Fix encapsulation: add replace_history() abstract method to MemoryBackend so summarize_if_needed() no longer reaches into backend internals - Fix Redis cleanup: add close() method to MemoryBackend (no-op base, explicit client.close() in RedisBackend), use from IGlobal.endGlobal() - Use list unpacking [summary, *recent] instead of concatenation - Batch Redis rpush calls in replace_history() for efficiency - Add tests for TTL alignment, replace_history, close, and Redis cleanup Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…tride-org#547) - os.path.join for cross-platform path construction - Explicit Redis close()/disconnect() on teardown (try/finally) - Atomic counter: INCR for Redis, threading.Lock for in-memory - list.extend instead of loop-append in memory_store - TTL inheritance: late-added keys inherit remaining session TTL Tests added for TTL inheritance, concurrent increment, Redis teardown. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
26495a1 to
4cbefeb
Compare
|
@nihalnihalani unit test failed |
|
@dylan-savage - as I know you are working on it. |
|
Could you please review the new commit? |
Review statusAll actionable CodeRabbit findings are addressed in code:
CI is green on b019664. PR is not draft. Requesting first-pass human review from @jmaionchi @Rod-Christensen @stepmikhaylov @kwit75 — this is the persistent cross-session memory node (Redis + in-memory backends). |
stepmikhaylov
left a comment
stepmikhaylov
left a comment
There was a problem hiding this comment.
Request Changes
Mark the node as experimental in capabilities
New nodes require the "experimental" gate before promotion.
Add a test section to services.json
Consider adding configuration defined tests. See docs/README-node-testing.md.
Remove sys.path manipulation from both test files
The built-in interpreter configures all paths — inserting NODES_SRC, ROCKETLIB, and AI_SRC manually is not needed and not allowed by project conventions.
📄 test_memory_store.py, test_node.py
Don't mock interpreter-provided modules (rocketlib, ai)
rocketlib.IGlobalBase, rocketlib.OPEN_MODE, and ai.common.* are replaced with MagicMock across both test files. These are built-in — import the real modules. The importlib isolated-load approach in test_memory_store.py is the root cause; replacing it with a standard import resolves both the sys.path and mock issues there.
📄 test_memory_store.py, test_node.py
Move FakeRedis / FakePipeline to nodes/test/mocks/
Node-specific mocks defined inline in a test file should live in the shared mocks directory instead.
Remove documentation changes from docs/adr
This location is not suitable for node documentation.
|
Thanks for adding the persistent memory node with Redis and in-memory backends! Please address the review comments above before merging. |
…e-org#547) stepmikhaylov's CHANGES_REQUESTED review of 2026-05-19. Each point: 1. Mark node experimental in capabilities — services.json now has "capabilities": ["experimental"]. 2. Add a test section to services.json — added a minimal profile-driven case per docs/README-node-testing.md (forwards a question with no session_id unchanged). 3. Remove sys.path manipulation from both test files — the NODES_SRC / ROCKETLIB / AI_SRC inserts at the top of test_memory_store.py and test_node.py are gone. Path resolution to nodes/test/mocks/ is now handled in a local conftest.py (conftest, not test files). 4. Don't mock rocketlib / ai — the importlib spec_from_file_location shim, the engine_mocks fixture, and the per-test patch.dict over sys.modules['rocketlib' | 'ai' | 'ai.common.*'] are all removed. The tests now use standard imports: from nodes.memory_persistent.IGlobal import IGlobal from nodes.memory_persistent.IInstance import IInstance from nodes.memory_persistent.memory_store import ... The brittle requirements-path regression test that mocked `depends` was dropped — the os.path.join usage is self-evident in IGlobal.py. 5. Move FakeRedis / FakePipeline to nodes/test/mocks/ — extracted to nodes/test/mocks/redis_fake/__init__.py, renamed to drop the underscore prefix and consumed via "from redis_fake import FakeRedis". 6. Remove docs/adr changes — deleted 0002-persistent-memory-session- contract.md and reverted the ADR index entry in docs/adr/README.md. Tests cannot be run locally — rocketlib imports engLib which only exists after a builder build into dist/server — so first signal will be CI. Ruff check + ruff format are clean on all touched files.
Maintainer review (stepmikhaylov, 2026-05-19) — point-by-point resolutionJust pushed
Verification
Note on CodeRabbit threadsAll 23 prior CodeRabbit inline threads are already resolved (confirmed at the GraphQL Ready for re-review when you have a moment. cc @asclearuc @dylan-savage |
My previous conftest.py at nodes/test/memory_persistent/ used sys.path.insert(0, mocks_dir), which prepended the shared mocks directory to sys.path for the entire pytest session. This caused `import weaviate` to resolve to the incomplete shadow mock at nodes/test/mocks/weaviate/ (which has no `client` submodule) instead of the real weaviate library in dist/server, breaking 6 unrelated weaviate tests. The shadow-style mocks in nodes/test/mocks/ are only meant to be activated subprocess-side via ROCKETRIDE_MOCK. At unit-test collection time the real libraries from dist/server must keep priority. Switching to sys.path.append leaves real-library resolution unchanged and only makes `redis_fake` (a name that doesn't collide with any real or shadow module) importable.
The local conftest.py added nodes/test/mocks/ to sys.path at collection
time. Because pytest loads every conftest at session start, that path
mutation was global and broke 6 unrelated weaviate tests
(ModuleNotFoundError: No module named 'weaviate.client') — their own
stub save/restore logic conflicts with mocks/ sitting on sys.path.
The node test runner invokes pytest with --rootdir=nodes and targets
nodes/test (nodes/scripts/tasks.js). Since nodes/__init__.py does not
exist but nodes/test/__init__.py does, pytest's prepend import mode puts
nodes/ on sys.path and imports the suite as the `test` package. So the
fake is reachable as a normal package import with no path hacks:
from test.mocks.redis_fake import FakeRedis
Deleted the conftest entirely. This both restores the weaviate tests and
fully satisfies the maintainer's "no sys.path manipulation" requirement.
3 participants
#Hack-with-bay-2
Contribution Type
Feature — Persistent cross-session memory for chatbots and agents
Problem / Use Case
The existing `memory_internal` node at `nodes/src/nodes/memory_internal/` only stores data within a single pipeline run. When the pipeline ends, all memory is lost. Chatbots and agents need to remember previous conversations across sessions — this is table stakes for production AI applications.
Proposed Solution
A `memory_persistent` node with:
Why This Feature Fits This Codebase
The node follows the exact same architecture as `memory_internal` — same `put/get/list/clear` operations exposed via the pipeline lane system. The key difference is persistence: `memory_internal` uses a plain Python dict that dies with the process; `memory_persistent` uses Redis (production) or a bounded in-memory store (testing).
The `IInstance.writeQuestions()` method loads session context and attaches it to `question.metadata['memory_context']` — making it available to downstream LLM nodes the same way the existing `memory_internal` node exposes data. The `writeAnswers()` method stores answers for future session retrieval.
Session ID validation (`^[a-zA-Z0-9_\-]{1,128}$`) prevents path traversal and Redis key injection — following the same security patterns as the guardrails node and the dataset loader's path validation.
Validation
```
79 passed
ruff check: 0 errors
ruff format: all files unchanged
```
79 tests across 14 classes: validation (11), session CRUD (8), memory ops (9), session isolation (3), Redis mocked (9), auto-summarization (7), history (5), TTL (4), thread safety (3), deep copy (3), facade (5), bounds (1), IGlobal (2), IInstance (11).
How This Could Be Extended
🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Tests