diff --git a/agent/agent_init.py b/agent/agent_init.py index 0e2dd48a9a..75a550d22a 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -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 diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index e5dffd458b..63be8747b5 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -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: diff --git a/agent/memory_manager.py b/agent/memory_manager.py index 5d94b6d18d..d373660ce4 100644 --- a/agent/memory_manager.py +++ b/agent/memory_manager.py @@ -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 diff --git a/agent/memory_provider.py b/agent/memory_provider.py index 317ad73d5a..5f6fdd0b9b 100644 --- a/agent/memory_provider.py +++ b/agent/memory_provider.py @@ -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: diff --git a/cli-config.yaml.example b/cli-config.yaml.example index e98391601f..e03df674c0 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -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 diff --git a/gateway/run.py b/gateway/run.py index 461d75b116..0f00f7ab36 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -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, diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 9512193253..cb644f3ed1 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -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), diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index cae665f25b..8d68834a9b 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -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 diff --git a/hermes_state.py b/hermes_state.py index 9c1adb9f9d..2850544909 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -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, diff --git a/hermes_state_common.py b/hermes_state_common.py index ff1da505f0..c224b8202c 100644 --- a/hermes_state_common.py +++ b/hermes_state_common.py @@ -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, diff --git a/run_agent.py b/run_agent.py index 6cfa4fe8fd..90e273f748 100644 --- a/run_agent.py +++ b/run_agent.py @@ -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 diff --git a/tests/agent/test_pre_compress_checkpoint_contract.py b/tests/agent/test_pre_compress_checkpoint_contract.py new file mode 100644 index 0000000000..27b6a37848 --- /dev/null +++ b/tests/agent/test_pre_compress_checkpoint_contract.py @@ -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)