fix(compression): a handoff row no longer collapses the tail cut onto itself

The tail-cut walk charged in-transcript handoff rows as protected-tail
candidates, but the handoff scan folds every row up to the newest handoff
into the summary window: none of them can stay in the tail. When the
handoff row alone pushed the transcript over the soft ceiling, the walk
stopped on it and the planned window held only the handoff, so compress()
returned the empty_post_handoff_window structural no-op at every trigger.

Live on main (gemini-2.5-flash, context_length 64000, prepare_ahead off):
a Desktop manual compress, then one automatic compaction (handoff now at
index 0), then four consecutive empty_post_handoff_window no-ops while the
prompt grew 47,085 -> 57,143 -> 66,716 tokens, past the 64k window.

The tail-cut walk now prices rows at or before the newest handoff at zero,
so the cut is decided by the rows that can really stay in the tail and the
window covers the old handoff plus the turns after it. Same scenario on
this branch: the second automatic compaction summarises 17 rows and the
next prompt is 18,156 tokens.

agent/context_compressor.py

@@ -36,7 +36,7 @@ from agent.context_compressor_prune import ProactivePruneMixin
 from agent.context_compressor_summary import SummaryDispatchMixin
 from agent.context_compressor_retry import retryable_user_text
 from agent.context_compressor_tail import (
-    demote_stale_tail_tools, lean_tail_budget, is_demoted_stub, skill_view_names_by_call_id,
+    demote_stale_tail_tools, lean_tail_budget, is_demoted_stub, skill_view_names_by_call_id, tail_cut_pricer,
     elide_keeping_skill_safety_rule, with_skill_safety_rule, without_skill_safety_rule,
 )
 from agent.context_compressor_telemetry import CompressionTelemetryMixin
@@ -3119,14 +3119,14 @@ class ContextCompressor(
 
     def _walk_tail_budget(
         self, messages: list[dict[str, Any]], head_end: int, ceiling: int, min_tail: int, *, cut_at_break: bool,
+        price: Optional[Callable[[int], int]] = None,
     ) -> tuple[int, int]:
         """Accumulate message tokens newest-first until ``ceiling`` (once ``min_tail`` rows are kept).
         Returns ``(cut_idx, accumulated)``; ``cut_idx`` is the first protected index. On the budget
         break the cut stays at the last accepted row, or moves onto the breaking row when
         ``cut_at_break``. Only the newest assistant turn's thinking is charged (#73624) unless the route
-        echoes stale thinking every turn — must agree with the preflight estimate (#84371)."""
-        n = len(messages)
-        price = self._tail_row_pricer(messages)
+        echoes stale thinking every turn — must agree with the preflight estimate (#84371); ``price`` overrides it."""
+        n, price = len(messages), price or self._tail_row_pricer(messages)
         accumulated = 0
         cut = n  # start from beyond the end
         for i in range(n - 1, head_end - 1, -1):
@@ -4999,11 +4999,11 @@ Write only the summary body. Do not include any preamble or prefix."""
         # token-respecting floor exists then. Required user/assistant anchors and atomic tool groups
         # are applied below and may still necessarily exceed the ceiling.
         walk_floor = 0 if soft_ceiling >= min_tail * _estimate_msg_budget_tokens({}) else min_tail
-        cut_idx, accumulated = self._walk_tail_budget(messages, head_end, soft_ceiling, walk_floor, cut_at_break=False)
-        # Whole transcript fits soft_ceiling: re-cut with the raw budget so a worthwhile middle
-        # exists (else #40803 loop).
+        price = tail_cut_pricer(self, messages, head_end)
+        cut_idx, accumulated = self._walk_tail_budget(messages, head_end, soft_ceiling, walk_floor, cut_at_break=False, price=price)
+        # Whole transcript fits soft_ceiling: re-cut with the raw budget so a worthwhile middle exists (else #40803 loop).
         if cut_idx <= head_end and 0 < accumulated <= soft_ceiling:
-            cut_idx, _ = self._walk_tail_budget(messages, head_end, token_budget, min_tail, cut_at_break=True)
+            cut_idx, _ = self._walk_tail_budget(messages, head_end, token_budget, min_tail, cut_at_break=True, price=price)
 
         fallback_cut = n - min_tail
         # The newest row never leaves the tail, whatever its role. When it alone exceeds the ceiling the walk accepts

agent/context_compressor_tail.py

@@ -118,6 +118,20 @@ SKILL_SAFETY_RULE = (
 )
 
 
+def tail_cut_pricer(compressor: Any, messages: list[dict[str, Any]], head_end: int):
+    """Tail-cut walk price ``index -> tokens`` in which rows at or before the newest handoff cost nothing.
+
+    The handoff scan folds every row up to the newest in-transcript handoff into the summary window (the tail starts
+    past it), so none of them can stay in the protected tail. Charging them let the handoff row alone break the soft
+    ceiling: the walk stopped on it, the window held only that handoff, and compress() returned the
+    ``empty_post_handoff_window`` no-op at every trigger while the session grew past its window."""
+    price = compressor._tail_row_pricer(messages)
+    newest = next(
+        (i for i in range(len(messages) - 1, head_end - 1, -1) if compressor._is_context_summary_message(messages[i])), -1,
+    )
+    return price if newest < 0 else (lambda i: 0 if i <= newest else price(i))
+
+
 def with_skill_safety_rule(
     summary: str, head: list[dict[str, Any]], tail: list[dict[str, Any]],
 ) -> tuple[str, list[dict[str, Any]], list[dict[str, Any]]]:

tests/agent/test_compressor_handoff_tail_price.py

@@ -0,0 +1,76 @@
+"""A handoff row never stays in the protected tail, so it must not decide where the tail cut lands.
+
+The handoff scan folds every row up to the newest in-transcript handoff into the summary window. When the
+tail walk charged that handoff, a summary row that alone pushed the transcript over the soft ceiling ended
+the walk on itself: the planned window held only the handoff and compress() returned the
+``empty_post_handoff_window`` no-op at every trigger (live: four no-ops in a row, prompt 47k -> 66.7k on a
+64k window, after a manual compress followed by one automatic compaction).
+"""
+
+from unittest.mock import patch
+
+from agent.context_compressor import SUMMARY_PREFIX, ContextCompressor, _estimate_msg_budget_tokens
+
+
+def _compressor() -> ContextCompressor:
+    with patch("agent.context_compressor.get_model_context_length", return_value=64_000):
+        return ContextCompressor(
+            model="test-model", quiet_mode=True, config_context_length=64_000, protect_first_n=3,
+            protect_last_n=20, summary_target_ratio=0.20, threshold_percent=0.50, tail_mode="lean",
+        )
+
+
+def _handoff(body_words: int = 1_800) -> dict:
+    return {"role": "user", "content": f"{SUMMARY_PREFIX}\n## Goal\n" + " ".join(["earlier"] * body_words),
+            "_compressed_summary": True}
+
+
+def _turns(n: int, words: int = 330) -> list[dict]:
+    rows = []
+    for i in range(n):
+        rows.append({"role": "user", "content": f"question {i} " + " ".join(["ask"] * 40)})
+        rows.append({"role": "assistant", "content": f"answer {i} " + " ".join(["reply"] * words)})
+    return rows
+
+
+def _over_ceiling_only_with_handoff(cc: ContextCompressor, head: list[dict]) -> list[dict]:
+    """Rows after the handoff fit the soft ceiling; adding the handoff breaks it."""
+    soft = cc._tail_soft_ceiling(cc.tail_token_budget)
+    handoff = _handoff()
+    for n in range(4, 80):
+        turns = _turns(n)
+        after = sum(map(_estimate_msg_budget_tokens, turns))
+        if after <= soft < after + _estimate_msg_budget_tokens(handoff):
+            return [*head, handoff, *turns]
+    raise AssertionError("fixture could not straddle the soft ceiling")
+
+
+def test_leading_handoff_does_not_collapse_the_window_to_itself():
+    cc = _compressor()
+    messages = _over_ceiling_only_with_handoff(cc, head=[])
+    with patch.object(cc, "_generate_summary", return_value="NEW SUMMARY BODY"):
+        result = cc.compress(messages, current_tokens=60_000)
+
+    telemetry = cc._last_compression_telemetry or {}
+    assert telemetry.get("failure_class") is None
+    assert cc._last_compression_made_progress is True
+    assert len(result) < len(messages)
+    assert sum(1 for m in result if cc._is_context_summary_message(m)) == 1
+    assert result[-1]["content"] == messages[-1]["content"]
+
+
+def test_handoff_behind_a_protected_head_folds_the_turns_after_it():
+    """Gateway manual compress keeps protect_first_n=3: the handoff sits at index 3 of the next transcript."""
+    cc = _compressor()
+    head = [
+        {"role": "user", "content": "first request"},
+        {"role": "assistant", "content": "first answer"},
+        {"role": "user", "content": "second request"},
+    ]
+    messages = _over_ceiling_only_with_handoff(cc, head=head)
+    start, end = cc._compress_window(messages)
+    handoff_idx = len(head)
+    assert start == 0
+    # The window reaches past the old handoff into the turns after it, not just the rows in front of it.
+    assert end > handoff_idx + 1
+    assert cc.has_content_to_compress(messages) is True