From 78b39e68c809589e23dd2e4dafe7e133ed819881 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Fri, 25 Sep 2026 14:18:25 +0800 Subject: [PATCH 1/2] feat(hermes): honor the pre-compress checkpoint API v2 contract Hermes' MemoryManager already dispatches the pre-compress checkpoint API v2 contract, but the PowerContext provider still advertises v1, so it never receives the host-normalized evidence list or the require_checkpoint signal. Declare pre_compress_checkpoint_api_version = 2, accept evidence_messages and require_checkpoint, and prefer the host-normalized evidence: turns Hermes already replaced with a compression summary are no longer stored as fresh evidence. When a checkpoint is required and cannot be committed, raise PreCompressCheckpointError so Hermes keeps the uncompressed transcript. Without a required checkpoint, capture failures stay fail-open as before. A committed capture is no longer downgraded by a later memory extraction failure: capture_content is the durable checkpoint, flush_memory is not. --- integrations/hermes/README.md | 6 +- .../hermes/plugins/powercontext/README.md | 8 ++ .../hermes/plugins/powercontext/provider.py | 71 ++++++++--- tests/integrations/test_hermes_provider.py | 116 ++++++++++++++++++ 4 files changed, 185 insertions(+), 16 deletions(-) diff --git a/integrations/hermes/README.md b/integrations/hermes/README.md index 974f78a06f..657ec5f234 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 b85440e5f6..d759569610 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: a checkpoint PowerContext cannot commit +raises, and Hermes then 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 855cb3f122..b1259354cf 100644 --- a/integrations/hermes/plugins/powercontext/provider.py +++ b/integrations/hermes/plugins/powercontext/provider.py @@ -162,9 +162,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,26 +941,44 @@ 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] @@ -959,6 +989,9 @@ def on_pre_compress(self, messages: list[dict[str, Any]]) -> str: 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 = { @@ -981,13 +1014,21 @@ def on_pre_compress(self, messages: list[dict[str, Any]]) -> str: "message_count": len(new_entries), }, ) - 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 "" + # 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 = [fingerprint for fingerprint, _message in entries] 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 67c6b58af2..133d43b060 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,117 @@ 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_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_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 From da00759e5acdc8e350a9a3f0d892331219eb6897 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Sat, 26 Sep 2026 17:20:00 +0800 Subject: [PATCH 2/2] fix(hermes): fail required pre-compress checkpoints on partial capture Ensure required pre-compress checkpoints only report success after the complete bounded window is stored for the same active Scope. Keep optional captures from advancing the snapshot past persisted entries and clarify the documented fail-closed behavior. Tested: python -m pytest tests/integrations/test_hermes_provider.py -q; ruff check provider/test files; ruff format --check provider/test/README; ty check provider/test files --- .../hermes/plugins/powercontext/README.md | 6 +- .../hermes/plugins/powercontext/provider.py | 72 +++++++++++++++++-- tests/integrations/test_hermes_provider.py | 37 ++++++++++ 3 files changed, 105 insertions(+), 10 deletions(-) diff --git a/integrations/hermes/plugins/powercontext/README.md b/integrations/hermes/plugins/powercontext/README.md index d759569610..1164614a1d 100644 --- a/integrations/hermes/plugins/powercontext/README.md +++ b/integrations/hermes/plugins/powercontext/README.md @@ -53,9 +53,9 @@ 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: a checkpoint PowerContext cannot commit -raises, and Hermes then keeps the uncompressed transcript instead of discarding -it behind a failed capture. +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 diff --git a/integrations/hermes/plugins/powercontext/provider.py b/integrations/hermes/plugins/powercontext/provider.py index b1259354cf..8172664e70 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: @@ -984,7 +1022,17 @@ def on_pre_compress( 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() @@ -997,7 +1045,7 @@ def on_pre_compress( idempotency_payload = { "stream": self._precompress_stream_id, "anchor": anchor, - "entries": [fingerprint for fingerprint, _message in new_entries], + "entries": captured_fingerprints, } source_id = ( "hermes-compression:" @@ -1011,17 +1059,27 @@ def on_pre_compress( metadata={ "kind": "hermes-context-compression", "session_id": self._session_id, - "message_count": len(new_entries), + "message_count": len(captured_fingerprints), }, ) 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 "" + 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 = [fingerprint for fingerprint, _message in entries] + self._precompress_snapshot = _precompress_snapshot_after_capture( + entries, + new_entries, + captured_fingerprints, + complete_checkpoint=complete_checkpoint, + ) return "" @staticmethod diff --git a/tests/integrations/test_hermes_provider.py b/tests/integrations/test_hermes_provider.py index 133d43b060..942971a069 100644 --- a/tests/integrations/test_hermes_provider.py +++ b/tests/integrations/test_hermes_provider.py @@ -718,6 +718,21 @@ def test_required_checkpoint_succeeds_when_the_transcript_is_stored(provider_and 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 @@ -733,6 +748,28 @@ def test_required_checkpoint_accepts_a_window_that_is_already_captured(provider_ 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