feat(compress): N-user tail guarantee via compression.min_tail_user_messages (salvage of #22566)

agent/agent_init.py

@@ -1810,6 +1810,28 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
     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))
+    # Minimum REAL (actionable) user messages guaranteed to survive in the
+    # uncompressed tail (compression.min_tail_user_messages).  Default 1
+    # preserves current behavior exactly — the existing single-user tail
+    # anchor.  Values > 1 extend the guarantee to the last N actionable
+    # user turns.  Booleans rejected (bool subclasses int), non-int-like
+    # values fall back to 1, floor at 1.
+    _raw_min_tail_users = _compression_cfg.get("min_tail_user_messages", 1)
+    if isinstance(_raw_min_tail_users, bool):
+        compression_min_tail_users = 1
+    elif isinstance(_raw_min_tail_users, int):
+        compression_min_tail_users = _raw_min_tail_users
+    elif isinstance(_raw_min_tail_users, float):
+        compression_min_tail_users = (
+            int(_raw_min_tail_users) if _raw_min_tail_users.is_integer() else 1
+        )
+    else:
+        try:
+            compression_min_tail_users = int(str(_raw_min_tail_users).strip())
+        except (TypeError, ValueError):
+            compression_min_tail_users = 1
+    if compression_min_tail_users < 1:
+        compression_min_tail_users = 1
     # Cap on compression retry rounds before a turn gives up with "max
     # compression attempts reached" (compression.max_attempts).  Hardcoding 3
     # strands sessions that legitimately need more rounds — e.g. a restart
@@ -2348,6 +2370,7 @@ def _parse_prune_int(raw, default):
             proactive_prune_tokens=compression_proactive_prune_tokens,
             proactive_prune_min_result_chars=compression_proactive_prune_min_chars,
             proactive_prune_min_reclaim_tokens=compression_proactive_prune_min_reclaim,
+            min_tail_user_messages=compression_min_tail_users,
         )
     _bind_session_state = getattr(agent.context_compressor, "bind_session_state", None)
     if callable(_bind_session_state):

agent/context_compressor.py

@@ -1646,6 +1646,7 @@ def __init__(
         proactive_prune_tokens: int = 0,
         proactive_prune_min_result_chars: int = 8000,
         proactive_prune_min_reclaim_tokens: int = 4096,
+        min_tail_user_messages: int = 1,
     ):
         self.model = model
         self.base_url = base_url
@@ -1700,6 +1701,7 @@ def __init__(
         self.proactive_prune_min_reclaim_tokens = max(
             0, int(proactive_prune_min_reclaim_tokens or 0)
         )
+        self.min_tail_user_messages = min_tail_user_messages
         self.summary_target_ratio = max(0.10, min(summary_target_ratio, 0.80))
         self.quiet_mode = quiet_mode
         # Output-token reservation: the provider carves max_tokens out of the
@@ -4263,6 +4265,69 @@ def _ensure_last_user_message_in_tail(
             return max(pair_end, head_end + 1)
         return adjusted
 
+    def _ensure_last_n_user_messages_in_tail(
+        self,
+        messages: List[Dict[str, Any]],
+        cut_idx: int,
+        head_end: int,
+        n: int,
+    ) -> int:
+        """Guarantee the last N actionable user messages are in the protected tail.
+
+        Generalizes ``_ensure_last_user_message_in_tail`` to preserve an
+        arbitrary number of recent user messages.  This prevents the token-
+        budget-based tail cut from consuming recent conversation turns
+        when large tool outputs fill the budget.
+
+        When *n* <= 1, delegates directly to the existing single-message
+        method for byte-identical regression safety.
+
+        If the conversation has fewer than *n* user messages, the earliest
+        available user message is used without error.
+
+        Only REAL actionable user turns count toward N — the collector uses
+        the same ``_is_actionable_user_turn`` /
+        ``_is_synthetic_compression_user_turn`` pair as
+        ``_find_last_user_message_idx``, so blank platform echoes, compaction
+        handoffs, continuation markers, and todo-snapshot rows never consume
+        a slot (#69291 bug class).
+
+        A user message is already a clean boundary — there is no
+        tool_call/result group that spans across it, so
+        ``_align_boundary_backward`` is intentionally NOT called.
+        Calling it can pull the cut past the user message into the
+        preceding assistant(tool_calls)→tool group and split it (#22566).
+        """
+        if n <= 1:
+            return self._ensure_last_user_message_in_tail(messages, cut_idx, head_end)
+
+        # Collect real user message indices walking backward from end.
+        # Mirror _find_last_user_message_idx's filters: compaction handoffs,
+        # blank platform echoes, and synthetic continuation/todo rows are
+        # continuity artifacts, not real user turns.
+        user_indices = []
+        for i in range(len(messages) - 1, head_end - 1, -1):
+            msg = messages[i]
+            if (
+                self._is_actionable_user_turn(msg)
+                and not self._is_synthetic_compression_user_turn(msg)
+            ):
+                user_indices.append(i)
+
+        if len(user_indices) == 0:
+            return cut_idx
+
+        if len(user_indices) < n:
+            target_idx = user_indices[-1]
+        else:
+            target_idx = user_indices[n - 1]
+
+        if target_idx >= cut_idx:
+            return cut_idx
+
+        cut_idx = target_idx
+        return max(cut_idx, head_end + 1)
+
     def _find_turn_pair_end(
         self,
         messages: List[Dict[str, Any]],
@@ -4392,6 +4457,26 @@ def _find_tail_cut_by_tokens(
         # monotonic — the tail can only grow, never shrink.
         cut_idx = self._ensure_last_assistant_message_in_tail(messages, cut_idx, head_end)
 
+        # Extend to the last N actionable user messages when configured
+        # (compression.min_tail_user_messages > 1).  This prevents the
+        # token-budget tail from consuming recent turns when large tool
+        # outputs fill the budget.  The anchor only walks ``cut_idx``
+        # backward (monotonic — the tail can only grow, never shrink), and
+        # a user message is a clean boundary, so the forward re-alignment
+        # below remains a no-op for the anchored index.  Gated at the call
+        # site so the default (1) path is byte-identical to the historical
+        # single-anchor pipeline — the single-user anchor already ran above,
+        # and re-invoking it here could re-trigger the causal-coupling
+        # forward push (#22523) after the assistant anchor adjusted the cut.
+        # getattr-guarded: bare ``ContextCompressor.__new__`` test doubles
+        # (and plugin engines) skip __init__, so the attribute may be absent
+        # (see the compression-path test-double pitfall).
+        _min_tail_users = getattr(self, "min_tail_user_messages", 1)
+        if isinstance(_min_tail_users, int) and not isinstance(_min_tail_users, bool) and _min_tail_users > 1:
+            cut_idx = self._ensure_last_n_user_messages_in_tail(
+                messages, cut_idx, head_end, _min_tail_users,
+            )
+
         # The floor guarantees forward progress — compression must always claim
         # at least one message or the caller's compress_start >= compress_end
         # guard turns the pass into a no-op that re-runs forever (the same loop

cli-config.yaml.example

@@ -448,6 +448,15 @@ compression:
   # compression of older turns.
   protect_last_n: 20
 
+  # Minimum number of REAL (actionable) user messages guaranteed to survive in
+  # the uncompressed tail (default: 1 = the existing single last-user anchor,
+  # behavior-preserving). Raise to e.g. 3 to keep the last 3 real user turns
+  # verbatim even when bulky tool outputs fill the tail token budget — blank
+  # platform echoes, compaction handoffs, and synthetic continuation rows never
+  # count toward N. The tail can exceed the token budget when this pulls the
+  # cut back; the guarantee wins over the budget by design.
+  min_tail_user_messages: 1
+
   # Compression retry rounds before a turn gives up with "max compression
   # attempts reached" (default: 3, same as the previous hardcoded value).
   # Raise (e.g. 6) for tool-schema-heavy sessions where 3 rounds cannot bring

cli.py

@@ -468,6 +468,7 @@ def load_cli_config() -> Dict[str, Any]:
         "compression": {
             "enabled": True,      # Auto-compress when approaching context limit
             "threshold": 0.50,    # Compress at 50% of model's context limit
+            "min_tail_user_messages": 1,  # Real user messages guaranteed in the tail (1 = existing single anchor)
         },
         "agent": {
             "max_turns": 90,  # Default max tool-calling iterations (shared with subagents)

contributors/emails/jerry@hermes.local

@@ -0,0 +1 @@
+zhangyang-crazy-one

gateway/run.py

@@ -18125,6 +18125,7 @@ async def _run_process_watcher(self, watcher: dict) -> None:
         ("compression", "proactive_prune_tokens"),
         ("compression", "proactive_prune_min_result_chars"),
         ("compression", "proactive_prune_min_reclaim_tokens"),
+        ("compression", "min_tail_user_messages"),
         ("agent", "disabled_toolsets"),
         ("memory", "provider"),
         ("checkpoints", "enabled"),

hermes_cli/config.py

@@ -1409,6 +1409,12 @@ def _ensure_hermes_home_managed(home: Path):
                                       # 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
+        "min_tail_user_messages": 1,  # REAL (actionable) user messages guaranteed to
+                                      # survive in the uncompressed tail. 1 = existing
+                                      # single last-user anchor (default, behavior-
+                                      # preserving); raise to e.g. 3 to keep the last
+                                      # 3 real user turns verbatim when bulky tool
+                                      # outputs fill the tail token budget.
         "max_attempts": 3,            # compression retry rounds before a turn gives up
                                       # with "max compression attempts reached". Raise
                                       # (e.g. 6) for tool-schema-heavy sessions where 3