feat(compression): absolute token threshold (compression.threshold_tokens)

agent/agent_init.py

@@ -1839,6 +1839,18 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
         }
     else:
         compression_model_thresholds = {}
+    # Absolute token cap: when set, compression triggers at the lower of
+    # the ratio-based threshold and this absolute count. Clamped to the
+    # model's context length at apply-time so a cap above the window is
+    # a no-op (ratio-based threshold wins).
+    compression_threshold_tokens = _compression_cfg.get("threshold_tokens")
+    if compression_threshold_tokens is not None:
+        try:
+            compression_threshold_tokens = int(compression_threshold_tokens)
+            if compression_threshold_tokens <= 0:
+                compression_threshold_tokens = None
+        except (TypeError, ValueError):
+            compression_threshold_tokens = None
     # In-place compaction: when True, compress_context() rewrites the message
     # list + rebuilds the system prompt WITHOUT rotating the session id (no
     # parent_session_id chain, no `name #N` renumber). See #38763 and
@@ -2271,6 +2283,7 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
             abort_on_summary_failure=compression_abort_on_summary_failure,
             max_tokens=agent.max_tokens,
             model_thresholds=compression_model_thresholds,
+            threshold_tokens_cap=compression_threshold_tokens,
         )
     _bind_session_state = getattr(agent.context_compressor, "bind_session_state", None)
     if callable(_bind_session_state):
@@ -2482,7 +2495,11 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
             _active_threshold_pct = getattr(
                 agent.context_compressor, "threshold_percent", compression_threshold
             )
-            print(f"📊 Context limit: {agent.context_compressor.context_length:,} tokens (compress at {int(_active_threshold_pct*100)}% = {agent.context_compressor.threshold_tokens:,})")
+            _cap_note = ""
+            _cap = getattr(agent.context_compressor, "threshold_tokens_cap", None)
+            if _cap and _cap > 0:
+                _cap_note = f" (capped at {_cap:,} tokens)"
+            print(f"📊 Context limit: {agent.context_compressor.context_length:,} tokens (compress at {int(_active_threshold_pct*100)}% = {agent.context_compressor.threshold_tokens:,}{_cap_note})")
         else:
             print(f"📊 Context limit: {agent.context_compressor.context_length:,} tokens (auto-compression disabled)")
         # Notice with the exact opt-back-out command. Printed inline at startup

agent/context_compressor.py

@@ -1231,6 +1231,11 @@ def update_model(
         self.threshold_tokens = self._compute_threshold_tokens(
             context_length, self.threshold_percent, self.max_tokens,
         )
+        # Re-apply the absolute token cap so it survives model switches
+        # and fallback activations. The cap is a first-class config value
+        # stored on the compressor instance, not a one-time post-construction
+        # patch — this is why update_model() must re-apply it.
+        self._apply_threshold_tokens_cap()
         # 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)
@@ -1293,6 +1298,36 @@ def _coerce_max_tokens(value: Any) -> int | None:
             return None
         return ivalue if ivalue > 0 else None
 
+    @staticmethod
+    def _coerce_threshold_tokens_cap(value: Any) -> int | None:
+        """Normalize a threshold_tokens cap to a positive int or None.
+
+        None means "no absolute cap — use the ratio-based threshold only".
+        Non-numeric or non-positive values are treated as None so a bad
+        config value never silently caps the threshold at zero.
+        """
+        if value is None:
+            return None
+        try:
+            ivalue = int(value)
+        except (TypeError, ValueError):
+            return None
+        return ivalue if ivalue > 0 else None
+
+    def _apply_threshold_tokens_cap(self) -> None:
+        """Apply the absolute token cap if configured.
+
+        After ``threshold_tokens`` is (re)computed from the ratio-based
+        percent, clamp it to the cap so compression never fires later
+        than the user's preferred absolute token count. The cap itself
+        is clamped to the current context length so a cap larger than
+        the model's window is a no-op (the ratio-based threshold wins).
+        """
+        if self.threshold_tokens_cap is not None and self.threshold_tokens_cap > 0:
+            _effective_cap = min(self.threshold_tokens_cap, self.context_length)
+            if _effective_cap < self.threshold_tokens:
+                self.threshold_tokens = _effective_cap
+
     @staticmethod
     def _effective_threshold_percent(
         context_length: int, threshold_percent: float,
@@ -1368,6 +1403,7 @@ def __init__(
         abort_on_summary_failure: bool = False,
         max_tokens: int | None = None,
         model_thresholds: dict[str, float] | None = None,
+        threshold_tokens_cap: Any = None,
     ):
         self.model = model
         self.base_url = base_url
@@ -1387,6 +1423,14 @@ def __init__(
             model, self.model_thresholds, threshold_percent,
         )
         self.threshold_percent = self._base_threshold_percent
+        # Absolute token cap from config (compression.threshold_tokens). When
+        # set, the effective trigger point is min(ratio-based threshold, cap)
+        # so compression never fires later than the user's preferred token
+        # count regardless of which model is active. Applied in __init__ and
+        # re-applied in update_model() so it survives model switches/fallbacks.
+        self.threshold_tokens_cap = self._coerce_threshold_tokens_cap(
+            threshold_tokens_cap,
+        )
         self.protect_first_n = protect_first_n
         self.protect_last_n = protect_last_n
         self.summary_target_ratio = max(0.10, min(summary_target_ratio, 0.80))
@@ -1432,6 +1476,9 @@ def __init__(
         self.threshold_tokens = self._compute_threshold_tokens(
             self.context_length, threshold_percent, self.max_tokens,
         )
+        # Apply absolute token cap (compression.threshold_tokens) — takes
+        # the lower of the ratio-based threshold and the cap.
+        self._apply_threshold_tokens_cap()
         self.compression_count = 0
 
         # Derive token budgets: ratio is relative to the threshold, not total context

cli-config.yaml.example

@@ -424,6 +424,15 @@ compression:
   #   "claude-sonnet": 0.35
   #   "gpt-5": 0.30
 
+  # Optional absolute token cap for the compression trigger (default: null = disabled).
+  # When set, compression fires at the LOWER of the ratio-based threshold and this
+  # absolute token count — first-fires-wins. It never fires later than this count
+  # regardless of which model is active (useful when switching between models with
+  # very different context windows). Clamped to the model's context length at
+  # apply-time, so a cap above the window is a no-op (ratio-based threshold wins).
+  # Survives model switches and fallback activations.
+  # threshold_tokens: 200000
+
   # Existing Codex gpt-5.5 behavior: raise Hermes' compaction trigger to 85%
   # for the ChatGPT Codex OAuth route. Set false to opt back down to threshold.
   codex_gpt55_autoraise: true

contributors/emails/maly.dan@gmail.com

@@ -0,0 +1 @@
+DanielMaly

gateway/run.py

@@ -17644,6 +17644,7 @@ async def _run_process_watcher(self, watcher: dict) -> None:
         ("compression", "enabled"),
         ("compression", "threshold"),
         ("compression", "model_thresholds"),
+        ("compression", "threshold_tokens"),
         ("compression", "codex_gpt55_autoraise"),
         ("compression", "codex_app_server_auto"),
         ("compression", "target_ratio"),

hermes_cli/config.py

@@ -1469,6 +1469,10 @@ def _ensure_hermes_home_managed(home: Path):
                                       # floored at 0.75 (raise-only) so compaction
                                       # doesn't fire with half the window still free;
                                       # set this above 0.75 to override the floor.
+        "threshold_tokens": None,     # absolute token cap — when set, compression
+                                      # triggers at the lower of the ratio-based
+                                      # threshold and this token count. Clamped to
+                                      # the model's context length at apply-time.
         "target_ratio": 0.20,         # fraction of threshold to preserve as recent tail
         "protect_last_n": 20,         # minimum recent messages to keep uncompressed
         "max_attempts": 3,            # compression retry rounds before a turn gives up
@@ -8548,6 +8552,14 @@ def show_config():
     print(f"  Enabled:      {'yes' if enabled else 'no'}")
     if enabled:
         print(f"  Threshold:    {compression.get('threshold', 0.50) * 100:.0f}%")
+        _tt = compression.get('threshold_tokens')
+        if _tt is not None:
+            try:
+                _tt = int(_tt)
+                if _tt > 0:
+                    print(f"  Token cap:    {_tt:,} tokens (takes lower of ratio vs absolute)")
+            except (TypeError, ValueError):
+                pass
         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")
         print(f"  Protect first: {compression.get('protect_first_n', 3)} non-system head messages")