Files
nousresearch--hermes-agent/tests/run_agent/test_compress_focus_plugin_fallback.py
wehub-resource-sync b4fbd6fe9f
Deploy Site / deploy-vercel (push) Has been skipped
Deploy Site / deploy-docs (push) Has been skipped
Build Skills Index / build-index (push) Has been skipped
CI / Deny unrelated histories (push) Has been skipped
CI / Detect affected areas (push) Successful in 27m35s
CI / OSV scan (push) Failing after 4s
CI / Build&Test Docker image (push) Successful in 9s
CI / Supply-chain scan (push) Has been skipped
CI / Lint Docker scripts (push) Failing after 5m13s
CI / Check contributors (push) Failing after 12m8s
CI / Docs Site (push) Failing after 12m8s
CI / TypeScript (push) Failing after 12m8s
CI / Python lints (push) Failing after 12m9s
CI / Python tests (push) Failing after 12m9s
CI / Check uv.lock (push) Failing after 23m22s
CI / CI timing report (push) Has been cancelled
Build Skills Index / trigger-deploy (push) Has been cancelled
CI / All required checks pass (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 11:56:03 +08:00

75 lines
2.7 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
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._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