diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index e11dc7c171d3..9e435454a0d1 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -380,21 +380,45 @@ def compress_context( agent.commit_memory_session(messages) agent._session_db.end_session(agent.session_id, "compression") old_session_id = agent.session_id - agent.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}" + new_session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}" + + # Retry create_session to survive transient SQLite lock contention. + # Without this, the new session_id is set but state.db has no row, + # producing an orphan session invisible in the WebUI sidebar (#33906). + import sqlite3 as _sqlite3 + import time as _time + + _created = False + for _attempt in range(3): + try: + agent._session_db.create_session( + session_id=new_session_id, + source=agent.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"), + model=agent.model, + model_config=agent._session_init_model_config, + parent_session_id=old_session_id, + ) + _created = True + break + except _sqlite3.OperationalError: + if _attempt < 2: + _time.sleep(0.1 * (_attempt + 1)) + else: + raise + + if not _created: + # Should be unreachable (the loop either succeeds or raises), + # but guard against logic errors. + raise RuntimeError("create_session retry loop exited without success") + + # Only commit the session_id rotation AFTER state.db write succeeds. + agent.session_id = new_session_id try: from gateway.session_context import set_current_session_id set_current_session_id(agent.session_id) except Exception: os.environ["HERMES_SESSION_ID"] = agent.session_id - agent._session_db_created = False - agent._session_db.create_session( - session_id=agent.session_id, - source=agent.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"), - model=agent.model, - model_config=agent._session_init_model_config, - parent_session_id=old_session_id, - ) agent._session_db_created = True # Auto-number the title for the continuation session if old_title: @@ -408,6 +432,19 @@ def compress_context( agent._last_flushed_db_idx = 0 except Exception as e: logger.warning("Session DB compression split failed — new session will NOT be indexed: %s", e) + # Roll back session_id so the agent stays on the old, registered session + # instead of continuing with an orphan session_id that state.db knows nothing about. + _rollback_sid = locals().get("old_session_id") + if _rollback_sid and agent.session_id != _rollback_sid: + logger.info("Rolling back session_id to %s after compression split failure", _rollback_sid) + agent.session_id = _rollback_sid + # Reopen the old session in state.db — end_session() already marked it + # as ended, but we're staying on it, so it must be active again. + if _rollback_sid: + try: + agent._session_db.reopen_session(_rollback_sid) + except Exception: + logger.debug("Could not reopen old session %s after rollback", _rollback_sid, exc_info=True) # Notify the context engine that the session_id rotated because of # compression (not a fresh /new). Plugin engines (e.g. hermes-lcm) use diff --git a/tests/agent/test_compression_orphan_session.py b/tests/agent/test_compression_orphan_session.py new file mode 100644 index 000000000000..8d6e28a4d0d7 --- /dev/null +++ b/tests/agent/test_compression_orphan_session.py @@ -0,0 +1,217 @@ +"""Regression tests for #33906 — compression split must not create orphan sessions. + +When context compression triggers a session rotation, the old session is +ended in state.db and a new one is created. If ``create_session()`` fails +(e.g. transient SQLite lock contention), the agent must roll back its +``session_id`` to the old value so it never runs with an orphan session that +has no state.db row. +""" + +from __future__ import annotations + +import sqlite3 +import time as _time +from datetime import datetime +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_agent(session_id: str = "20260101_120000_abc123"): + """Build a minimal agent-like object for compress_context.""" + agent = MagicMock() + # Use real attribute (not mock) so assignment sticks + agent.session_id = session_id + agent.platform = "test" + agent.model = "test-model" + agent._session_init_model_config = {} + agent._session_db = MagicMock() + agent._session_db.get_session_title.return_value = "Test Session" + agent._session_db.create_session.return_value = session_id + agent._session_db_created = True + agent._cached_system_prompt = None + agent._last_flushed_db_idx = 100 + agent._todo_store = MagicMock() + agent._todo_store.format_for_injection.return_value = "" + agent._invalidate_system_prompt = MagicMock() + agent._build_system_prompt = MagicMock(return_value="system prompt") + agent.commit_memory_session = MagicMock() + agent.context_compressor = MagicMock() + agent.context_compressor.compress = MagicMock(return_value=[{"role": "user", "content": "compressed"}]) + agent.context_compressor._last_compress_aborted = False + agent.context_compressor.compression_count = 0 + agent._memory_manager = MagicMock() + agent._gateway_session_key = None + agent._emit_status = MagicMock() + agent._emit_warning = MagicMock() + agent._compression_feasibility_checked = True + return agent + + +# --------------------------------------------------------------------------- +# Test 1: Normal compression — session_id rotates successfully +# --------------------------------------------------------------------------- + +class TestCompressionSessionRotationSuccess: + """Happy path: create_session succeeds, session_id rotates.""" + + def test_session_id_rotates_on_success(self): + from agent.conversation_compression import compress_context + + agent = _make_agent("old_session") + old_sid = agent.session_id + + messages = [{"role": "user", "content": "hello"}] * 50 + + with patch.dict("os.environ", {"HERMES_SESSION_SOURCE": "test"}): + compress_context(agent, messages, system_message="sys") + + # session_id must have changed + assert agent.session_id != old_sid + # old session was ended + agent._session_db.end_session.assert_called_once_with(old_sid, "compression") + # new session was created with parent_session_id + create_call = agent._session_db.create_session.call_args + assert create_call.kwargs["parent_session_id"] == old_sid + # flush cursor reset + assert agent._last_flushed_db_idx == 0 + + +# --------------------------------------------------------------------------- +# Test 2: Retry on sqlite3.OperationalError +# --------------------------------------------------------------------------- + +class TestCompressionRetryOnLock: + """create_session retries on transient SQLite lock errors.""" + + def test_retries_on_operational_error(self): + from agent.conversation_compression import compress_context + + agent = _make_agent("lock_session") + + # Fail first 2 attempts, succeed on 3rd + call_count = [0] + original_create = agent._session_db.create_session + + def _flaky_create(**kwargs): + call_count[0] += 1 + if call_count[0] <= 2: + raise sqlite3.OperationalError("database is locked") + return kwargs["session_id"] + + agent._session_db.create_session = _flaky_create + + messages = [{"role": "user", "content": "hello"}] * 50 + + with patch.dict("os.environ", {"HERMES_SESSION_SOURCE": "test"}): + compress_context(agent, messages, system_message="sys") + + # All 3 attempts used, session rotated + assert call_count[0] == 3 + assert agent.session_id != "lock_session" + + def test_raises_after_3_retries_exhausted(self): + from agent.conversation_compression import compress_context + + agent = _make_agent("exhaust_session") + + def _always_fail(**kwargs): + raise sqlite3.OperationalError("database is locked") + + agent._session_db.create_session = _always_fail + + messages = [{"role": "user", "content": "hello"}] * 50 + + with patch.dict("os.environ", {"HERMES_SESSION_SOURCE": "test"}): + compress_context(agent, messages, system_message="sys") + + # After failure, session_id should be rolled back to original + assert agent.session_id == "exhaust_session" + # Old session should have been reopened + agent._session_db.reopen_session.assert_called_once_with("exhaust_session") + + +# --------------------------------------------------------------------------- +# Test 3: Rollback session_id on non-SQLite exceptions +# --------------------------------------------------------------------------- + +class TestCompressionRollbackOnFailure: + """session_id rolls back when create_session fails for any reason.""" + + def test_rollback_on_generic_exception(self): + from agent.conversation_compression import compress_context + + agent = _make_agent("rollback_session") + + def _fail(**kwargs): + raise IOError("disk full") + + agent._session_db.create_session = _fail + + messages = [{"role": "user", "content": "hello"}] * 50 + + with patch.dict("os.environ", {"HERMES_SESSION_SOURCE": "test"}): + compress_context(agent, messages, system_message="sys") + + # session_id must be rolled back + assert agent.session_id == "rollback_session" + # Old session reopened + agent._session_db.reopen_session.assert_called_once_with("rollback_session") + + def test_no_rollback_when_old_session_id_not_set(self): + """If end_session itself fails, old_session_id isn't in locals; + rollback must not crash.""" + from agent.conversation_compression import compress_context + + agent = _make_agent("early_fail") + # Make end_session fail so old_session_id is never assigned + agent._session_db.end_session.side_effect = RuntimeError("boom") + # create_session should never be called + agent._session_db.create_session = MagicMock( + side_effect=AssertionError("should not be called") + ) + + messages = [{"role": "user", "content": "hello"}] * 50 + + # Should not raise — the outer try/except catches it + with patch.dict("os.environ", {"HERMES_SESSION_SOURCE": "test"}): + compress_context(agent, messages, system_message="sys") + + # session_id unchanged (no old_session_id was set to trigger rollback) + assert agent.session_id == "early_fail" + + +# --------------------------------------------------------------------------- +# Test 4: Session_id is set AFTER successful create_session (not before) +# --------------------------------------------------------------------------- + +class TestCompressionSessionIdOrdering: + """Verify session_id is only mutated after state.db write succeeds.""" + + def test_session_id_not_changed_if_create_fails(self): + from agent.conversation_compression import compress_context + + agent = _make_agent("order_session") + captured_sid = [] + + def _failing_create(**kwargs): + # Capture what session_id the agent has at create time + captured_sid.append(agent.session_id) + raise sqlite3.OperationalError("locked") + + agent._session_db.create_session = _failing_create + + messages = [{"role": "user", "content": "hello"}] * 50 + + with patch.dict("os.environ", {"HERMES_SESSION_SOURCE": "test"}): + compress_context(agent, messages, system_message="sys") + + # At the time create_session was called, session_id should still + # be the old value (not the new generated one) + assert captured_sid[0] == "order_session" + # After failure, session_id is rolled back + assert agent.session_id == "order_session"