diff --git a/integrations/hermes/README.md b/integrations/hermes/README.md index 974f78a06..657ec5f23 100644 --- a/integrations/hermes/README.md +++ b/integrations/hermes/README.md @@ -154,7 +154,11 @@ profiles, users, repositories, or directories. - `on_session_end()` waits for queued writes and calls `/v1/memory/flush`. - `on_pre_compress()` optionally persists only filtered new user/assistant turns and flushes them before Hermes discards old messages. It is disabled by - default and uses stable source IDs for overlapping compression windows. + default and uses stable source IDs for overlapping compression windows. The + provider advertises the pre-compress checkpoint API v2 contract: it captures + the host-normalized evidence list when Hermes supplies one, and a checkpoint + that cannot be committed raises, so `compression.checkpoint_required` keeps the + uncompressed transcript instead of discarding it behind a failed capture. - `on_memory_write()` mirrors built-in Hermes memory additions as explicit entries and retires the mapped PowerContext entry for replacements/removals. - Agent tools expose the complete PowerContext operation groups: Memory diff --git a/integrations/hermes/plugins/powercontext/README.md b/integrations/hermes/plugins/powercontext/README.md index b85440e5f..1164614a1 100644 --- a/integrations/hermes/plugins/powercontext/README.md +++ b/integrations/hermes/plugins/powercontext/README.md @@ -49,6 +49,14 @@ provider configuration, or set assistant turns are captured; system/tool messages are excluded and detected secrets are redacted before sending them to PowerContext. +The provider advertises Hermes' pre-compress checkpoint API v2. When Hermes +supplies its host-normalized evidence list, that list is captured instead of the +raw transcript, so turns Hermes already replaced with a compression summary are +not stored as fresh evidence. With `compression.checkpoint_required: true`, +enable `capture_pre_compress` as well: if PowerContext cannot commit the +checkpoint, the provider raises and Hermes keeps the uncompressed transcript +instead of discarding it behind a failed capture. + Evaluation tracing is also opt-in. Set `evaluation_trace: true` or `POWERCONTEXT_HERMES_EVALUATION_TRACE=1` to record context injections in per-session JSONL files under `$HERMES_HOME/powercontext/evaluation-trace/`. diff --git a/integrations/hermes/plugins/powercontext/provider.py b/integrations/hermes/plugins/powercontext/provider.py index 855cb3f12..8172664e7 100644 --- a/integrations/hermes/plugins/powercontext/provider.py +++ b/integrations/hermes/plugins/powercontext/provider.py @@ -78,9 +78,6 @@ from .helpers import ( message_text as _message_text, ) -from .helpers import ( - messages_to_text as _messages_to_text, -) from .helpers import ( new_precompress_entries as _new_precompress_entries, ) @@ -124,6 +121,47 @@ } +def _precompress_content_for_entries( + entries: list[tuple[str, dict[str, Any]]], + *, + limit: int, +) -> tuple[str, list[str], bool]: + lines: list[str] = [] + fingerprints: list[str] = [] + total = 0 + for fingerprint, message in entries: + role = str(message.get("role", "unknown")) + text = _message_text(message.get("content")) + if not text: + continue + line = f"[{role}] {text}" + projected = total + (1 if lines else 0) + len(line) + if projected > limit: + break + lines.append(line) + fingerprints.append(fingerprint) + total = projected + return "\n".join(lines).strip(), fingerprints, len(fingerprints) == len(entries) + + +def _precompress_snapshot_after_capture( + entries: list[tuple[str, dict[str, Any]]], + new_entries: list[tuple[str, dict[str, Any]]], + captured_fingerprints: list[str], + *, + complete_checkpoint: bool, +) -> list[str]: + current_fingerprints = [fingerprint for fingerprint, _message in entries] + if complete_checkpoint: + return current_fingerprints + + captured_count = len(captured_fingerprints) + for start in range(len(entries) - len(new_entries) + 1): + if entries[start : start + len(new_entries)] == new_entries: + return current_fingerprints[: start + captured_count] + return captured_fingerprints + + def _merge_config(existing: dict[str, Any], values: dict[str, Any]) -> dict[str, Any]: merged = {**existing, **values} if "base_url" in values and "allow_insecure_http" not in values and "allow_insecure_http" in existing: @@ -162,9 +200,21 @@ def __init__(self) -> None: super().__init__("PowerContext returned an invalid Scope binding") +class PreCompressCheckpointError(PowerContextError): + """Raised when a required pre-compress checkpoint cannot be committed.""" + + def __init__(self, reason: str) -> None: + super().__init__(f"PowerContext did not commit the required pre-compress checkpoint: {reason}") + + class PowerContextMemoryProvider(MemoryProvider): """Hermes provider backed by a running PowerContext server.""" + # Hermes' PRE_COMPRESS_CHECKPOINT_API_VERSION. Declaring v2 promises that a normal + # on_pre_compress() return means the captured transcript is stored, and that a + # checkpoint this provider cannot commit raises instead of reporting success. + pre_compress_checkpoint_api_version = 2 + _tool_names: ClassVar[set[str]] = { "powercontext_search_memory", "powercontext_get_memory", @@ -929,42 +979,73 @@ def on_session_switch( self._precompress_stream_id = new_session_id self._precompress_snapshot = [] - def on_pre_compress(self, messages: list[dict[str, Any]]) -> str: + def on_pre_compress( + self, + messages: list[dict[str, Any]], + *, + evidence_messages: list[dict[str, Any]] | None = None, + require_checkpoint: bool = False, + ) -> str: + """Persist new user/assistant turns before Hermes discards the transcript. + + ``evidence_messages`` is the host-normalized transcript that Hermes hands only to + checkpoint API v2 providers. It is preferred over ``messages`` because the host also + removes earlier compression summaries, which this provider cannot recognize on its own. + With ``require_checkpoint`` set, a checkpoint that cannot be committed raises so the + caller keeps the uncompressed transcript. + """ scope_id = self._scope_id client = self._client - if ( - not client - or not scope_id - or not messages - or not _as_bool( - _config_value( - self._config, - "capture_pre_compress", - "POWERCONTEXT_HERMES_CAPTURE_PRE_COMPRESS", - False, - ), + if not _as_bool( + _config_value( + self._config, + "capture_pre_compress", + "POWERCONTEXT_HERMES_CAPTURE_PRE_COMPRESS", False, - ) + ), + False, ): + self._fail_required_checkpoint( + require_checkpoint, "capture_pre_compress is disabled, so no transcript was stored" + ) + return "" + if not client or not scope_id: + self._fail_required_checkpoint( + require_checkpoint, "the provider has no client or active Scope to store the transcript in" + ) return "" - entries = _precompress_entries(messages) + evidence = evidence_messages if evidence_messages is not None else messages + entries = _precompress_entries(evidence) new_entries = _new_precompress_entries(self._precompress_snapshot, entries) if not new_entries: self._precompress_snapshot = [fingerprint for fingerprint, _message in entries] return "" - content = _messages_to_text([message for _fingerprint, message in new_entries], limit=_MAX_PRECOMPRESS_CHARS) + content, captured_fingerprints, complete_checkpoint = _precompress_content_for_entries( + new_entries, + limit=_MAX_PRECOMPRESS_CHARS, + ) + if not complete_checkpoint: + self._fail_required_checkpoint( + require_checkpoint, + "the transcript exceeds the maximum checkpoint payload size", + ) + if not captured_fingerprints: + return "" if not content: return "" self._wait_for_background() if scope_id != self._scope_id: + self._fail_required_checkpoint( + require_checkpoint, "the active Scope changed while the transcript was being captured" + ) return "" anchor = self._precompress_snapshot[-1] if self._precompress_snapshot else "" idempotency_payload = { "stream": self._precompress_stream_id, "anchor": anchor, - "entries": [fingerprint for fingerprint, _message in new_entries], + "entries": captured_fingerprints, } source_id = ( "hermes-compression:" @@ -978,16 +1059,34 @@ def on_pre_compress(self, messages: list[dict[str, Any]]) -> str: metadata={ "kind": "hermes-context-compression", "session_id": self._session_id, - "message_count": len(new_entries), + "message_count": len(captured_fingerprints), }, ) - self._flush_memory_if_supported(scope_id=scope_id) except PowerContextError as error: self._emit_failure_diagnostic("pre_compression_capture", error) + self._fail_required_checkpoint(require_checkpoint, f"storing the transcript failed: {error}") return "" - self._precompress_snapshot = [fingerprint for fingerprint, _message in entries] + if scope_id != self._scope_id: + self._fail_required_checkpoint( + require_checkpoint, "the active Scope changed while the transcript was being captured" + ) + return "" + # The transcript is durable once capture_content returns, so a later memory + # extraction failure must not turn a committed checkpoint into a failed one. + self._flush_memory_if_supported(scope_id=scope_id) + self._precompress_snapshot = _precompress_snapshot_after_capture( + entries, + new_entries, + captured_fingerprints, + complete_checkpoint=complete_checkpoint, + ) return "" + @staticmethod + def _fail_required_checkpoint(required: bool, reason: str) -> None: + if required: + raise PreCompressCheckpointError(reason) + def on_memory_write( self, action: str, diff --git a/tests/integrations/test_hermes_provider.py b/tests/integrations/test_hermes_provider.py index 67c6b58af..942971a06 100644 --- a/tests/integrations/test_hermes_provider.py +++ b/tests/integrations/test_hermes_provider.py @@ -179,6 +179,11 @@ def provider_and_client(tmp_path, hermes_modules): provider.shutdown() +def hermes_provider_module(provider): + """The loaded plugin module, so tests can reach its public error types.""" + return sys.modules[type(provider).__module__] + + def test_prefetch_uses_profile_and_user_scoped_context(provider_and_client): provider, client = provider_and_client @@ -637,6 +642,154 @@ def test_pre_compress_captures_only_new_overlapping_windows(provider_and_client) assert len({call[1][1] for call in capture_calls}) == 3 +def test_provider_advertises_the_fail_closed_checkpoint_contract(provider_and_client): + provider, _client = provider_and_client + + assert provider.pre_compress_checkpoint_api_version == 2 + + +def test_pre_compress_prefers_host_normalized_evidence(provider_and_client): + provider, client = provider_and_client + provider._config["capture_pre_compress"] = True + + provider.on_pre_compress( + [{"role": "user", "content": "raw transcript turn"}], + evidence_messages=[{"role": "user", "content": "normalized evidence turn"}], + ) + + content = client.calls[0][1][2] + assert "normalized evidence turn" in content + assert "raw transcript turn" not in content + + +def test_required_checkpoint_raises_when_capture_is_disabled(provider_and_client): + provider, client = provider_and_client + + with pytest.raises(hermes_provider_module(provider).PreCompressCheckpointError): + provider.on_pre_compress( + [{"role": "user", "content": "Not captured by default."}], + require_checkpoint=True, + ) + + assert client.calls == [] + + +def test_required_checkpoint_raises_when_the_transcript_cannot_be_stored(provider_and_client, monkeypatch): + provider, client = provider_and_client + provider._config["capture_pre_compress"] = True + provider_module = hermes_provider_module(provider) + failure = provider_module.PowerContextTransportError("server unreachable") + + def capture_fails(*_args, **_kwargs): + raise failure + + monkeypatch.setattr(client, "capture_content", capture_fails) + + with pytest.raises(provider_module.PreCompressCheckpointError): + provider.on_pre_compress( + [{"role": "user", "content": "Capture me before compression."}], + require_checkpoint=True, + ) + + +def test_optional_checkpoint_still_fails_open_when_the_transcript_cannot_be_stored(provider_and_client, monkeypatch): + provider, client = provider_and_client + provider._config["capture_pre_compress"] = True + + failure = hermes_provider_module(provider).PowerContextTransportError("server unreachable") + + def capture_fails(*_args, **_kwargs): + raise failure + + monkeypatch.setattr(client, "capture_content", capture_fails) + + assert provider.on_pre_compress([{"role": "user", "content": "Capture me."}]) == "" + + +def test_required_checkpoint_succeeds_when_the_transcript_is_stored(provider_and_client): + provider, client = provider_and_client + provider._config["capture_pre_compress"] = True + + provider.on_pre_compress( + [{"role": "user", "content": "The service must stay backward compatible."}], + require_checkpoint=True, + ) + + assert [call[0] for call in client.calls] == ["capture_content", "get_capabilities", "flush_memory"] + + +def test_required_checkpoint_raises_when_the_transcript_would_be_truncated(provider_and_client): + provider, client = provider_and_client + provider._config["capture_pre_compress"] = True + provider_module = hermes_provider_module(provider) + + with pytest.raises(provider_module.PreCompressCheckpointError): + provider.on_pre_compress( + [{"role": "user", "content": "x" * 30_001}], + require_checkpoint=True, + ) + + assert client.calls == [] + assert provider._precompress_snapshot == [] + + +def test_required_checkpoint_accepts_a_window_that_is_already_captured(provider_and_client): + provider, client = provider_and_client + provider._config["capture_pre_compress"] = True + window = [ + {"role": "user", "content": "Only user turn."}, + {"role": "assistant", "content": "Only assistant turn."}, + ] + + provider.on_pre_compress(window, require_checkpoint=True) + provider.on_pre_compress(window, require_checkpoint=True) + + capture_calls = [call for call in client.calls if call[0] == "capture_content"] + assert len(capture_calls) == 1 + + +def test_required_checkpoint_raises_when_scope_changes_during_capture(provider_and_client, monkeypatch): + provider, client = provider_and_client + provider._config["capture_pre_compress"] = True + provider_module = hermes_provider_module(provider) + capture_content = client.capture_content + + def capture_switches_scope(*args, **kwargs): + capture_content(*args, **kwargs) + provider._switch_scope("scp_other_scope") + + monkeypatch.setattr(client, "capture_content", capture_switches_scope) + + with pytest.raises(provider_module.PreCompressCheckpointError): + provider.on_pre_compress( + [{"role": "user", "content": "Capture before the scope switch."}], + require_checkpoint=True, + ) + + assert [call[0] for call in client.calls] == ["capture_content"] + assert provider._precompress_snapshot == [] + + +def test_committed_checkpoint_survives_a_memory_extraction_failure(provider_and_client, monkeypatch): + provider, client = provider_and_client + provider._config["capture_pre_compress"] = True + + failure = hermes_provider_module(provider).PowerContextTransportError("flush unavailable") + + def flush_fails(*_args, **_kwargs): + raise failure + + monkeypatch.setattr(client, "flush_memory", flush_fails) + + # The transcript is already stored, so the required checkpoint is still satisfied. + provider.on_pre_compress( + [{"role": "user", "content": "Capture me before compression."}], + require_checkpoint=True, + ) + + assert [call[0] for call in client.calls] == ["capture_content", "get_capabilities"] + + def test_memory_write_retires_mapped_entries_for_replace_and_remove(provider_and_client): provider, client = provider_and_client