fix(compression): harden pre-compress provider checkpoints

agent/agent_init.py

@@ -1229,6 +1229,9 @@ def init_agent(
     compression_abort_on_summary_failure = str(
         _compression_cfg.get("abort_on_summary_failure", False)
     ).lower() in {"true", "1", "yes"}
+    compression_memory_checkpoint_required = str(
+        _compression_cfg.get("require_memory_checkpoint", False)
+    ).lower() in {"true", "1", "yes"}
 
     # Read optional explicit context_length override for the auxiliary
     # compression model. Custom endpoints often cannot report this via
@@ -1448,6 +1451,9 @@ def init_agent(
             abort_on_summary_failure=compression_abort_on_summary_failure,
         )
     agent.compression_enabled = compression_enabled
+    agent.compression_memory_checkpoint_required = (
+        compression_memory_checkpoint_required
+    )
 
     # Reject models whose context window is below the minimum required
     # for reliable tool-calling workflows (64K tokens).

agent/context_engine.py

@@ -84,6 +84,8 @@ def compress(
         messages: List[Dict[str, Any]],
         current_tokens: int = None,
         focus_topic: str = None,
+        force: bool = False,
+        pre_compress_context: str = "",
     ) -> List[Dict[str, Any]]:
         """Compact the message list and return the new message list.
 
@@ -98,6 +100,11 @@ def compress(
                 Engines that support guided compression should prioritise
                 preserving information related to this topic.  Engines that
                 don't support it may simply ignore this argument.
+            force: Whether to bypass an engine's compression-failure cooldown
+                for an explicit manual retry.
+            pre_compress_context: Optional text returned by active memory
+                providers immediately before compaction. Preserve it in the
+                durable handoff instead of relying on summarization.
         """
 
     # -- Optional: pre-flight check ----------------------------------------
@@ -125,6 +132,23 @@ def has_content_to_compress(self, messages: List[Dict[str, Any]]) -> bool:
         """
         return True
 
+    def messages_to_compress(
+        self,
+        messages: List[Dict[str, Any]],
+    ) -> List[Dict[str, Any]]:
+        """Return messages this engine expects to summarize or discard.
+
+        Memory providers receive this preview in ``on_pre_compress`` so they
+        checkpoint the drop window rather than the protected head and tail.
+        Engines that cannot expose a precise window retain the historical
+        full-history behavior by inheriting this default.
+
+        Implementations must not mutate ``messages`` or persistent engine
+        state. Returned messages are defensively copied before providers can
+        inspect them.
+        """
+        return messages
+
     # -- Optional: session lifecycle ---------------------------------------
 
     def on_session_start(self, session_id: str, **kwargs) -> None:

agent/conversation_compression.py

@@ -28,6 +28,8 @@
 
 from __future__ import annotations
 
+import copy
+import inspect
 import logging
 import os
 import tempfile
@@ -37,9 +39,92 @@
 from typing import Any, List, Optional, Tuple
 
 from agent.model_metadata import estimate_request_tokens_rough
+from agent.redact import redact_sensitive_text
 
 logger = logging.getLogger(__name__)
 
+_PRE_COMPRESS_CONTEXT_MAX_CHARS = 16_000
+
+
+def _accepts_keyword(
+    callable_obj: Any,
+    keyword: str,
+    *,
+    allow_var_keyword: bool = True,
+) -> bool:
+    """Return whether *callable_obj* accepts *keyword* without invoking it."""
+    try:
+        parameters = inspect.signature(callable_obj).parameters.values()
+    except (TypeError, ValueError):
+        return False
+    return any(
+        parameter.name == keyword
+        and parameter.kind
+        in {
+            inspect.Parameter.POSITIONAL_OR_KEYWORD,
+            inspect.Parameter.KEYWORD_ONLY,
+        }
+        for parameter in parameters
+    ) or (
+        allow_var_keyword
+        and any(
+            parameter.kind == inspect.Parameter.VAR_KEYWORD
+            for parameter in parameters
+        )
+    )
+
+
+def _collect_pre_compress_context(
+    agent: Any,
+    messages: list,
+) -> Tuple[str, Optional[str]]:
+    """Collect provider context and enforce the optional checkpoint policy."""
+    required = bool(
+        getattr(agent, "compression_memory_checkpoint_required", False)
+    )
+    manager = getattr(agent, "_memory_manager", None)
+    if manager is None:
+        if required:
+            return "", "required memory checkpoint has no active provider"
+        return "", None
+
+    try:
+        result = manager.on_pre_compress(messages)
+    except Exception as exc:
+        if required:
+            return "", (
+                "required memory checkpoint failed "
+                f"({type(exc).__name__})"
+            )
+        logger.debug(
+            "memory manager on_pre_compress failed (%s)",
+            type(exc).__name__,
+        )
+        return "", None
+
+    if not isinstance(result, str):
+        if required:
+            return "", "required memory checkpoint returned no text reference"
+        logger.debug(
+            "memory manager on_pre_compress returned non-text context: %s",
+            type(result).__name__,
+        )
+        return "", None
+
+    context = result.strip()
+    if len(context) > _PRE_COMPRESS_CONTEXT_MAX_CHARS:
+        return "", (
+            "memory checkpoint context exceeds the "
+            f"{_PRE_COMPRESS_CONTEXT_MAX_CHARS:,}-character safety limit"
+        )
+
+    from agent.context_compressor import sanitize_pre_compress_context
+
+    context = sanitize_pre_compress_context(context)
+    if required and not context:
+        return "", "required memory checkpoint returned an empty reference"
+    return context, None
+
 
 def check_compression_model_feasibility(agent: Any) -> None:
     """Warn at session start if the auxiliary compression model's context
@@ -281,13 +366,71 @@ def compress_context(
         prompt — the session is NOT rotated.  Callers should detect the
         no-op via ``len(returned) == len(input)`` and stop the retry loop.
     """
-    # Lazy feasibility check — run the auxiliary-provider probe + context
-    # length lookup just-in-time on the first compression attempt instead of
-    # at AIAgent.__init__. Saves ~400ms cold off every short session that
-    # never reaches the threshold (the vast majority of ``chat -q`` runs).
-    # The check itself sets ``agent._compression_warning`` so the
-    # status-callback replay machinery still emits the warning to the user
-    # the first time it would matter.
+    compress_callable = agent.context_compressor.compress
+
+    def _abort_for_checkpoint(reason: str) -> Tuple[list, str]:
+        """Return unchanged context before any compaction side effect."""
+        safe_reason = redact_sensitive_text(reason, force=True)
+        logger.warning("compression aborted before message drop: %s", safe_reason)
+        if getattr(agent, "_last_compression_checkpoint_warning", None) != safe_reason:
+            agent._last_compression_checkpoint_warning = safe_reason
+            try:
+                agent._emit_warning(
+                    f"⚠ Compression aborted: {safe_reason}. No messages were dropped. "
+                    "Resolve the memory-provider/context-engine checkpoint path, "
+                    "then retry."
+                )
+            except Exception:
+                pass
+        existing_system_prompt = getattr(agent, "_cached_system_prompt", None)
+        if not existing_system_prompt:
+            existing_system_prompt = agent._build_system_prompt(system_message)
+        return messages, existing_system_prompt
+
+    checkpoint_messages = messages
+    preview_callable = getattr(
+        agent.context_compressor,
+        "messages_to_compress",
+        None,
+    )
+    if callable(preview_callable):
+        try:
+            preview = preview_callable(copy.deepcopy(messages))
+            if not isinstance(preview, list):
+                raise TypeError("compression preview must return a list")
+            checkpoint_messages = preview
+        except Exception as exc:
+            if getattr(agent, "compression_memory_checkpoint_required", False):
+                return _abort_for_checkpoint(
+                    "required memory checkpoint could not determine the "
+                    f"compression window ({type(exc).__name__})"
+                )
+            logger.debug(
+                "context engine compression-window preview failed (%s); "
+                "falling back to full history",
+                type(exc).__name__,
+            )
+
+    pre_compress_context, checkpoint_error = _collect_pre_compress_context(
+        agent,
+        checkpoint_messages,
+    )
+    if checkpoint_error:
+        return _abort_for_checkpoint(checkpoint_error)
+
+    accepts_pre_compress_context = _accepts_keyword(
+        compress_callable,
+        "pre_compress_context",
+        allow_var_keyword=False,
+    )
+    if pre_compress_context and not accepts_pre_compress_context:
+        return _abort_for_checkpoint(
+            "configured context engine cannot preserve memory checkpoint context"
+        )
+
+    # Probe auxiliary feasibility only after the checkpoint gate.  A failed
+    # required checkpoint must not trigger provider calls or mutate the
+    # one-time feasibility state.
     if not getattr(agent, "_compression_feasibility_checked", True):
         try:
             check_compression_model_feasibility(agent)
@@ -305,39 +448,20 @@ def compress_context(
         "🗜️ Compacting context — summarizing earlier conversation so I can continue..."
     )
 
-    # Notify external memory provider before compression discards context.
-    # The provider's on_pre_compress() may return a string of insights it
-    # wants surfaced inside the compression summary; capture and forward
-    # it to the compressor (fixes #7195 — return value was silently
-    # discarded for every plugin).
-    memory_context = ""
-    if agent._memory_manager:
-        try:
-            _maybe_ctx = agent._memory_manager.on_pre_compress(messages)
-            if isinstance(_maybe_ctx, str):
-                memory_context = _maybe_ctx
-        except Exception:
-            pass
-
-    try:
-        compressed = agent.context_compressor.compress(
-            messages,
-            current_tokens=approx_tokens,
-            focus_topic=focus_topic,
-            force=force,
-            memory_context=memory_context,
-        )
-    except TypeError:
-        # Plugin context engine with strict signature that doesn't accept
-        # focus_topic / force / memory_context — fall back progressively.
-        try:
-            compressed = agent.context_compressor.compress(
-                messages,
-                current_tokens=approx_tokens,
-                memory_context=memory_context,
-            )
-        except TypeError:
-            compressed = agent.context_compressor.compress(messages, current_tokens=approx_tokens)
+    candidate_kwargs: dict[str, Any] = {
+        "current_tokens": approx_tokens,
+        "focus_topic": focus_topic,
+        "force": force,
+    }
+    compress_kwargs = {
+        key: value
+        for key, value in candidate_kwargs.items()
+        if _accepts_keyword(compress_callable, key)
+    }
+    if pre_compress_context:
+        compress_kwargs["pre_compress_context"] = pre_compress_context
+
+    compressed = compress_callable(messages, **compress_kwargs)
 
     # If compression aborted (aux LLM failed to produce a usable summary)
     # the compressor returns the input messages unchanged.  Surface the
@@ -358,6 +482,14 @@ def compress_context(
             _existing_sp = agent._build_system_prompt(system_message)
         return messages, _existing_sp
 
+    if compressed is messages or compressed == messages:
+        existing_system_prompt = getattr(agent, "_cached_system_prompt", None)
+        if not existing_system_prompt:
+            existing_system_prompt = agent._build_system_prompt(system_message)
+        return messages, existing_system_prompt
+
+    agent._last_compression_checkpoint_warning = None
+
     summary_error = getattr(agent.context_compressor, "_last_summary_error", None)
     if summary_error:
         if getattr(agent, "_last_compression_summary_warning", None) != summary_error:

agent/memory_manager.py

@@ -25,9 +25,10 @@
 
 from __future__ import annotations
 
+import copy
+import inspect
 import logging
 import re
-import inspect
 from typing import Any, Dict, List, Optional
 
 from agent.memory_provider import MemoryProvider
@@ -492,19 +493,19 @@ def on_session_switch(
     def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str:
         """Notify all providers before context compression.
 
-        Returns combined text from providers to include in the compression
-        summary prompt. Empty string if no provider contributes.
+        Returns combined text for the context engine to preserve in its
+        compressed handoff. Empty string if no provider contributes.
         """
         parts = []
         for provider in self._providers:
             try:
-                result = provider.on_pre_compress(messages)
+                result = provider.on_pre_compress(copy.deepcopy(messages))
                 if result and result.strip():
                     parts.append(result)
             except Exception as e:
                 logger.debug(
-                    "Memory provider '%s' on_pre_compress failed: %s",
-                    provider.name, e,
+                    "Memory provider '%s' on_pre_compress failed (%s)",
+                    provider.name, type(e).__name__,
                 )
         return "\n\n".join(parts)
 

agent/memory_provider.py

@@ -205,9 +205,12 @@ def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str:
         Use to extract insights from messages about to be compressed.
         messages is the list that will be summarized/discarded.
 
-        Return text to include in the compression summary prompt so the
-        compressor preserves provider-extracted insights. Return empty
-        string for no contribution (backwards-compatible default).
+        Return text for the context engine to preserve in its compressed
+        handoff. The built-in compressor appends it deterministically so
+        checkpoint references do not depend on the summarizer reproducing
+        them. Return empty string for no contribution (backwards-compatible
+        default). When ``compression.require_memory_checkpoint`` is enabled,
+        an empty return prevents compaction and leaves messages unchanged.
         """
         return ""