mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-25 17:18:11 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6
hermes-agent/tests/agent/test_ghost_skill_pruning.py
Teknium 69365109b3 fix(compression): mark raw skill_view bodies summarized away, not only pre-pruned rows 
_collect_ghosted_skill_names() covers both ghost-skill shapes in the
compressed middle window: rows already demoted to a [SKILL_PRUNED: ...]
marker AND raw skill_view bodies (> _SKILL_VIEW_PRUNE_MIN_CHARS) that
survived Phase-1 inside an earlier protected tail and then aged into the
compression window — the summarizer paraphrases those instructions away
too. Shared threshold constant between the emit site and the scan.
Pinned by a live-probe-shaped test (real compress(), mocked aux LLM).
2026-07-23 16:58:06 -07:00

388 lines
16 KiB
Python
Raw Blame History

"""Ghost-skill defense tests (#32106, salvage of PR #44166).
When compaction reduces an old ``skill_view`` result to a metadata-only
summary, the model still believes the skill is loaded even though its
instructions are gone. The defense has three layers:
- P0/P1: the pruned tool-result summary carries a canonical
``[SKILL_PRUNED: ...]`` marker with the exact reload call, and the
system prompt (SKILLS_GUIDANCE) tells the model how to react to it.
- Phase-1 protection: a skill loaded just before compaction (or actively
referenced in the protected tail) keeps its full body through the
ordinary prune passes.
- P2: markers entering the summarizer are extracted BEFORE the aux LLM
call and deterministically re-injected if the model paraphrased them
away — including on the static fallback path.
Test patterns for the marker emit checks adapted from PR #32375
(@LeonSGP43) with credit.
"""
from unittest.mock import MagicMock, patch
from agent.context_compressor import (
SKILL_PRUNED_MARKER_PREFIX,
SUMMARY_PREFIX,
ContextCompressor,
_collect_protected_skill_names,
_extract_pruned_skill_names,
_MAX_PRUNED_SKILL_MARKERS,
_reinject_pruned_skill_markers,
_skill_pruned_marker,
_summarize_tool_result,
)
def _make_compressor(**overrides):
kwargs = dict(
model="test/model",
quiet_mode=True,
protect_first_n=1,
protect_last_n=2,
)
kwargs.update(overrides)
with patch(
"agent.context_compressor.get_model_context_length", return_value=100000
):
return ContextCompressor(**kwargs)
def _skill_view_pair(call_id, skill_name, size=6000):
return [
{
"role": "assistant",
"content": "",
"tool_calls": [{
"id": call_id,
"type": "function",
"function": {
"name": "skill_view",
"arguments": f'{{"name":"{skill_name}"}}',
},
}],
},
{
"role": "tool",
"tool_call_id": call_id,
"content": f"# {skill_name} instructions\n" + "x" * size,
},
]
class TestSkillPrunedMarkerEmit:
"""Marker emit — patterns adapted from PR #32375 (@LeonSGP43)."""
def test_skill_view_summary_marks_pruned_content(self):
summary = _summarize_tool_result(
"skill_view", '{"name":"docker-management"}', "x" * 6000
)
assert summary.startswith("[skill_view] name=docker-management (6,000 chars)")
assert _skill_pruned_marker("docker-management") in summary
assert "reload with skill_view(name='docker-management')" in summary
def test_small_skill_view_summary_not_marked(self):
summary = _summarize_tool_result(
"skill_view", '{"name":"docker-management"}', "x" * 1234
)
assert summary == "[skill_view] name=docker-management (1,234 chars)"
assert SKILL_PRUNED_MARKER_PREFIX not in summary
def test_other_skill_tool_summaries_remain_metadata_only(self):
summary = _summarize_tool_result(
"skills_list", '{"name":"docker-management"}', "x" * 6000
)
assert summary == "[skills_list] name=docker-management (6,000 chars)"
assert SKILL_PRUNED_MARKER_PREFIX not in summary
def test_marker_extractor_round_trips_the_emitted_marker(self):
"""Emit and check sides share one canonical string.
The original PR #44166 emitted ``[SKILL_PRUNED:`` but presence-
checked ``[SKILL_PRUNED]`` — re-injection fired even when the
marker had survived. Pin the round trip.
"""
summary = _summarize_tool_result("skill_view", '{"name":"pdf"}', "x" * 6000)
assert _extract_pruned_skill_names(summary) == ["pdf"]
assert _skill_pruned_marker("pdf") in summary
class TestReinjectPrunedSkillMarkers:
def test_no_duplicate_when_canonical_marker_survived(self):
marker = _skill_pruned_marker("pdf")
out = _reinject_pruned_skill_markers("summary body\n" + marker, ["pdf"])
assert out.count(SKILL_PRUNED_MARKER_PREFIX) == 1
def test_reinjects_when_marker_paraphrased_away(self):
out = _reinject_pruned_skill_markers(
"The pdf skill was loaded earlier but its content was summarized.",
["pdf"],
)
assert _skill_pruned_marker("pdf") in out
assert "## Pruned Skills" in out
def test_partial_survival_reinjects_only_missing(self):
marker_a = _skill_pruned_marker("alpha")
out = _reinject_pruned_skill_markers("body\n" + marker_a, ["alpha", "beta"])
assert out.count(_skill_pruned_marker("alpha")) == 1
assert out.count(_skill_pruned_marker("beta")) == 1
def test_empty_name_list_is_a_no_op(self):
assert _reinject_pruned_skill_markers("body", []) == "body"
def test_marker_block_is_not_classified_as_handoff_content(self):
"""Markers must never turn a row into summary/handoff content."""
marker = _skill_pruned_marker("pdf")
assert ContextCompressor.classify_summary_content(marker) is None
row = "[skill_view] name=pdf (6,000 chars) " + marker
assert ContextCompressor.classify_summary_content(row) is None
# And the strip helper leaves marker-bearing rows untouched.
c = ContextCompressor.__new__(ContextCompressor)
msg = {"role": "user", "content": row}
assert c._strip_context_summary_handoff_message(msg) == msg
def test_reinjected_summary_still_classifies_standalone(self):
body = _reinject_pruned_skill_markers("## Goal\nwork\n", ["pdf"])
full = SUMMARY_PREFIX + "\n\n" + body
assert ContextCompressor.classify_summary_content(full) == "standalone"
class TestProtectedSkillPrune:
"""Phase-1 prune must not demote a just-loaded skill (#32106)."""
def _filler(self, n, start=0):
out = []
for i in range(n):
role = "user" if (start + i) % 2 == 0 else "assistant"
out.append({"role": role, "content": f"filler {start + i} " + "y" * 400})
return out
def test_old_single_use_skill_is_pruned_with_marker(self):
c = _make_compressor()
msgs = _skill_view_pair("call_s", "old-skill") + self._filler(14)
result, pruned = c._prune_old_tool_results(msgs, protect_tail_count=4)
assert pruned >= 1
skill_row = result[1]
assert _skill_pruned_marker("old-skill") in skill_row["content"]
def test_recently_loaded_skill_survives_prune(self):
c = _make_compressor()
# skill loaded within the last 10 messages, but OUTSIDE the
# protected tail count — without the guard it would be demoted.
msgs = (
self._filler(10)
+ _skill_view_pair("call_s", "fresh-skill")
+ self._filler(6, start=10)
)
result, _ = c._prune_old_tool_results(msgs, protect_tail_count=4)
skill_row = result[11]
assert skill_row["content"].startswith("# fresh-skill instructions")
assert SKILL_PRUNED_MARKER_PREFIX not in skill_row["content"]
def test_skill_named_in_tail_user_message_survives_prune(self):
c = _make_compressor()
msgs = (
_skill_view_pair("call_s", "steered-skill")
+ self._filler(14)
+ [{"role": "user", "content": "keep following the steered-skill steps"}]
)
result, _ = c._prune_old_tool_results(msgs, protect_tail_count=4)
skill_row = result[1]
assert skill_row["content"].startswith("# steered-skill instructions")
def test_pressure_demotion_overrides_skill_protection(self):
"""Pass-4 must still demote protected skill bodies (#61932 guard)."""
c = _make_compressor()
msgs = (
self._filler(2)
+ _skill_view_pair("call_s", "fresh-skill", size=60000)
+ [{"role": "user", "content": "active ask"}]
)
# Tiny token budget → protected region exceeds the soft ceiling and
# the pressure pass must reclaim the skill body despite protection.
result, pruned = c._prune_old_tool_results(
msgs, protect_tail_count=4, protect_tail_tokens=100
)
skill_row = result[3]
assert pruned >= 1
assert _skill_pruned_marker("fresh-skill") in skill_row["content"]
class TestMarkerSurvivesRealCompress:
"""P2 layer: markers survive a real compress() with a mocked aux LLM."""
def _mock_response(self, text):
response = MagicMock()
response.choices = [MagicMock()]
response.choices[0].message.content = text
return response
def _messages_with_pruned_skill_in_middle(self):
"""Transcript whose compressed middle carries a prune marker row."""
pruned_row_content = (
"[skill_view] name=pdf (48,201 chars) " + _skill_pruned_marker("pdf")
)
msgs = [
{"role": "system", "content": "System prompt"},
{"role": "user", "content": "Build the PDF report"},
{
"role": "assistant",
"content": "",
"tool_calls": [{
"id": "call_pdf",
"type": "function",
"function": {
"name": "skill_view",
"arguments": '{"name":"pdf"}',
},
}],
},
{"role": "tool", "tool_call_id": "call_pdf", "content": pruned_row_content},
{"role": "assistant", "content": "Loaded the skill, working."},
{"role": "user", "content": "continue"},
{"role": "assistant", "content": "more work " + "z" * 500},
{"role": "user", "content": "latest ask"},
{"role": "assistant", "content": "ack"},
]
return msgs
def _summary_text_of(self, result):
for msg in result:
if ContextCompressor.classify_summary_content(msg.get("content")):
return msg["content"]
raise AssertionError(f"no summary message found in {result!r}")
def test_marker_reinjected_when_summarizer_drops_it(self):
c = _make_compressor(protect_first_n=1, protect_last_n=2)
msgs = self._messages_with_pruned_skill_in_middle()
drop_response = self._mock_response(
"## Goal\nBuild the PDF report.\n\n## Completed Actions\n"
"1. Loaded some skills and worked on the report."
)
with (
patch.object(c, "_find_tail_cut_by_tokens", return_value=7),
patch(
"agent.context_compressor.call_llm", return_value=drop_response
) as mock_call,
):
result = c.compress(msgs, force=True)
assert mock_call.called
summary_text = self._summary_text_of(result)
assert _skill_pruned_marker("pdf") in summary_text
# Stored iterative-update state carries the marker too.
assert _skill_pruned_marker("pdf") in c._previous_summary
def test_marker_not_duplicated_when_summarizer_preserves_it(self):
c = _make_compressor(protect_first_n=1, protect_last_n=2)
msgs = self._messages_with_pruned_skill_in_middle()
keep_response = self._mock_response(
"## Goal\nBuild the PDF report.\n\n## Pruned Skills\n"
+ _skill_pruned_marker("pdf")
)
with (
patch.object(c, "_find_tail_cut_by_tokens", return_value=7),
patch("agent.context_compressor.call_llm", return_value=keep_response),
):
result = c.compress(msgs, force=True)
summary_text = self._summary_text_of(result)
assert summary_text.count(_skill_pruned_marker("pdf")) == 1
def test_marker_survives_static_fallback_summary(self):
c = _make_compressor(protect_first_n=1, protect_last_n=2)
msgs = self._messages_with_pruned_skill_in_middle()
with (
patch.object(c, "_find_tail_cut_by_tokens", return_value=7),
patch(
"agent.context_compressor.call_llm",
side_effect=RuntimeError("no provider"),
),
):
result = c.compress(msgs, force=True)
summary_text = self._summary_text_of(result)
assert _skill_pruned_marker("pdf") in summary_text
assert c._last_summary_fallback_used is True
def test_raw_skill_body_in_compressed_middle_gets_marker(self):
"""A never-demoted skill_view body summarized away still ghosts.
The skill body can survive Phase-1 (protected tail of an earlier
prune) and then age into the compression window as RAW content.
The summarizer paraphrases it away — the P2 layer must emit the
marker for it as well, not only for already-pruned rows.
"""
c = _make_compressor(protect_first_n=1, protect_last_n=2)
skill_body = "# pdf skill\n" + ("Detailed instructions line.\n" * 400)
msgs = self._messages_with_pruned_skill_in_middle()
msgs[3] = {"role": "tool", "tool_call_id": "call_pdf", "content": skill_body}
drop_response = self._mock_response(
"## Goal\nBuild the PDF report.\n\n## Completed Actions\n"
"1. Loaded some skills and worked on the report."
)
with (
patch.object(c, "_find_tail_cut_by_tokens", return_value=7),
patch("agent.context_compressor.call_llm", return_value=drop_response),
):
result = c.compress(msgs, force=True)
summary_text = self._summary_text_of(result)
assert _skill_pruned_marker("pdf") in summary_text
def test_marker_survives_iterative_recompression(self):
"""Markers in a rehydrated handoff summary survive iterative rewrites.
On re-compression the previous handoff (carrying the marker) is
rehydrated into ``_previous_summary``; even when the summarizer's
iterative update drops the marker, re-injection restores it.
"""
c = _make_compressor(protect_first_n=1, protect_last_n=2)
prior_handoff = (
SUMMARY_PREFIX
+ "\n\n## Goal\nOld work.\n\n## Pruned Skills\n"
+ _skill_pruned_marker("pdf")
)
msgs = [
{"role": "system", "content": "System prompt"},
{"role": "user", "content": prior_handoff},
] + [
{"role": "assistant" if i % 2 == 0 else "user", "content": f"turn {i} " + "q" * 300}
for i in range(8)
]
drop_response = self._mock_response("## Goal\nNext task in flight.")
with (
patch.object(c, "_find_tail_cut_by_tokens", return_value=8),
patch("agent.context_compressor.call_llm", return_value=drop_response),
):
result = c.compress(msgs, force=True)
summary_text = self._summary_text_of(result)
assert _skill_pruned_marker("pdf") in summary_text
class TestReinjectionBoundsAndRedaction:
def test_marker_list_is_capped(self):
names = [f"skill-{i}" for i in range(_MAX_PRUNED_SKILL_MARKERS + 15)]
out = _reinject_pruned_skill_markers("body", names)
assert out.count(SKILL_PRUNED_MARKER_PREFIX) == len(names)
# The cap is applied at the collection sites in _generate_summary /
# _build_static_fallback_summary; the helper itself is mechanical.
def test_reinjection_block_is_redacted(self, monkeypatch):
import agent.redact as redact_mod
# force=True redaction must win even when redaction is disabled.
monkeypatch.setattr(redact_mod, "_REDACT_ENABLED", False, raising=False)
secret = "ghp_" + "a1B2" * 6
out = _reinject_pruned_skill_markers("body", [f"x {secret}"])
assert secret not in out
class TestSkillsGuidanceSafetyRule:
def test_safety_rule_present_with_real_newlines(self):
from agent.prompt_builder import SKILLS_GUIDANCE
assert "## Skill Safety Rule" in SKILLS_GUIDANCE
assert "[SKILL_PRUNED]" in SKILLS_GUIDANCE
assert "skill_view(name='...')" in SKILLS_GUIDANCE
# The rule list must use REAL newlines — the original PR hunk risked
# literal backslash-n escape text rendering into the system prompt.
assert "\\n" not in SKILLS_GUIDANCE
assert SKILLS_GUIDANCE.count("\n") >= 6
for rule in ("UNAVAILABLE", "RELOAD", "WAIT", "DEDUP"):
assert rule in SKILLS_GUIDANCE
Reference in a new issue View git blame Copy permalink