feat: support fixed token threshold and per-model compression settings

agent/context_compressor.py

@@ -381,6 +381,7 @@ def update_model(
         api_key: str = "",
         provider: str = "",
         api_mode: str = "",
+        threshold_tokens: int | None = None,
     ) -> None:
         """Update model info after a model switch or fallback activation."""
         self.model = model
@@ -389,10 +390,16 @@ def update_model(
         self.provider = provider
         self.api_mode = api_mode
         self.context_length = context_length
-        self.threshold_tokens = max(
-            int(context_length * self.threshold_percent),
-            MINIMUM_CONTEXT_LENGTH,
-        )
+        if threshold_tokens is not None:
+            self.threshold_tokens = max(
+                threshold_tokens,
+                MINIMUM_CONTEXT_LENGTH,
+            )
+        else:
+            self.threshold_tokens = max(
+                int(context_length * self.threshold_percent),
+                MINIMUM_CONTEXT_LENGTH,
+            )
         # Recalculate token budgets for the new context length so the
         # compressor stays calibrated after a model switch (e.g. 200K → 32K).
         target_tokens = int(self.threshold_tokens * self.summary_target_ratio)
@@ -405,6 +412,7 @@ def __init__(
         self,
         model: str,
         threshold_percent: float = 0.50,
+        threshold_tokens: int | None = None,
         protect_first_n: int = 3,
         protect_last_n: int = 20,
         summary_target_ratio: float = 0.20,
@@ -432,14 +440,22 @@ def __init__(
             config_context_length=config_context_length,
             provider=provider,
         )
+        # Resolve threshold_tokens: if an absolute value is given, use it
+        # directly; otherwise derive from threshold_percent × context_length.
         # Floor: never compress below MINIMUM_CONTEXT_LENGTH tokens even if
         # the percentage would suggest a lower value.  This prevents premature
         # compression on large-context models at 50% while keeping the % sane
         # for models right at the minimum.
-        self.threshold_tokens = max(
-            int(self.context_length * threshold_percent),
-            MINIMUM_CONTEXT_LENGTH,
-        )
+        if threshold_tokens is not None:
+            self.threshold_tokens = max(
+                threshold_tokens,
+                MINIMUM_CONTEXT_LENGTH,
+            )
+        else:
+            self.threshold_tokens = max(
+                int(self.context_length * threshold_percent),
+                MINIMUM_CONTEXT_LENGTH,
+            )
         self.compression_count = 0
 
         # Derive token budgets: ratio is relative to the threshold, not total context
@@ -450,12 +466,16 @@ def __init__(
         )
 
         if not quiet_mode:
+            if threshold_tokens is not None:
+                _pct_display = "fixed"
+            else:
+                _pct_display = f"{threshold_percent * 100:.0f}%"
             logger.info(
                 "Context compressor initialized: model=%s context_length=%d "
-                "threshold=%d (%.0f%%) target_ratio=%.0f%% tail_budget=%d "
+                "threshold=%d (%s) target_ratio=%.0f%% tail_budget=%d "
                 "provider=%s base_url=%s",
                 model, self.context_length, self.threshold_tokens,
-                threshold_percent * 100, self.summary_target_ratio * 100,
+                _pct_display, self.summary_target_ratio * 100,
                 self.tail_token_budget,
                 provider or "none", base_url or "none",
             )
@@ -1426,9 +1446,8 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f
                 self.threshold_tokens,
             )
             logger.info(
-                "Model context limit: %d tokens (%.0f%% = %d)",
+                "Model context limit: %d tokens (threshold=%d)",
                 self.context_length,
-                self.threshold_percent * 100,
                 self.threshold_tokens,
             )
             tail_msgs = n_messages - compress_end

hermes_cli/config.py

@@ -728,6 +728,8 @@ def _ensure_hermes_home_managed(home: Path):
     "compression": {
         "enabled": True,
         "threshold": 0.50,            # compress when context usage exceeds this ratio
+        "threshold_tokens": None,     # optional absolute token threshold (overrides threshold %)
+        "per_model": {},              # per-model overrides: {"model-name": {"threshold": 0.75} or {"threshold_tokens": 120000}}
         "target_ratio": 0.20,         # fraction of threshold to preserve as recent tail
         "protect_last_n": 20,         # minimum recent messages to keep uncompressed
         "hygiene_hard_message_limit": 400,  # gateway session-hygiene force-compress threshold by message count
@@ -4800,7 +4802,21 @@ def show_config():
     enabled = compression.get('enabled', True)
     print(f"  Enabled:      {'yes' if enabled else 'no'}")
     if enabled:
-        print(f"  Threshold:    {compression.get('threshold', 0.50) * 100:.0f}%")
+        _threshold_tokens = compression.get('threshold_tokens')
+        _threshold_pct = compression.get('threshold', 0.50)
+        if _threshold_tokens:
+            print(f"  Threshold:    {_threshold_tokens:,} tokens (fixed)")
+        else:
+            print(f"  Threshold:    {_threshold_pct * 100:.0f}% of context")
+        _per_model = compression.get('per_model', {})
+        if _per_model:
+            for _mname, _mcfg in _per_model.items():
+                _mt = _mcfg.get('threshold_tokens')
+                _mp = _mcfg.get('threshold')
+                if _mt:
+                    print(f"    ├ {_mname}: {_mt:,} tokens (fixed)")
+                elif _mp:
+                    print(f"    ├ {_mname}: {_mp * 100:.0f}% of context")
         print(f"  Target ratio: {compression.get('target_ratio', 0.20) * 100:.0f}% of threshold preserved")
         print(f"  Protect last: {compression.get('protect_last_n', 20)} messages")
         _aux_comp = config.get('auxiliary', {}).get('compression', {})

hermes_cli/setup.py

@@ -1747,18 +1747,67 @@ def setup_agent_settings(config: dict):
 
     config.setdefault("compression", {})["enabled"] = True
 
+    # Ask about percentage threshold or fixed token threshold
     current_threshold = cfg_get(config, "compression", "threshold", default=0.50)
-    threshold_str = prompt("Compression threshold (0.5-0.95)", str(current_threshold))
-    try:
-        threshold = float(threshold_str)
-        if 0.5 <= threshold <= 0.95:
-            config["compression"]["threshold"] = threshold
-    except ValueError:
-        pass
-
-    print_success(
-        f"Context compression threshold set to {config['compression'].get('threshold', 0.50)}"
+    current_tt = config.get("compression", {}).get("threshold_tokens")
+    use_fixed = current_tt is not None
+    threshold_type = prompt(
+        "Threshold type: (p)ercentage of context or (f)ixed token count",
+        "f" if use_fixed else "p",
     )
+    if threshold_type.lower().startswith("f"):
+        current_tt_str = str(current_tt) if current_tt else ""
+        tt_str = prompt("Fixed threshold in tokens (e.g. 120000)", current_tt_str)
+        try:
+            tt_val = int(tt_str)
+            if tt_val > 0:
+                config["compression"]["threshold_tokens"] = tt_val
+                # Clear percentage-based threshold to avoid confusion
+                config["compression"].pop("threshold", None)
+        except (ValueError, TypeError):
+            pass
+    else:
+        threshold_str = prompt("Compression threshold (0.5-0.95)", str(current_threshold))
+        try:
+            threshold = float(threshold_str)
+            if 0.5 <= threshold <= 0.95:
+                config["compression"]["threshold"] = threshold
+                # Clear fixed token threshold
+                config["compression"].pop("threshold_tokens", None)
+        except ValueError:
+            pass
+
+    print_success("Compression threshold configured")
+
+    # Ask about per-model overrides
+    per_model_config = config.setdefault("compression", {}).setdefault("per_model", {})
+    add_per_model = prompt("Add per-model compression overrides? (y/N)", "n")
+    if add_per_model.lower().startswith("y"):
+        while True:
+            model_name = prompt("Model name (e.g. deepseek-chat) or empty to finish", "")
+            if not model_name:
+                break
+            model_entry = {}
+            tt_str = prompt("Fixed token threshold (e.g. 120000, or empty to skip)", "")
+            try:
+                tt_val = int(tt_str)
+                if tt_val > 0:
+                    model_entry["threshold_tokens"] = tt_val
+            except (ValueError, TypeError):
+                pass
+            if "threshold_tokens" not in model_entry:
+                pct_str = prompt("Percentage threshold (e.g. 0.75, or empty to skip)", "")
+                try:
+                    pct_val = float(pct_str)
+                    if 0.0 < pct_val <= 1.0:
+                        model_entry["threshold"] = pct_val
+                except (ValueError, TypeError):
+                    pass
+            if model_entry:
+                per_model_config[model_name] = model_entry
+                print_success(f"  Added override for {model_name}")
+            else:
+                print_warning("  No valid settings provided, skipped")
 
     # ── Session Reset Policy ──
     print_header("Session Reset Policy")

run_agent.py

@@ -2116,13 +2116,42 @@ def __init__(
         if not isinstance(_compression_cfg, dict):
             _compression_cfg = {}
         compression_threshold = float(_compression_cfg.get("threshold", 0.50))
+        compression_threshold_tokens: int | None = None
         try:
             from agent.auxiliary_client import _compression_threshold_for_model as _cthresh_fn
             _model_cthresh = _cthresh_fn(self.model)
             if _model_cthresh is not None:
                 compression_threshold = _model_cthresh
         except Exception:
             pass
+
+        # Per-model compression overrides from config.yaml
+        _per_model_cfg: dict = _compression_cfg.get("per_model", {}) or {}
+        if isinstance(_per_model_cfg, dict) and self.model in _per_model_cfg:
+            _model_override = _per_model_cfg[self.model]
+            if isinstance(_model_override, dict):
+                _mt = _model_override.get("threshold_tokens")
+                if _mt is not None:
+                    try:
+                        compression_threshold_tokens = int(_mt)
+                    except (TypeError, ValueError):
+                        pass
+                _mp = _model_override.get("threshold")
+                if _mp is not None:
+                    try:
+                        compression_threshold = float(_mp)
+                    except (TypeError, ValueError):
+                        pass
+
+        # Global threshold_tokens overrides the percentage when per-model doesn't specify
+        if compression_threshold_tokens is None:
+            _global_tt = _compression_cfg.get("threshold_tokens")
+            if _global_tt is not None:
+                try:
+                    compression_threshold_tokens = int(_global_tt)
+                except (TypeError, ValueError):
+                    pass
+
         compression_enabled = str(_compression_cfg.get("enabled", True)).lower() in {"true", "1", "yes"}
         compression_target_ratio = float(_compression_cfg.get("target_ratio", 0.20))
         compression_protect_last = int(_compression_cfg.get("protect_last_n", 20))
@@ -2319,13 +2348,15 @@ def __init__(
                 base_url=self.base_url,
                 api_key=getattr(self, "api_key", ""),
                 provider=self.provider,
+                threshold_tokens=compression_threshold_tokens,
             )
             if not self.quiet_mode:
                 logger.info("Using context engine: %s", _selected_engine.name)
         else:
             self.context_compressor = ContextCompressor(
                 model=self.model,
                 threshold_percent=compression_threshold,
+                threshold_tokens=compression_threshold_tokens,
                 protect_first_n=3,
                 protect_last_n=compression_protect_last,
                 summary_target_ratio=compression_target_ratio,
@@ -2750,13 +2781,40 @@ def switch_model(self, new_model, new_provider, api_key='', base_url='', api_mod
                 config_context_length=getattr(self, "_config_context_length", None),
                 custom_providers=_sm_custom_providers,
             )
+            # Re-resolve threshold_tokens for the new model (per-model config)
+            _sm_threshold_tokens: int | None = None
+            try:
+                from hermes_cli.config import load_config
+                _sm_cfg = load_config()
+                _sm_comp_cfg = _sm_cfg.get("compression", {}) if isinstance(_sm_cfg, dict) else {}
+                if isinstance(_sm_comp_cfg, dict):
+                    _sm_per_model = _sm_comp_cfg.get("per_model", {}) or {}
+                    if isinstance(_sm_per_model, dict) and self.model in _sm_per_model:
+                        _sm_mo = _sm_per_model[self.model]
+                        if isinstance(_sm_mo, dict):
+                            _sm_mt = _sm_mo.get("threshold_tokens")
+                            if _sm_mt is not None:
+                                try:
+                                    _sm_threshold_tokens = int(_sm_mt)
+                                except (TypeError, ValueError):
+                                    pass
+                    if _sm_threshold_tokens is None:
+                        _sm_global_tt = _sm_comp_cfg.get("threshold_tokens")
+                        if _sm_global_tt is not None:
+                            try:
+                                _sm_threshold_tokens = int(_sm_global_tt)
+                            except (TypeError, ValueError):
+                                pass
+            except Exception:
+                pass
             self.context_compressor.update_model(
                 model=self.model,
                 context_length=new_context_length,
                 base_url=self.base_url,
                 api_key=getattr(self, "api_key", ""),
                 provider=self.provider,
                 api_mode=self.api_mode,
+                threshold_tokens=_sm_threshold_tokens,
             )
 
         # ── Invalidate cached system prompt so it rebuilds next turn ──

tests/agent/test_context_compressor.py

@@ -38,6 +38,71 @@ def test_explicit_tokens(self, compressor):
         assert compressor.should_compress(prompt_tokens=50000) is False
 
 
+class TestThresholdTokens:
+    """Tests for threshold_tokens parameter (absolute token threshold)."""
+
+    def test_threshold_tokens_used_when_given(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=1_000_000):
+            c = ContextCompressor(
+                model="deepseek-chat",
+                threshold_percent=0.50,
+                threshold_tokens=120000,
+                quiet_mode=True,
+            )
+        assert c.threshold_tokens == 120000
+        assert c.should_compress(prompt_tokens=100000) is False
+        assert c.should_compress(prompt_tokens=120000) is True
+        assert c.should_compress(prompt_tokens=150000) is True
+
+    def test_threshold_tokens_falls_back_to_percent_when_none(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=200_000):
+            c = ContextCompressor(
+                model="claude-sonnet-4",
+                threshold_percent=0.75,
+                threshold_tokens=None,
+                quiet_mode=True,
+            )
+        assert c.threshold_tokens == 150000
+
+    def test_threshold_tokens_with_minimum_floor(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=1_000_000):
+            c = ContextCompressor(
+                model="deepseek-chat",
+                threshold_percent=0.50,
+                threshold_tokens=1000,
+                quiet_mode=True,
+            )
+        from agent.model_metadata import MINIMUM_CONTEXT_LENGTH
+        assert c.threshold_tokens >= MINIMUM_CONTEXT_LENGTH
+
+    def test_update_model_respects_threshold_tokens(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=1_000_000):
+            c = ContextCompressor(
+                model="deepseek-chat",
+                threshold_percent=0.50,
+                threshold_tokens=120000,
+                quiet_mode=True,
+            )
+        c.update_model(
+            model="deepseek-chat",
+            context_length=1_000_000,
+            threshold_tokens=80000,
+        )
+        assert c.threshold_tokens == 80000
+
+    def test_update_model_falls_back_to_percent_when_no_threshold_tokens(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=200_000):
+            c = ContextCompressor(
+                model="claude-sonnet-4",
+                threshold_percent=0.50,
+                quiet_mode=True,
+            )
+        c.update_model(
+            model="claude-sonnet-4",
+            context_length=200_000,
+        )
+        assert c.threshold_tokens == 100000
+
 
 class TestUpdateFromResponse:
     def test_updates_fields(self, compressor):