diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py index e59f9467ac236..c732de975cac7 100644 --- a/plugins/memory/openviking/__init__.py +++ b/plugins/memory/openviking/__init__.py @@ -2716,6 +2716,18 @@ def _mark_session_committed(self, sid: str) -> None: with self._committed_session_lock: self._committed_session_ids.add(sid) + def _clear_session_committed(self, sid: str) -> None: + """Re-arm the commit guard for a session that is still live. + + A permanent per-sid latch is correct for a session being left behind: + it dedupes that id's ``_finalize_session_async`` against the commit + compression already performed. In-place compression keeps the *same* + id, so the latch would otherwise reject every later commit for a + session that is still accumulating turns (#74695). + """ + with self._committed_session_lock: + self._committed_session_ids.discard(sid) + def _pending_session_dir(self) -> Optional[Path]: if not self._hermes_home: return None @@ -4182,6 +4194,20 @@ def on_session_switch( self._profile_prefetched_sessions.discard(old_session_id) self._profile_prefetched_sessions.discard(new_id) + if not rotate and old_session_id: + # In-place compression (the default) keeps the same session id. + # compress_context() has just committed it, latching the guard — + # but the session is still live, so every later commit for it + # (the next compression, /new, normal session end, startup + # recovery) would be rejected and post-compression turns would + # never be extracted. Re-arm the guard now that compression has + # finished; turns arriving after this point are genuinely new. + # + # Rotation mode is untouched: there a fresh child id is minted + # and the old id stays latched, which is what dedupes its + # _finalize_session_async against this same commit. + self._clear_session_committed(old_session_id) + if not rotate: # Same-session rewind (/undo) or no-op rotation: no commit and no # counter reset. diff --git a/tests/plugins/memory/test_openviking_provider.py b/tests/plugins/memory/test_openviking_provider.py index 2b8d92e61d84b..6494556ad8b08 100644 --- a/tests/plugins/memory/test_openviking_provider.py +++ b/tests/plugins/memory/test_openviking_provider.py @@ -993,3 +993,96 @@ def post(self, path, payload=None, **kwargs): assert "target_uri" not in payload + + +def test_in_place_compression_rearms_commit_guard(): + """Post-compression turns must still be committable (#74695). + + ``compress_context()`` commits before rewriting the transcript, which + latches the per-sid guard. In-place mode (the default) keeps the SAME sid, + so the latch then rejected every later commit for a still-live session — + the next compression, /new, normal session end and startup recovery all + silently did nothing, and post-compression turns were never extracted. + """ + provider = _make_provider_with_session("sid-123", turn_count=4) + provider._ensure_client = lambda: True + + # Compression commits the live session, latching the guard. + provider._mark_session_committed("sid-123") + assert provider._session_needs_commit("sid-123", 4) is False + + # In-place compression: same id in, no rotation. + provider.on_session_switch("sid-123", reason="compression") + + # The session is still live, so new turns must be committable again. + assert provider._has_committed_session("sid-123") is False + assert provider._session_needs_commit("sid-123", 2) is True + + +def test_rotating_compression_keeps_old_session_latched(): + """Rotation mode must keep the guard, which dedupes the old id's finalize. + + With ``compression.in_place: false`` a fresh child id is minted. The old id + stays committed so its ``_finalize_session_async`` does not double-commit + what compression already committed — the behavior the guard exists for. + """ + provider = _make_provider_with_session("old-sid", turn_count=4) + provider._ensure_client = lambda: True + provider._finalize_session_async = MagicMock() + + provider._mark_session_committed("old-sid") + provider.on_session_switch("new-sid", reason="compression") + + assert provider._has_committed_session("old-sid") is True + assert provider._session_needs_commit("old-sid", 4) is False + + +def test_undo_rewind_does_not_rearm_commit_guard(): + """Only compression re-arms; a same-session /undo must not.""" + provider = _make_provider_with_session("sid-123", turn_count=4) + provider._ensure_client = lambda: True + + provider._mark_session_committed("sid-123") + provider.on_session_switch("sid-123", rewound=True) + + assert provider._has_committed_session("sid-123") is True + + +def test_in_place_compression_lifecycle_allows_a_later_commit(monkeypatch): + """End-to-end wiring, not a hand-set latch (#74695). + + Drives the real sequence a session goes through: commit at the compression + boundary, same-id ``on_session_switch``, a post-compression turn via + ``sync_turn``, then a later commit. Before the fix the second commit never + reached the server, so every turn after the first compression was lost. + """ + provider = _make_provider_with_session("sid-123", turn_count=3) + provider._ensure_client = lambda: True + provider._drain_writers = lambda sid, timeout=None: True + # Keep the async write worker out of it; the counter bump is what matters. + monkeypatch.setattr(provider, "_queue_memory_write", lambda *a, **k: None, raising=False) + + def _commit_calls(): + return [ + c for c in provider._client.post.call_args_list + if c.args and str(c.args[0]).endswith("/commit") + ] + + # 1. Compression commits the live session through the real path. + provider.on_session_end([{"role": "user", "content": "before"}]) + assert len(_commit_calls()) == 1 + assert provider._has_committed_session("sid-123") is True + + # 2. In-place compression: same id back in, no rotation. + provider.on_session_switch("sid-123", reason="compression") + + # 3. A genuinely new turn lands on the still-live session. + provider.sync_turn("after compression", "reply", session_id="sid-123") + assert provider._turn_count > 0 + + # 4. That turn must still be committable. + provider.on_session_end([{"role": "user", "content": "after"}]) + assert len(_commit_calls()) == 2, ( + "post-compression turns were never committed: " + f"{provider._client.post.call_args_list}" + )