repair(compression): preserve plugin no-op session boundaries

agent/conversation_compression.py

@@ -3160,6 +3160,31 @@ def _release_lock() -> None:
             finally:
                 _release_lock()
 
+        # Plugin no-op: engine reports it made no structural change.
+        # Check this BEFORE the semantic equality test so a plugin that
+        # returns a freshly-built equal list (not the input object) is
+        # still recognized as a no-op via its status flag.
+        if (
+            getattr(
+                agent.context_compressor,
+                "last_compression_status",
+                getattr(agent.context_compressor, "_last_compression_status", ""),
+            )
+            == "noop"
+        ):
+            try:
+                logger.info(
+                    "context compression no-op: session=%s messages=%d unchanged; skipping session boundary",
+                    agent.session_id or "none",
+                    _pre_msg_count,
+                )
+                _existing_sp = getattr(agent, "_cached_system_prompt", None)
+                if not _existing_sp:
+                    _existing_sp = agent._build_system_prompt(system_message)
+                return compressed, _existing_sp
+            finally:
+                _release_lock()
+
         # Compare against the pre-dispatch semantic state, not object identity:
         # legacy/plugin engines may return an equal copy for a no-op, or mutate
         # the live list while returning an unchanged snapshot. Neither case may

tests/run_agent/test_compression_noop_status.py

@@ -0,0 +1,87 @@
+"""Regression tests for plugin-reported compression no-op boundaries.
+
+These cases intentionally use output that differs from the input. Equal-copy
+results are already covered by the semantic no-progress guard merged upstream
+in NousResearch/hermes-agent#67938; this file protects the residual contract
+where an external context engine reports ``noop`` after cleanup-only active
+context changes that must be adopted without minting a compression boundary.
+"""
+
+import sqlite3
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock
+
+from tests.run_agent.test_compression_boundary_hook import (
+    TestCompressionBoundaryHook as _BoundaryHarness,
+)
+
+
+def _noop_compressor(cleaned_messages):
+    compressor = MagicMock()
+
+    def _compress(_messages, **_kwargs):
+        compressor.last_compression_status = "noop"
+        return list(cleaned_messages)
+
+    compressor.compress.side_effect = _compress
+    compressor.compression_count = 0
+    compressor.last_prompt_tokens = 0
+    compressor.last_completion_tokens = 0
+    compressor._last_summary_error = None
+    compressor._last_compress_aborted = False
+    compressor._last_compression_made_progress = False
+    compressor._last_summary_fallback_used = False
+    compressor._last_compression_feasibility_reason = None
+    compressor.last_compression_status = ""
+    return compressor
+
+
+def test_plugin_noop_adopts_cleanup_without_session_boundary():
+    """A reported no-op may change active context without being a split."""
+    from hermes_state import SessionDB
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        db_path = Path(tmpdir) / "test.db"
+        db = SessionDB(db_path=db_path)
+        agent = _BoundaryHarness()._make_agent(db)
+
+        messages = [
+            {"role": "user", "content": "replayed scaffold"},
+            {"role": "assistant", "content": "fresh tail"},
+        ]
+        cleaned = [{"role": "assistant", "content": "fresh tail"}]
+        compressor = _noop_compressor(cleaned)
+        agent.context_compressor = compressor
+        agent._cached_system_prompt = "cached-system-prompt"
+
+        original_sid = agent.session_id
+        compressed, prompt = agent._compress_context(
+            messages,
+            "sys",
+            approx_tokens=10_000,
+        )
+
+        # This must exercise the explicit status seam, not #67938's equality
+        # guard: cleanup changed the active context.
+        assert compressed == cleaned
+        assert compressed != messages
+        assert prompt == "cached-system-prompt"
+        assert agent.session_id == original_sid
+
+        compression_boundary_calls = [
+            call
+            for call in compressor.on_session_start.call_args_list
+            if call.kwargs.get("boundary_reason") == "compression"
+        ]
+        assert not compression_boundary_calls
+
+        conn = sqlite3.connect(str(db_path))
+        try:
+            child_count = conn.execute(
+                "SELECT COUNT(*) FROM sessions WHERE parent_session_id = ?",
+                (original_sid,),
+            ).fetchone()[0]
+        finally:
+            conn.close()
+        assert child_count == 0