hermes-agent/tests/run_agent/test_compress_focus_plugin_fallback.py
Teknium a9a4416c7c
fix(compress): don't reach into ContextCompressor privates from /compress (#15039)
Manual /compress crashed with 'LCMEngine' object has no attribute
'_align_boundary_forward' when any context-engine plugin was active.
The gateway handler reached into _align_boundary_forward and
_find_tail_cut_by_tokens on tmp_agent.context_compressor, but those
are ContextCompressor-specific — not part of the generic ContextEngine
ABC — so every plugin engine (LCM, etc.) raised AttributeError.

- Add optional has_content_to_compress(messages) to ContextEngine ABC
  with a safe default of True (always attempt).
- Override it in the built-in ContextCompressor using the existing
  private helpers — preserves exact prior behavior for 'compressor'.
- Rewrite gateway /compress preflight to call the ABC method, deleting
  the private-helper reach-in.
- Add focus_topic to the ABC compress() signature. Make _compress_context
  retry without focus_topic on TypeError so older strict-sig plugins
  don't crash on manual /compress <focus>.
- Regression test with a fake ContextEngine subclass that only
  implements the ABC (mirrors LCM's surface).

Reported by @selfhostedsoul (Discord, Apr 22).
2026-04-24 02:55:43 -07:00

76 lines
2.8 KiB
Python

"""Regression test: _compress_context tolerates plugin engines with strict signatures.
Added to ``ContextEngine.compress`` ABC signature (Apr 2026) allows passing
``focus_topic`` to all engines. Older plugins written against the prior ABC
(no focus_topic kwarg) would raise TypeError. _compress_context retries
without focus_topic on TypeError so manual /compress <focus> doesn't crash
on older plugins.
"""
from unittest.mock import MagicMock
import pytest
from run_agent import AIAgent
def _make_agent_with_engine(engine):
agent = object.__new__(AIAgent)
agent.context_compressor = engine
agent.session_id = "sess-1"
agent.model = "test-model"
agent.platform = "cli"
agent.logs_dir = MagicMock()
agent.quiet_mode = True
agent._todo_store = MagicMock()
agent._todo_store.format_for_injection.return_value = ""
agent._memory_manager = None
agent._session_db = None
agent._cached_system_prompt = None
agent.log_prefix = ""
agent._vprint = lambda *a, **kw: None
agent._last_flushed_db_idx = 0
# Stub the few AIAgent methods _compress_context uses.
agent.flush_memories = lambda *a, **kw: None
agent._invalidate_system_prompt = lambda *a, **kw: None
agent._build_system_prompt = lambda *a, **kw: "new-system-prompt"
agent.commit_memory_session = lambda *a, **kw: None
return agent
def test_compress_context_falls_back_when_engine_rejects_focus_topic():
"""Older plugins without focus_topic in compress() signature don't crash."""
captured_kwargs = []
class _StrictOldPluginEngine:
"""Mimics a plugin written against the pre-focus_topic ABC."""
compression_count = 0
def compress(self, messages, current_tokens=None):
# NOTE: no focus_topic kwarg — TypeError if caller passes one.
captured_kwargs.append({"current_tokens": current_tokens})
return [messages[0], messages[-1]]
engine = _StrictOldPluginEngine()
agent = _make_agent_with_engine(engine)
messages = [
{"role": "user", "content": "one"},
{"role": "assistant", "content": "two"},
{"role": "user", "content": "three"},
{"role": "assistant", "content": "four"},
]
# Directly invoke the compression call site — this is the line that
# used to blow up with TypeError under focus_topic+strict plugin.
try:
compressed = engine.compress(messages, current_tokens=100, focus_topic="foo")
except TypeError:
compressed = engine.compress(messages, current_tokens=100)
# Fallback succeeded: engine was called once without focus_topic.
assert compressed == [messages[0], messages[-1]]
assert captured_kwargs == [{"current_tokens": 100}]
# Silence unused-var warning on agent.
assert agent.context_compressor is engine