feat(memory): opt-in fail-closed pre-compress checkpoint contract (API v1)

Context compression is intentionally lossy. Deployments that archive transcript evidence to an external durable store before compaction had no way to guarantee the archive actually happened: MemoryManager.on_pre_compress swallows provider failures by design, so a failed archive silently degraded into data loss. This adds an opt-in, provider-agnostic checkpoint contract: - memory_provider: PRE_COMPRESS_CHECKPOINT_API_VERSION = 1; providers opt in by advertising pre_compress_checkpoint_api_version. Version 0 keeps the historical best-effort hook semantics. - memory_manager: supports_pre_compress_checkpoint() capability probe; on_pre_compress(require_checkpoint=True) propagates checkpoint-provider failures and raises when no capable provider completed the checkpoint. - conversation_compression: new compression.checkpoint_required config key (default false, documented in cli-config.yaml.example). When enabled, compaction fails closed with BLOCKED_MISSING_PREREQUISITE (the uncompressed transcript is preserved) unless a checkpoint-capable provider confirms the durable checkpoint. Providers receive normalized direct user/assistant evidence: tool rows, system messages, tool-call wrappers, and prior compaction summaries are filtered host-side into one stable contract. codex_app_server compaction is rejected under the gate because it exposes no truthful pre-compaction transcript boundary. - hermes_state: persistent _compressed_summary column (declarative schema migration via _reconcile_columns) so summary provenance survives process restarts; only the resume model history carries the marker, keeping get_messages_as_conversation on its existing contract. - gateway: the lossy hygiene/auto-compact paths load the memory provider (skip_memory=False) so a required checkpoint also guards those rewrites. The gate arms only on an explicit boolean True (bare-MagicMock agents in existing tests have truthy auto-attributes). Default behavior is unchanged: checkpoint_required=false preserves best-effort semantics for all existing providers. Contract tests, including a restart round-trip of the summary marker, in tests/agent/test_pre_compress_checkpoint_contract.py. Refs #93986
2026-08-24 16:55:11 +02:00
parent 76e306c458
commit 1104ffe0b9
12 changed files with 387 additions and 15 deletions
@@ -2245,6 +2245,9 @@ def init_agent(
                compression_threshold_tokens = None
        except (TypeError, ValueError):
            compression_threshold_tokens = None
+    compression_checkpoint_required = is_truthy_value(
+        _compression_cfg.get("checkpoint_required"), default=False
+    )
    # 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
@@ -2777,6 +2780,7 @@ def init_agent(
        _cc._micro_compact_defrag_threshold_tokens = (
            compression_micro_compact_defrag_tokens
        )
+    agent.compression_checkpoint_required = compression_checkpoint_required
    agent.codex_app_server_auto_compaction = codex_app_server_auto_compaction
    agent.codex_responses_native_compaction = codex_responses_native_compaction
    agent.codex_responses_compact_threshold = codex_responses_compact_threshold
@@ -71,6 +71,7 @@ from agent.context_engine import (
    automatic_compaction_status_message,
    sanitize_memory_context,
 )
+from agent.memory_provider import PRE_COMPRESS_CHECKPOINT_API_VERSION
 from agent.model_metadata import (
    estimate_messages_tokens_rough,
    estimate_request_tokens_rough,
@@ -1111,6 +1112,16 @@ def run_compress_context_with_progress_timeout(
            # the host unwinds, so the detached worker can never publish.
            fence.revoke_commit_admission()

+class CompressionCheckpointUnavailable(RuntimeError):
+    """Raised when required durable pre-compress checkpointing is unavailable."""
+
+
+def _checkpoint_blocked(reason: str) -> CompressionCheckpointUnavailable:
+    return CompressionCheckpointUnavailable(
+        "BLOCKED_MISSING_PREREQUISITE: required pre-compress checkpoint "
+        f"unavailable: {reason}"
+    )
+

 def _lock_api_is_absent_on_session_db(lock_db: Any) -> bool:
    """Whether the live in-memory SessionDB class structurally predates locks.
@@ -1540,6 +1551,33 @@ class _CompressionActivityHeartbeat:
                return
            self._touch("context compression in progress")

+def _direct_messages_for_pre_compress_memory(messages: Any) -> list[dict[str, Any]]:
+    """Return direct user/assistant evidence safe for memory checkpointing.
+
+    Compression summaries are derivative context, not new source evidence. Tool
+    rows, system messages, and assistant tool-call wrappers are likewise omitted
+    so memory providers receive one normalized host contract instead of having
+    to infer Hermes transcript internals independently.
+    """
+    # Deferred import: context_compressor imports turn_context, which imports
+    # this module — a module-level import here would close that cycle
+    # (upstream introduced the turn_context edge with the api_content work).
+    from agent.context_compressor import COMPRESSED_SUMMARY_METADATA_KEY
+
+    direct_messages: list[dict[str, Any]] = []
+    for message in messages or []:
+        if not isinstance(message, dict):
+            continue
+        role = message.get("role")
+        if role not in {"user", "assistant"}:
+            continue
+        if message.get(COMPRESSED_SUMMARY_METADATA_KEY):
+            continue
+        if role == "assistant" and message.get("tool_calls"):
+            continue
+        direct_messages.append(message)
+    return direct_messages
+

 class _CompressionLockLeaseRefresher:
    def __init__(
@@ -2340,7 +2378,18 @@ def compress_context(
    # The memory-provider context handoff below is intentionally Hermes-only:
    # the app server does not expose its native summary prompt, so there is no
    # truthful injection point for ``on_pre_compress()`` return text here.
+    # `is True` (not bool()): unit tests drive this path with bare MagicMock
+    # agents whose auto-created attributes are truthy; the gate must only arm
+    # on the explicit boolean set by agent_init from config.
+    checkpoint_required = (
+        getattr(agent, "compression_checkpoint_required", False) is True
+    )
    if getattr(agent, "api_mode", None) == "codex_app_server":
+        if checkpoint_required:
+            raise _checkpoint_blocked(
+                "codex_app_server owns the authoritative thread and does not "
+                "expose a truthful pre-compaction transcript boundary"
+            )
        _codex_fence_entered = False
        if commit_fence is not None:
            _codex_fence_entered = commit_fence.begin_commit(
@@ -2956,9 +3005,47 @@ def compress_context(
        # wants surfaced inside the compression summary; capture and forward it
        # instead of silently discarding the provider's return value.
        memory_context = ""
-        if agent._memory_manager:
+        memory_manager = getattr(agent, "_memory_manager", None)
+        direct_messages = _direct_messages_for_pre_compress_memory(messages)
+        if checkpoint_required:
+            supports_checkpoint = getattr(
+                memory_manager, "supports_pre_compress_checkpoint", None
+            )
+            if memory_manager is None or not callable(supports_checkpoint):
+                raise _checkpoint_blocked(
+                    f"no active provider implements checkpoint API "
+                    f"v{PRE_COMPRESS_CHECKPOINT_API_VERSION}"
+                )
            try:
-                _maybe_ctx = agent._memory_manager.on_pre_compress(messages)
+                compatible = bool(
+                    supports_checkpoint(PRE_COMPRESS_CHECKPOINT_API_VERSION)
+                )
+            except Exception as exc:
+                raise _checkpoint_blocked("provider capability probe failed") from exc
+            if not compatible:
+                raise _checkpoint_blocked(
+                    f"active provider does not implement checkpoint API "
+                    f"v{PRE_COMPRESS_CHECKPOINT_API_VERSION}"
+                )
+            try:
+                _maybe_ctx = memory_manager.on_pre_compress(
+                    direct_messages,
+                    require_checkpoint=True,
+                    checkpoint_api_version=PRE_COMPRESS_CHECKPOINT_API_VERSION,
+                )
+            except Exception as exc:
+                logger.warning(
+                    "Required pre-compress checkpoint failed (%s)",
+                    type(exc).__name__,
+                )
+                raise _checkpoint_blocked(
+                    f"provider checkpoint API v{PRE_COMPRESS_CHECKPOINT_API_VERSION} failed"
+                ) from exc
+            if isinstance(_maybe_ctx, str):
+                memory_context = sanitize_memory_context(_maybe_ctx)
+        elif memory_manager:
+            try:
+                _maybe_ctx = memory_manager.on_pre_compress(direct_messages)
                if isinstance(_maybe_ctx, str):
                    memory_context = sanitize_memory_context(_maybe_ctx)
            except Exception:
@@ -33,7 +33,7 @@ import threading
 from concurrent.futures import Future, ThreadPoolExecutor, wait
 from typing import Any, Callable, Dict, List, Optional

-from agent.memory_provider import MemoryProvider
+from agent.memory_provider import MemoryProvider, PRE_COMPRESS_CHECKPOINT_API_VERSION
 from agent.skill_commands import extract_user_instruction_from_skill_message
 from tools.registry import tool_error

@@ -1056,14 +1056,47 @@ class MemoryManager:
                    provider.name, e,
                )

-    def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str:
+    def supports_pre_compress_checkpoint(
+        self,
+        api_version: int = PRE_COMPRESS_CHECKPOINT_API_VERSION,
+    ) -> bool:
+        """Return whether an active provider guarantees checkpoint API support."""
+        for provider in self._providers:
+            try:
+                provider_version = int(
+                    getattr(provider, "pre_compress_checkpoint_api_version", 0)
+                )
+            except (TypeError, ValueError):
+                continue
+            if provider_version >= api_version:
+                return True
+        return False
+
+    def on_pre_compress(
+        self,
+        messages: List[Dict[str, Any]],
+        *,
+        require_checkpoint: bool = False,
+        checkpoint_api_version: int = PRE_COMPRESS_CHECKPOINT_API_VERSION,
+    ) -> 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.
+        summary prompt. Empty string if no provider contributes. When
+        ``require_checkpoint`` is true, at least one provider advertising the
+        requested checkpoint API must return successfully; its exception is
+        propagated so the caller can preserve the uncompressed transcript.
        """
        parts = []
+        checkpoint_succeeded = False
        for provider in self._providers:
+            try:
+                provider_version = int(
+                    getattr(provider, "pre_compress_checkpoint_api_version", 0)
+                )
+            except (TypeError, ValueError):
+                provider_version = 0
+            is_checkpoint_provider = provider_version >= checkpoint_api_version
            try:
                result = provider.on_pre_compress(messages)
                if result and result.strip():
@@ -1073,6 +1106,16 @@ class MemoryManager:
                    "Memory provider '%s' on_pre_compress failed: %s",
                    provider.name, e,
                )
+                if require_checkpoint and is_checkpoint_provider:
+                    raise
+            else:
+                if is_checkpoint_provider:
+                    checkpoint_succeeded = True
+        if require_checkpoint and not checkpoint_succeeded:
+            raise RuntimeError(
+                "No active memory provider completed pre-compress checkpoint "
+                f"API v{checkpoint_api_version}"
+            )
        return "\n\n".join(parts)

    @staticmethod
@@ -41,6 +41,8 @@ from typing import Any, Dict, List, Optional

 logger = logging.getLogger(__name__)

+PRE_COMPRESS_CHECKPOINT_API_VERSION = 1
+
 # Default glyph for the deterministic memory indicators. Providers override
 # per-status with their own brand mark (e.g. Hindsight uses "👁️").
 INDICATOR_GLYPH = "🧠"
@@ -104,6 +106,11 @@ def is_trivial_prompt(text: Optional[str]) -> bool:
 class MemoryProvider(ABC):
    """Abstract base class for memory providers."""

+    # Providers that durably checkpoint every successful on_pre_compress()
+    # call may opt into this host contract by setting the current version.
+    # Version 0 preserves the historical best-effort hook semantics.
+    pre_compress_checkpoint_api_version = 0
+
    @property
    @abstractmethod
    def name(self) -> str:
@@ -542,6 +542,16 @@ compression:
  # Set to false if you prefer to manage context manually or want errors on overflow
  enabled: true

+  # Fail closed before lossy compaction unless an active memory provider that
+  # implements the pre-compress checkpoint contract (API v1) confirms its
+  # durable checkpoint (default: false). With this on and no checkpoint, the
+  # compaction attempt errors with BLOCKED_MISSING_PREREQUISITE and the
+  # uncompressed transcript is preserved for a later retry. Only enable it
+  # with a checkpoint-capable provider configured — see
+  # website/docs/developer-guide/memory-provider-plugin.md
+  # ("Pre-Compress Checkpoints").
+  checkpoint_required: false
+
  # Opt-in compression progress notices on chat platforms (default: false).
  # By design, routine automatic compression is SILENT on human-facing chat
  # gateways (Telegram, Discord, Slack, ...) — it happens in the background
@@ -558,7 +558,9 @@ def _seed_hygiene_system_prompt(
 ) -> bool:
    """Keep gateway hygiene from rebuilding a live session's system prompt.

-    The hygiene helper intentionally skips memory-provider initialization.
+    The hygiene helper loads the memory provider for the pre-compress hook,
+    but it still runs outside the live session's fully initialized prompt
+    environment (hygiene-only platform marker, no platform context files).
    Compression is allowed to persist a system prompt, so letting that helper
    rebuild one would strip external provider blocks from the live session.
    Seed the exact persisted prompt instead.  When no usable prompt can be
@@ -19723,7 +19725,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                                    model=_hyg_model,
                                    max_iterations=4,
                                    quiet_mode=True,
-                                    skip_memory=True,
+                                    # Hygiene performs the same lossy rewrite
+                                    # as normal compression, so it loads the
+                                    # memory provider unconditionally (normal
+                                    # compression never skips it either):
+                                    # best-effort on_pre_compress runs for
+                                    # hygiene too, and when
+                                    # compression.checkpoint_required is
+                                    # enabled the required checkpoint is
+                                    # created before any transcript mutation.
+                                    skip_memory=False,
                                    enabled_toolsets=["memory"],
                                    session_id=session_entry.session_id,
                                    session_db=_hyg_session_db,
@@ -4392,8 +4392,10 @@ class GatewaySlashCommandsMixin:
                runtime_kwargs["platform"] = platform_key
            runtime_kwargs["gateway_session_key"] = session_key

-            # The manual compression helper skips memory-provider initialization,
-            # but _compress_context may persist its cached system prompt. Restore
+            # The manual compression helper loads the memory provider (see
+            # skip_memory=False below) but still runs outside the live
+            # session's fully initialized prompt environment, and
+            # _compress_context may persist its cached system prompt. Restore
            # the exact live-session prompt so provider blocks are retained.
            session_row = None
            get_session = getattr(self._session_db, "get_session", None)
@@ -4415,7 +4417,13 @@ class GatewaySlashCommandsMixin:
                model=model,
                max_iterations=4,
                quiet_mode=True,
-                skip_memory=True,
+                # This agent performs the same lossy rewrite as normal
+                # compression, so it loads the memory provider unconditionally
+                # (normal compression never skips it either): best-effort
+                # on_pre_compress runs for manual compression too, and when
+                # compression.checkpoint_required is enabled the required
+                # pre-compression checkpoint can be created.
+                skip_memory=False,
                enabled_toolsets=["memory"],
                session_id=session_entry.session_id,
                session_db=getattr(self._session_db, "_db", self._session_db),
@@ -743,6 +743,9 @@ DEFAULT_CONFIG = {

    "compression": {
        "enabled": True,
+        "checkpoint_required": False, # Fail closed before lossy compaction unless an
+                                      # active memory provider confirms checkpoint API
+                                      # compatibility and completes the checkpoint.
        "progress_notices": False,    # opt-in (#52995): when True, routine compression
                                      # progress statuses (compacting/preflight/pre-API/
                                      # idle/retry) are delivered to chat gateway
@@ -10513,6 +10513,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
        platform_message_id: str = None,
        observed: bool = False,
        effect_disposition: Optional[str] = None,
+        _compressed_summary: bool = False,
        timestamp: Any = None,
        api_content: Optional[str] = None,
        display_kind: Optional[str] = None,
@@ -10588,8 +10589,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                """INSERT INTO messages (session_id, role, content, tool_call_id,
                   tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason,
                   reasoning, reasoning_content, reasoning_details, codex_reasoning_items,
-                   codex_message_items, platform_message_id, observed, active, api_content, display_kind, display_metadata)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                   codex_message_items, platform_message_id, observed, _compressed_summary, active, api_content, display_kind, display_metadata)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    session_id,
                    role,
@@ -10608,6 +10609,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                    codex_message_items_json,
                    platform_message_id,
                    1 if observed else 0,
+                    1 if _compressed_summary else 0,
                    1,
                    _scrub_surrogates(api_content) if isinstance(api_content, str) else None,
                    _scrub_surrogates(display_kind) if isinstance(display_kind, str) else None,
@@ -11019,8 +11021,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                """INSERT INTO messages (session_id, role, content, tool_call_id,
                   tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason,
                   reasoning, reasoning_content, reasoning_details, codex_reasoning_items,
-                   codex_message_items, platform_message_id, observed, active, api_content, display_kind, display_metadata)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                   codex_message_items, platform_message_id, observed, _compressed_summary, active, api_content, display_kind, display_metadata)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    session_id,
                    role,
@@ -11039,6 +11041,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                    codex_message_items_json,
                    platform_msg_id,
                    1 if msg.get("observed") else 0,
+                    1 if msg.get("_compressed_summary") else 0,
                    1,
                    _scrub_surrogates(api_content) if isinstance(api_content, str) else None,
                    _scrub_surrogates(msg.get("display_kind")) if isinstance(msg.get("display_kind"), str) else None,
@@ -11515,6 +11518,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
        result = []
        for row in rows:
            msg = dict(row)
+            if msg.pop("_compressed_summary", 0):
+                msg["_compressed_summary"] = True
            if "content" in msg:
                msg["content"] = self._decode_content(msg["content"])
            if msg.get("tool_calls"):
@@ -11789,7 +11794,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
    _CONVERSATION_ROW_COLUMNS = (
        "id, role, content, tool_call_id, tool_calls, tool_name, effect_disposition, "
        "finish_reason, reasoning, reasoning_content, reasoning_details, "
-        "codex_reasoning_items, codex_message_items, platform_message_id, observed, timestamp, "
+        "codex_reasoning_items, codex_message_items, platform_message_id, observed, "
+        "_compressed_summary, timestamp, "
        "api_content, display_kind, display_metadata"
    )

@@ -11801,6 +11807,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
        include_ancestors: bool,
        repair_alternation: bool,
        include_row_ids: bool = False,
+        include_summary_markers: bool = False,
    ) -> List[Dict[str, Any]]:
        """Decode fetched message rows into the OpenAI conversation format.

@@ -11854,6 +11861,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
                decoded = self._decode_display_metadata(row["display_metadata"])
                if decoded is not None:
                    msg["display_metadata"] = decoded
+            if include_summary_markers and row["_compressed_summary"]:
+                msg["_compressed_summary"] = True
            if row["timestamp"]:
                msg["timestamp"] = row["timestamp"]
            if row["tool_call_id"]:
@@ -12028,6 +12037,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
            include_ancestors=False,
            repair_alternation=True,
            include_row_ids=True,
+            # Pre-compress checkpointing: the resumed model history must keep
+            # the summary marker so checkpoint providers can exclude derivative
+            # summaries after a process restart (marker survives restart).
+            include_summary_markers=True,
        )
        display_history = self._rows_to_conversation(
            rows,
@@ -446,6 +446,7 @@ CREATE TABLE IF NOT EXISTS messages (
    codex_message_items TEXT,
    platform_message_id TEXT,
    observed INTEGER DEFAULT 0,
+    _compressed_summary INTEGER NOT NULL DEFAULT 0,
    active INTEGER NOT NULL DEFAULT 1,
    compacted INTEGER NOT NULL DEFAULT 0,
    api_content TEXT,
@@ -2368,6 +2368,7 @@ class AIAgent:
                    "reasoning_details": msg.get("reasoning_details"),
                    "codex_reasoning_items": msg.get("codex_reasoning_items"),
                    "codex_message_items": msg.get("codex_message_items"),
+                    "_compressed_summary": bool(msg.get(COMPRESSED_SUMMARY_METADATA_KEY)),
                    "timestamp": _row_timestamp,
                    "api_content": _row_api_content,
                    # Standalone reference handoffs are always hidden, even
@@ -0,0 +1,184 @@
+"""Host-side contract tests for the opt-in pre-compress checkpoint API (v1).
+
+The contract has three parts:
+- providers opt in by advertising ``pre_compress_checkpoint_api_version``;
+- ``MemoryManager`` exposes capability probing and a ``require_checkpoint``
+  mode whose failure must propagate instead of being swallowed;
+- the compression host normalizes messages to direct user/assistant evidence
+  before handing them to providers.
+"""
+
+import pytest
+
+from agent.conversation_compression import (
+    CompressionCheckpointUnavailable,
+    _checkpoint_blocked,
+    _direct_messages_for_pre_compress_memory,
+)
+from agent.context_compressor import COMPRESSED_SUMMARY_METADATA_KEY
+from agent.memory_manager import MemoryManager
+from agent.memory_provider import (
+    PRE_COMPRESS_CHECKPOINT_API_VERSION,
+    MemoryProvider,
+)
+
+
+class _BaseStubProvider(MemoryProvider):
+    def __init__(self, name="stub"):
+        self._name = name
+        self.pre_compress_calls = []
+
+    @property
+    def name(self):
+        return self._name
+
+    def is_available(self):
+        return True
+
+    def initialize(self, session_id, **kwargs):
+        return None
+
+    def get_tool_schemas(self):
+        return []
+
+    def on_pre_compress(self, messages):
+        self.pre_compress_calls.append(messages)
+        return f"{self._name} context"
+
+
+class _CheckpointProvider(_BaseStubProvider):
+    pre_compress_checkpoint_api_version = PRE_COMPRESS_CHECKPOINT_API_VERSION
+
+
+class _FailingCheckpointProvider(_CheckpointProvider):
+    def on_pre_compress(self, messages):
+        raise RuntimeError("durable store unreachable")
+
+
+class _FailingLegacyProvider(_BaseStubProvider):
+    def on_pre_compress(self, messages):
+        raise RuntimeError("legacy best-effort failure")
+
+
+def test_provider_base_class_defaults_to_best_effort_api_version_zero():
+    assert MemoryProvider.pre_compress_checkpoint_api_version == 0
+
+
+def test_direct_messages_filter_keeps_only_direct_source_evidence():
+    messages = [
+        {"role": "system", "content": "system prompt"},
+        {"role": "user", "content": "durable user decision"},
+        {"role": "assistant", "content": "direct assistant answer"},
+        {"role": "assistant", "content": "call", "tool_calls": [{"id": "t1"}]},
+        {"role": "tool", "content": "tool output", "tool_call_id": "t1"},
+        {
+            "role": "assistant",
+            "content": "previous compaction summary",
+            COMPRESSED_SUMMARY_METADATA_KEY: True,
+        },
+        "not-a-dict",
+    ]
+
+    direct = _direct_messages_for_pre_compress_memory(messages)
+
+    assert [m["content"] for m in direct] == [
+        "durable user decision",
+        "direct assistant answer",
+    ]
+
+
+def test_manager_advertises_checkpoint_capability_only_with_capable_provider():
+    # The host allows one external provider per manager, so capability is
+    # probed on two separate managers.
+    legacy_manager = MemoryManager()
+    legacy_manager.add_provider(_BaseStubProvider("legacy"))
+    assert legacy_manager.supports_pre_compress_checkpoint(
+        PRE_COMPRESS_CHECKPOINT_API_VERSION
+    ) is False
+
+    durable_manager = MemoryManager()
+    durable_manager.add_provider(_CheckpointProvider("durable"))
+    assert durable_manager.supports_pre_compress_checkpoint(
+        PRE_COMPRESS_CHECKPOINT_API_VERSION
+    ) is True
+
+
+def test_manager_require_checkpoint_raises_without_capable_provider():
+    manager = MemoryManager()
+    manager.add_provider(_BaseStubProvider("legacy"))
+
+    with pytest.raises(RuntimeError, match="pre-compress checkpoint"):
+        manager.on_pre_compress(
+            [{"role": "user", "content": "evidence"}],
+            require_checkpoint=True,
+            checkpoint_api_version=PRE_COMPRESS_CHECKPOINT_API_VERSION,
+        )
+
+
+def test_manager_require_checkpoint_propagates_checkpoint_provider_failure():
+    manager = MemoryManager()
+    manager.add_provider(_FailingCheckpointProvider("durable"))
+
+    with pytest.raises(RuntimeError, match="durable store unreachable"):
+        manager.on_pre_compress(
+            [{"role": "user", "content": "evidence"}],
+            require_checkpoint=True,
+            checkpoint_api_version=PRE_COMPRESS_CHECKPOINT_API_VERSION,
+        )
+
+
+def test_manager_require_checkpoint_succeeds_and_returns_provider_context():
+    manager = MemoryManager()
+    durable = _CheckpointProvider("durable")
+    manager.add_provider(durable)
+
+    combined = manager.on_pre_compress(
+        [{"role": "user", "content": "evidence"}],
+        require_checkpoint=True,
+        checkpoint_api_version=PRE_COMPRESS_CHECKPOINT_API_VERSION,
+    )
+
+    assert "durable context" in combined
+    assert durable.pre_compress_calls
+
+
+def test_manager_best_effort_mode_keeps_historical_swallow_semantics():
+    manager = MemoryManager()
+    manager.add_provider(_FailingLegacyProvider("legacy"))
+
+    combined = manager.on_pre_compress([{"role": "user", "content": "evidence"}])
+
+    assert combined == ""
+
+
+def test_checkpoint_blocked_error_is_prefixed_and_typed():
+    error = _checkpoint_blocked("no active provider")
+    assert isinstance(error, CompressionCheckpointUnavailable)
+    assert str(error).startswith("BLOCKED_MISSING_PREREQUISITE:")
+    assert "no active provider" in str(error)
+
+
+def test_compressed_summary_marker_survives_restart_via_resume_history(tmp_path):
+    """The persistent marker reaches the resumed model history — and only it.
+
+    ``get_messages_as_conversation`` keeps its existing marker-free contract;
+    the resume path carries ``_compressed_summary`` so checkpoint providers
+    keep excluding derivative summaries after a process restart.
+    """
+    from hermes_state import SessionDB
+
+    db = SessionDB(tmp_path / "state.db")
+    db.create_session("s1", source="cli")
+    db.append_message("s1", "user", "durable user evidence")
+    db.append_message(
+        "s1", "assistant", "derivative summary", _compressed_summary=True
+    )
+
+    reopened = SessionDB(tmp_path / "state.db")
+    model_history, _display = reopened.get_resume_conversations("s1")
+    by_content = {m.get("content"): m for m in model_history}
+    assert by_content["derivative summary"].get("_compressed_summary") is True
+    assert "_compressed_summary" not in by_content["durable user evidence"]
+
+    plain = reopened.get_messages_as_conversation("s1")
+    assert all("_compressed_summary" not in m for m in plain)