mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-08 13:12:08 +00:00
eab208db70
Port from openai/codex#21069 ("Spill large hook outputs from context"). Both shell hooks and Python plugins can return {"context": "..."} from pre_llm_call, which gets appended to the current turn's user message on every subsequent API call. A plugin that emits a large blob inflates every turn and blows out the prompt cache prefix. - tools/hook_output_spill.py: shared helper that writes oversized context to $HERMES_HOME/hook_outputs/<session_id>/<uuid>.txt and returns a head/tail preview plus the saved path. Never raises. - agent/turn_context.py: apply the cap at the pre_llm_call aggregation site (moved here from run_agent.py since the original PR), covering both Python plugins and shell hooks. - agent/shell_hooks.py: reserve output_spill as a sub-key under hooks: so the config block doesn't emit unknown-hook-event warnings. - Docs: document the cap + config in build-a-hermes-plugin.md. Config (behaviour-preserving when absent): hooks.output_spill: enabled/max_chars/preview_head/preview_tail/directory Tests: 14 unit tests; shell_hooks (56) and plugins (100) suites green. E2E validated with isolated HERMES_HOME (spill, passthrough, traversal sanitisation, reserved-key skip).
205 lines
8.2 KiB
Python
205 lines
8.2 KiB
Python
"""Tests for tools.hook_output_spill."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from tools import hook_output_spill as hos
|
|
|
|
|
|
class GetSpillConfigTests(unittest.TestCase):
|
|
def test_defaults_when_no_config(self):
|
|
with patch.object(hos, "load_config", create=True, return_value={}):
|
|
# load_config is resolved at call time via local import;
|
|
# patch the module's source instead.
|
|
pass
|
|
with patch("hermes_cli.config.load_config", return_value={}):
|
|
cfg = hos.get_spill_config()
|
|
self.assertTrue(cfg["enabled"])
|
|
self.assertEqual(cfg["max_chars"], hos.DEFAULT_MAX_CHARS)
|
|
self.assertEqual(cfg["preview_head"], hos.DEFAULT_PREVIEW_HEAD)
|
|
self.assertEqual(cfg["preview_tail"], hos.DEFAULT_PREVIEW_TAIL)
|
|
self.assertIsNone(cfg["directory"])
|
|
|
|
def test_user_overrides_are_respected(self):
|
|
user_cfg = {
|
|
"hooks": {
|
|
"output_spill": {
|
|
"enabled": False,
|
|
"max_chars": 500,
|
|
"preview_head": 25,
|
|
"preview_tail": 10,
|
|
"directory": "/tmp/spill-test",
|
|
}
|
|
}
|
|
}
|
|
with patch("hermes_cli.config.load_config", return_value=user_cfg):
|
|
cfg = hos.get_spill_config()
|
|
self.assertFalse(cfg["enabled"])
|
|
self.assertEqual(cfg["max_chars"], 500)
|
|
self.assertEqual(cfg["preview_head"], 25)
|
|
self.assertEqual(cfg["preview_tail"], 10)
|
|
self.assertEqual(cfg["directory"], "/tmp/spill-test")
|
|
|
|
def test_bad_values_fall_back_to_defaults(self):
|
|
user_cfg = {
|
|
"hooks": {
|
|
"output_spill": {
|
|
"max_chars": "not-a-number",
|
|
"preview_head": -100,
|
|
"preview_tail": None,
|
|
"directory": 123, # not a string
|
|
}
|
|
}
|
|
}
|
|
with patch("hermes_cli.config.load_config", return_value=user_cfg):
|
|
cfg = hos.get_spill_config()
|
|
self.assertEqual(cfg["max_chars"], hos.DEFAULT_MAX_CHARS)
|
|
self.assertEqual(cfg["preview_head"], hos.DEFAULT_PREVIEW_HEAD)
|
|
self.assertEqual(cfg["preview_tail"], hos.DEFAULT_PREVIEW_TAIL)
|
|
self.assertIsNone(cfg["directory"])
|
|
|
|
def test_load_config_exception_is_swallowed(self):
|
|
with patch("hermes_cli.config.load_config", side_effect=RuntimeError("bad")):
|
|
cfg = hos.get_spill_config()
|
|
self.assertEqual(cfg["max_chars"], hos.DEFAULT_MAX_CHARS)
|
|
self.assertTrue(cfg["enabled"])
|
|
|
|
|
|
class SpillIfOversizedTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmpdir = tempfile.mkdtemp(prefix="hermes-spill-test-")
|
|
|
|
def tearDown(self):
|
|
import shutil
|
|
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
|
|
|
def _cfg(self, **overrides):
|
|
base = {
|
|
"enabled": True,
|
|
"max_chars": 100,
|
|
"preview_head": 20,
|
|
"preview_tail": 20,
|
|
"directory": self.tmpdir,
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
def test_empty_and_none_are_noops(self):
|
|
self.assertEqual(hos.spill_if_oversized("", config=self._cfg()), "")
|
|
self.assertEqual(hos.spill_if_oversized(None, config=self._cfg()), "")
|
|
|
|
def test_text_under_cap_is_unchanged(self):
|
|
small = "x" * 50
|
|
self.assertEqual(hos.spill_if_oversized(small, config=self._cfg()), small)
|
|
|
|
def test_disabled_bypasses_spill_even_if_oversized(self):
|
|
big = "y" * 10_000
|
|
cfg = self._cfg(enabled=False)
|
|
self.assertEqual(hos.spill_if_oversized(big, config=cfg), big)
|
|
# No spill files written.
|
|
self.assertEqual(list(Path(self.tmpdir).rglob("*")), [])
|
|
|
|
def test_oversized_writes_spill_and_returns_preview(self):
|
|
big = "A" * 60 + "B" * 60 + "C" * 60 # 180 chars > cap 100
|
|
result = hos.spill_if_oversized(
|
|
big,
|
|
session_id="sess-123",
|
|
source="plugin hook",
|
|
config=self._cfg(),
|
|
)
|
|
# Preview contains the header, head, and tail markers.
|
|
self.assertIn("plugin hook output truncated — 180 chars", result)
|
|
self.assertIn("--- head ---", result)
|
|
self.assertIn("--- tail ---", result)
|
|
# Head is the first 20 chars, tail is the last 20.
|
|
self.assertIn("A" * 20, result)
|
|
self.assertIn("C" * 20, result)
|
|
# Spill file exists under the session subdir and has full content.
|
|
session_dir = Path(self.tmpdir) / "sess-123"
|
|
self.assertTrue(session_dir.is_dir())
|
|
files = list(session_dir.iterdir())
|
|
self.assertEqual(len(files), 1)
|
|
self.assertEqual(files[0].read_text().rstrip("\n"), big)
|
|
# Preview references the spill path.
|
|
self.assertIn(str(files[0]), result)
|
|
|
|
def test_missing_session_id_uses_no_session_segment(self):
|
|
big = "z" * 500
|
|
cfg = self._cfg(max_chars=10)
|
|
hos.spill_if_oversized(big, session_id=None, config=cfg)
|
|
self.assertTrue((Path(self.tmpdir) / "no-session").is_dir())
|
|
|
|
def test_session_id_with_path_separators_is_sanitised(self):
|
|
big = "q" * 500
|
|
cfg = self._cfg(max_chars=10)
|
|
# An attacker-style session id with .. and / must not escape the
|
|
# base directory.
|
|
hos.spill_if_oversized(big, session_id="../../etc/passwd", config=cfg)
|
|
# Nothing leaks outside self.tmpdir.
|
|
self.assertFalse(Path("/etc/passwd-hermes-test").exists())
|
|
# A sanitised path should exist under tmpdir.
|
|
entries = list(Path(self.tmpdir).rglob("*.txt"))
|
|
self.assertEqual(len(entries), 1)
|
|
# The path should be inside tmpdir.
|
|
self.assertTrue(str(entries[0]).startswith(self.tmpdir))
|
|
|
|
def test_spill_write_failure_falls_back_to_preview_only(self):
|
|
big = "w" * 500
|
|
# Point at a path that cannot be created (a file, not a dir).
|
|
existing_file = os.path.join(self.tmpdir, "not-a-dir")
|
|
with open(existing_file, "w") as f:
|
|
f.write("blocker")
|
|
cfg = self._cfg(max_chars=10, directory=existing_file)
|
|
result = hos.spill_if_oversized(big, session_id="x", config=cfg)
|
|
# Preview still returned, but with failure notice.
|
|
self.assertIn("spill write failed", result)
|
|
self.assertIn("--- head ---", result)
|
|
# Content still bounded (not the full 500 chars).
|
|
self.assertLess(len(result), 500)
|
|
|
|
def test_preview_head_only_no_tail(self):
|
|
big = "a" * 1000
|
|
cfg = self._cfg(max_chars=10, preview_head=30, preview_tail=0)
|
|
result = hos.spill_if_oversized(big, session_id="s", config=cfg)
|
|
self.assertIn("--- head ---", result)
|
|
self.assertNotIn("--- tail ---", result)
|
|
|
|
def test_non_string_input_coerced(self):
|
|
cfg = self._cfg(max_chars=5)
|
|
|
|
class StrFriendly:
|
|
def __str__(self):
|
|
return "stringified-" + "x" * 200
|
|
|
|
result = hos.spill_if_oversized(StrFriendly(), session_id="s", config=cfg)
|
|
self.assertIn("truncated", result)
|
|
|
|
def test_default_directory_uses_hermes_home(self):
|
|
"""When no directory override, spill under HERMES_HOME/hook_outputs."""
|
|
test_home = tempfile.mkdtemp(prefix="hermes-home-")
|
|
try:
|
|
with patch.dict(os.environ, {"HERMES_HOME": test_home}):
|
|
# Also patch get_hermes_home to the env var to mirror production.
|
|
cfg = self._cfg(directory=None, max_chars=5)
|
|
hos.spill_if_oversized("x" * 200, session_id="sess", config=cfg)
|
|
# Spill directory exists somewhere under test_home OR default
|
|
# ~/.hermes/hook_outputs depending on get_hermes_home behaviour.
|
|
candidates = [
|
|
Path(test_home) / "hook_outputs" / "sess",
|
|
Path(os.path.expanduser("~/.hermes/hook_outputs/sess")),
|
|
]
|
|
# At least one of the candidate dirs now exists and has a file.
|
|
existing = [c for c in candidates if c.is_dir() and list(c.iterdir())]
|
|
self.assertTrue(existing, f"No spill dir found in {candidates}")
|
|
finally:
|
|
import shutil
|
|
shutil.rmtree(test_home, ignore_errors=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|