feat(compression): configurable protect_first_n/protect_last_n turns (reopen #756)

hermes_cli/config.py

@@ -89,6 +89,8 @@ def ensure_hermes_home():
         "threshold": 0.85,
         "summary_model": "google/gemini-3-flash-preview",
         "summary_provider": "auto",
+        "protect_first_n": 3,  # Number of initial turns to always preserve during compression
+        "protect_last_n": 4,   # Number of recent turns to always preserve during compression
     },
 
     # Auxiliary model overrides (advanced).  By default Hermes auto-selects
@@ -166,7 +168,7 @@ def ensure_hermes_home():
     "command_allowlist": [],
 
     # Config schema version - bump this when adding new required fields
-    "_config_version": 5,
+    "_config_version": 6,
 }
 
 # =============================================================================

run_agent.py

@@ -582,16 +582,32 @@ def __init__(
         # Initialize context compressor for automatic context management
         # Compresses conversation when approaching model's context limit
         # Configuration via config.yaml (compression section) or environment variables
-        compression_threshold = float(os.getenv("CONTEXT_COMPRESSION_THRESHOLD", "0.85"))
+        compression_config = {}
+        try:
+            from hermes_cli.config import load_config as _load_compression_config
+            compression_config = _load_compression_config().get("compression", {})
+        except Exception:
+            pass
+
+        compression_threshold = float(os.getenv(
+            "CONTEXT_COMPRESSION_THRESHOLD",
+            str(compression_config.get("threshold", 0.85))
+        ))
         compression_enabled = os.getenv("CONTEXT_COMPRESSION_ENABLED", "true").lower() in ("true", "1", "yes")
-        compression_summary_model = os.getenv("CONTEXT_COMPRESSION_MODEL") or None
-
+        if not compression_enabled:
+            compression_enabled = compression_config.get("enabled", True)
+        compression_summary_model = os.getenv("CONTEXT_COMPRESSION_MODEL") or compression_config.get("summary_model") or None
+
+        # Configurable turn protection (clamped 0-12, inspired by openclaw recentTurnsPreserve)
+        protect_first_n = max(0, min(12, int(compression_config.get("protect_first_n", 3))))
+        protect_last_n = max(0, min(12, int(compression_config.get("protect_last_n", 4))))
+
         self.context_compressor = ContextCompressor(
             model=self.model,
             threshold_percent=compression_threshold,
-            protect_first_n=3,
-            protect_last_n=4,
-            summary_target_tokens=500,
+            protect_first_n=protect_first_n,
+            protect_last_n=protect_last_n,
+            summary_target_tokens=2500,
             summary_model_override=compression_summary_model,
             quiet_mode=self.quiet_mode,
             base_url=self.base_url,

tests/test_compression_config.py

@@ -0,0 +1,68 @@
+"""Tests for configurable compaction protection turns."""
+
+import unittest
+from unittest.mock import patch, MagicMock
+
+
+class TestCompressionConfigDefaults(unittest.TestCase):
+    """Verify DEFAULT_CONFIG includes protect_first_n / protect_last_n."""
+
+    def test_default_config_has_protection_fields(self):
+        from hermes_cli.config import DEFAULT_CONFIG
+        compression = DEFAULT_CONFIG["compression"]
+        self.assertIn("protect_first_n", compression)
+        self.assertIn("protect_last_n", compression)
+
+    def test_default_values(self):
+        from hermes_cli.config import DEFAULT_CONFIG
+        compression = DEFAULT_CONFIG["compression"]
+        self.assertEqual(compression["protect_first_n"], 3)
+        self.assertEqual(compression["protect_last_n"], 4)
+
+    def test_config_version_bumped(self):
+        from hermes_cli.config import DEFAULT_CONFIG
+        self.assertGreaterEqual(DEFAULT_CONFIG["_config_version"], 6)
+
+
+class TestContextCompressorAcceptsConfig(unittest.TestCase):
+    """Verify ContextCompressor properly receives custom protection values."""
+
+    @patch("agent.context_compressor.get_text_auxiliary_client")
+    def test_custom_protection_values(self, mock_aux):
+        mock_aux.return_value = (None, "test-model")
+        from agent.context_compressor import ContextCompressor
+        compressor = ContextCompressor(
+            model="test/model",
+            protect_first_n=5,
+            protect_last_n=8,
+        )
+        self.assertEqual(compressor.protect_first_n, 5)
+        self.assertEqual(compressor.protect_last_n, 8)
+
+    @patch("agent.context_compressor.get_text_auxiliary_client")
+    def test_default_protection_values(self, mock_aux):
+        mock_aux.return_value = (None, "test-model")
+        from agent.context_compressor import ContextCompressor
+        compressor = ContextCompressor(model="test/model")
+        self.assertEqual(compressor.protect_first_n, 3)
+        self.assertEqual(compressor.protect_last_n, 4)
+
+
+class TestProtectionClamping(unittest.TestCase):
+    """Verify protection values are clamped to 0-12 range."""
+
+    def test_clamp_negative_to_zero(self):
+        val = max(0, min(12, -5))
+        self.assertEqual(val, 0)
+
+    def test_clamp_over_max_to_twelve(self):
+        val = max(0, min(12, 50))
+        self.assertEqual(val, 12)
+
+    def test_valid_value_unchanged(self):
+        val = max(0, min(12, 7))
+        self.assertEqual(val, 7)
+
+
+if __name__ == "__main__":
+    unittest.main()