feat(compression): configurable protect_first_n/protect_last_n turns

Add protect_first_n and protect_last_n to the compression config section
in config.yaml, allowing users to control how many initial and recent
turns are preserved during context compression.

- Default values: protect_first_n=3, protect_last_n=4 (no behavior change)
- Values clamped to 0-12 range (inspired by openclaw recentTurnsPreserve)
- Config version bumped to 6 for migration
- Also improved run_agent.py to read compression config from config.yaml
  (previously only read from env vars)

Related: #525 (microcompact)
This commit is contained in:
teknium1 2026-03-09 02:17:12 -07:00
parent a2d0d07109
commit 27e25c5419
3 changed files with 93 additions and 7 deletions

View file

@ -89,6 +89,8 @@ DEFAULT_CONFIG = {
"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 @@ DEFAULT_CONFIG = {
"command_allowlist": [],
# Config schema version - bump this when adding new required fields
"_config_version": 5,
"_config_version": 6,
}
# =============================================================================

View file

@ -582,16 +582,32 @@ class AIAgent:
# 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,

View file

@ -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()