fix(compression): skip session rotation when compress returns message…

agent/conversation_compression.py

@@ -482,6 +482,20 @@ def _release_lock() -> None:
         _release_lock()  # compression aborted — no rotation will happen
         return messages, _existing_sp
 
+    # If compress() returned messages unchanged without setting
+    # _last_compress_aborted (e.g. compress_start >= compress_end),
+    # there is nothing to rotate — skip session rotation entirely.
+    # Guard on both length (no reduction) AND savings_pct == 0 so that
+    # a legitimate compression that happens to produce the same message
+    # count (e.g. summary replaces many messages 1:1) is not blocked.
+    _savings_pct = getattr(agent.context_compressor, "_last_compression_savings_pct", 100.0)
+    if len(compressed) >= _pre_msg_count and _savings_pct == 0.0:
+        _existing_sp = getattr(agent, "_cached_system_prompt", None)
+        if not _existing_sp:
+            _existing_sp = agent._build_system_prompt(system_message)
+        _release_lock()  # no-op compression — no rotation will happen
+        return messages, _existing_sp
+
     summary_error = getattr(agent.context_compressor, "_last_summary_error", None)
     if summary_error:
         if getattr(agent, "_last_compression_summary_warning", None) != summary_error:

tests/run_agent/test_infinite_compaction_loop.py

@@ -248,3 +248,44 @@ def test_below_threshold_allows(self):
         comp = _make_compressor(config_context_length=96000)
         comp.last_prompt_tokens = 10_000
         assert not comp.should_compress(10_000)
+
+
+class TestNoOpCompressionSkipsRotation:
+    """When compress() returns messages unchanged (no-op), session rotation
+    must be skipped — no end_session, no new session_id."""
+
+    def test_no_op_compression_does_not_rotate_session(self):
+        """compress_start >= compress_end path must not trigger session rotation."""
+        from unittest.mock import MagicMock
+        from agent.conversation_compression import compress_context
+
+        messages = _build_session(10, words_per_turn=10)
+
+        agent = MagicMock()
+        agent.session_id = "test-no-op-session"
+        agent.compression_in_place = False
+        agent.compression_enabled = True
+        agent._compression_feasibility_checked = True
+        agent._cached_system_prompt = "system prompt"
+        agent._memory_manager = None
+        agent._todo_store = MagicMock()
+        agent._todo_store.format_for_injection.return_value = ""
+
+        session_db = MagicMock()
+        session_db.try_acquire_compression_lock.return_value = True
+        agent._session_db = session_db
+
+        compressor = MagicMock()
+        compressor._last_compress_aborted = False
+        compressor._last_summary_error = None
+        compressor._last_aux_model_failure_model = None
+        compressor._last_compression_savings_pct = 0.0
+        compressor.compress.return_value = list(messages)
+        agent.context_compressor = compressor
+
+        result_messages, _ = compress_context(agent, messages, "system", approx_tokens=1000)
+
+        session_db.end_session.assert_not_called()
+        assert agent.session_id == "test-no-op-session", (
+            "Session ID must not change when compression is a no-op"
+        )