fix(compression): recover merged-handoff prior-tail content through the decay scan

Composing #57835's multi-fossil summary scan with #47274's merged-handoff
unwrap: when the restart-decay path pulls a merged handoff into the
compression window, its genuine prior-tail user content must enter the
summarizer input (folded into the fresh summary) rather than being
dropped with the summary row. Standalone handoffs still drop. The
continuity test now pins the composed contract: recovered verbatim OR
via summarizer input, never silently deleted, never duplicated.

agent/context_compressor.py

@@ -4193,16 +4193,39 @@ This compaction should PRIORITISE preserving all information related to the focu
                     summary_body and _NO_USER_TASK_SENTINEL in summary_body
                 )
             summary_indices = {idx for idx, _ in summary_hits}
+            # Summary rows are excluded from the summarizer input (their
+            # bodies already ride _previous_summary), BUT a merged handoff
+            # carries genuine prior-tail user content before the delimiter —
+            # unwrap it (via the strip helper) into the window instead of
+            # dropping it (#47274 interplay with the multi-fossil scan).
+            def _window_row(idx: int, msg: Dict[str, Any]):
+                if idx not in summary_indices:
+                    return msg
+                stripped = self._strip_context_summary_handoff_message(
+                    _fresh_compaction_message_copy(msg)
+                )
+                return stripped  # None for standalone handoffs → dropped
             pre_summary_turns = [
-                msg for idx, msg in enumerate(
+                row for idx, msg in enumerate(
                     messages[compress_start:summary_idx],
                     start=compress_start,
                 )
-                if idx not in summary_indices
+                if (row := _window_row(idx, msg)) is not None
             ]
             turns_to_summarize = (
                 pre_summary_turns + messages[summary_idx + 1:compress_end]
             )
+            # The newest hit itself may be a merged handoff too — recover its
+            # prior-tail content the same way.
+            _newest_stripped = self._strip_context_summary_handoff_message(
+                _fresh_compaction_message_copy(messages[summary_idx])
+            )
+            if _newest_stripped is not None:
+                turns_to_summarize = (
+                    pre_summary_turns
+                    + [_newest_stripped]
+                    + messages[summary_idx + 1:compress_end]
+                )
             if summary_idx >= compress_end:
                 tail_start = summary_idx + 1
         elif self._previous_summary:

tests/agent/test_context_compressor_summary_continuity.py

@@ -203,25 +203,42 @@ def test_legacy_string_merged_handoff_preserves_real_tail_text():
 
 
 def test_recompression_of_current_merged_handoff_preserves_prior_tail_once():
-    """Current merged handoffs lose only stale summary data on recompression."""
+    """Current merged handoffs lose only stale summary data on recompression.
+
+    Composed contract after #57835 (restart head-protection decay): the
+    merged handoff's genuine prior-tail content must be RECOVERED — either
+    verbatim in the output (pre-decay head protection) or by entering the
+    summarizer input so the fresh summary folds it in (post-decay). It must
+    never be silently deleted, and the stale summary body must never be
+    re-emitted verbatim.
+    """
     compressor = _compressor()
     old_summary = "CURRENT-MERGED-OLD-SUMMARY unique continuity facts"
     prior_tail = "PRESERVED-PRIOR-TAIL real user content"
 
+    seen_turns = []
+
+    def _capture(turns, **kwargs):
+        seen_turns.extend(turns)
+        return ContextCompressor._with_summary_prefix(
+            "fresh replacement summary"
+        )
+
     with patch.object(
         compressor,
         "_generate_summary",
-        return_value=ContextCompressor._with_summary_prefix(
-            "fresh replacement summary"
-        ),
+        side_effect=_capture,
     ):
         result = compressor.compress(
             _messages_with_merged_handoff(old_summary, prior_tail)
         )
 
     joined = "\n".join(str(message.get("content", "")) for message in result)
-    assert prior_tail in joined
-    assert joined.count(prior_tail) == 1
+    summarizer_input = "\n".join(str(t.get("content", "")) for t in seen_turns)
+    # Prior tail recovered: verbatim in output OR folded via summarizer input.
+    assert prior_tail in joined or prior_tail in summarizer_input
+    # Never duplicated in the output.
+    assert joined.count(prior_tail) <= 1
     assert old_summary not in joined
     assert joined.count(SUMMARY_PREFIX) == 1
     assert "fresh replacement summary" in joined