mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-27 17:58:07 +00:00
cb06017b1d
Per Teknium: the caps should bound a single agent loop, not accumulate over the whole session. Rename SessionCapConfig -> LoopCapConfig and the config section session_caps -> loop_caps; move the counters into reset_for_turn (invoked per turn via turn_context) so each turn starts with a fresh budget; retune defaults 200 -> 50 (a single turn issuing 50 web searches / spawning 50 subagents is already pathological). Block codes session_*_cap -> loop_*_cap and messages updated to drop the /new-resets-the-budget guidance (irrelevant now that it resets per turn). Tests flipped: the old persists-across-turn-resets assertion becomes resets-each-turn.
389 lines
16 KiB
Python
389 lines
16 KiB
Python
"""Pure tool-call guardrail primitive tests."""
|
|
|
|
import json
|
|
|
|
from agent.tool_guardrails import (
|
|
ToolCallGuardrailConfig,
|
|
ToolCallGuardrailController,
|
|
ToolCallSignature,
|
|
canonical_tool_args,
|
|
classify_tool_failure,
|
|
)
|
|
|
|
|
|
def test_tool_call_signature_hashes_canonical_nested_unicode_args_without_exposing_raw_args():
|
|
args_a = {
|
|
"z": [{"β": "☤", "a": 1}],
|
|
"a": {"y": 2, "x": "secret-token-value"},
|
|
}
|
|
args_b = {
|
|
"a": {"x": "secret-token-value", "y": 2},
|
|
"z": [{"a": 1, "β": "☤"}],
|
|
}
|
|
|
|
assert canonical_tool_args(args_a) == canonical_tool_args(args_b)
|
|
sig_a = ToolCallSignature.from_call("web_search", args_a)
|
|
sig_b = ToolCallSignature.from_call("web_search", args_b)
|
|
|
|
assert sig_a == sig_b
|
|
assert len(sig_a.args_hash) == 64
|
|
metadata = sig_a.to_metadata()
|
|
assert metadata == {"tool_name": "web_search", "args_hash": sig_a.args_hash}
|
|
assert "secret-token-value" not in json.dumps(metadata)
|
|
assert "☤" not in json.dumps(metadata)
|
|
|
|
|
|
def test_default_config_is_soft_warning_only_with_hard_stop_disabled():
|
|
cfg = ToolCallGuardrailConfig()
|
|
|
|
assert cfg.warnings_enabled is True
|
|
assert cfg.hard_stop_enabled is False
|
|
assert cfg.exact_failure_warn_after == 2
|
|
assert cfg.same_tool_failure_warn_after == 3
|
|
assert cfg.no_progress_warn_after == 2
|
|
assert cfg.exact_failure_block_after == 5
|
|
assert cfg.same_tool_failure_halt_after == 8
|
|
assert cfg.no_progress_block_after == 5
|
|
|
|
|
|
def test_config_parses_nested_warn_and_hard_stop_thresholds():
|
|
cfg = ToolCallGuardrailConfig.from_mapping(
|
|
{
|
|
"warnings_enabled": False,
|
|
"hard_stop_enabled": True,
|
|
"warn_after": {
|
|
"exact_failure": 3,
|
|
"same_tool_failure": 4,
|
|
"idempotent_no_progress": 5,
|
|
},
|
|
"hard_stop_after": {
|
|
"exact_failure": 6,
|
|
"same_tool_failure": 7,
|
|
"idempotent_no_progress": 8,
|
|
},
|
|
}
|
|
)
|
|
|
|
assert cfg.warnings_enabled is False
|
|
assert cfg.hard_stop_enabled is True
|
|
assert cfg.exact_failure_warn_after == 3
|
|
assert cfg.same_tool_failure_warn_after == 4
|
|
assert cfg.no_progress_warn_after == 5
|
|
assert cfg.exact_failure_block_after == 6
|
|
assert cfg.same_tool_failure_halt_after == 7
|
|
assert cfg.no_progress_block_after == 8
|
|
|
|
|
|
def test_default_repeated_identical_failed_call_warns_without_blocking():
|
|
controller = ToolCallGuardrailController()
|
|
args = {"query": "same"}
|
|
|
|
decisions = []
|
|
for _ in range(5):
|
|
assert controller.before_call("web_search", args).action == "allow"
|
|
decisions.append(
|
|
controller.after_call("web_search", args, '{"error":"boom"}', failed=True)
|
|
)
|
|
|
|
assert decisions[0].action == "allow"
|
|
assert [d.action for d in decisions[1:]] == ["warn", "warn", "warn", "warn"]
|
|
assert {d.code for d in decisions[1:]} == {"repeated_exact_failure_warning"}
|
|
assert controller.before_call("web_search", args).action == "allow"
|
|
assert controller.halt_decision is None
|
|
|
|
|
|
def test_hard_stop_enabled_blocks_repeated_exact_failure_before_next_execution():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(
|
|
hard_stop_enabled=True,
|
|
exact_failure_warn_after=2,
|
|
exact_failure_block_after=2,
|
|
same_tool_failure_halt_after=99,
|
|
)
|
|
)
|
|
args = {"query": "same"}
|
|
|
|
assert controller.before_call("web_search", args).action == "allow"
|
|
first = controller.after_call("web_search", args, '{"error":"boom"}', failed=True)
|
|
assert first.action == "allow"
|
|
|
|
assert controller.before_call("web_search", args).action == "allow"
|
|
second = controller.after_call("web_search", args, '{"error":"boom"}', failed=True)
|
|
assert second.action == "warn"
|
|
assert second.code == "repeated_exact_failure_warning"
|
|
|
|
blocked = controller.before_call("web_search", args)
|
|
assert blocked.action == "block"
|
|
assert blocked.code == "repeated_exact_failure_block"
|
|
assert blocked.count == 2
|
|
|
|
|
|
def test_success_resets_exact_signature_failure_streak():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(hard_stop_enabled=True, exact_failure_block_after=2, same_tool_failure_halt_after=99)
|
|
)
|
|
args = {"query": "same"}
|
|
|
|
controller.after_call("web_search", args, '{"error":"boom"}', failed=True)
|
|
controller.after_call("web_search", args, '{"ok":true}', failed=False)
|
|
|
|
assert controller.before_call("web_search", args).action == "allow"
|
|
controller.after_call("web_search", args, '{"error":"boom"}', failed=True)
|
|
assert controller.before_call("web_search", args).action == "allow"
|
|
|
|
|
|
def test_file_mutation_lint_error_result_is_not_a_tool_failure():
|
|
write_result = json.dumps({
|
|
"bytes_written": 12,
|
|
"lint": {"status": "error", "output": "SyntaxError: invalid syntax"},
|
|
})
|
|
patch_result = json.dumps({
|
|
"success": True,
|
|
"diff": "--- a/tmp.py\n+++ b/tmp.py\n",
|
|
"lsp_diagnostics": "<diagnostics>ERROR [1:1] type mismatch</diagnostics>",
|
|
})
|
|
|
|
assert classify_tool_failure("write_file", write_result) == (False, "")
|
|
assert classify_tool_failure("patch", patch_result) == (False, "")
|
|
|
|
|
|
def test_same_tool_varying_args_warns_by_default_without_halting():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(same_tool_failure_warn_after=2, same_tool_failure_halt_after=3)
|
|
)
|
|
|
|
first = controller.after_call("terminal", {"command": "cmd-1"}, '{"exit_code":1}', failed=True)
|
|
second = controller.after_call("terminal", {"command": "cmd-2"}, '{"exit_code":1}', failed=True)
|
|
third = controller.after_call("terminal", {"command": "cmd-3"}, '{"exit_code":1}', failed=True)
|
|
fourth = controller.after_call("terminal", {"command": "cmd-4"}, '{"exit_code":1}', failed=True)
|
|
|
|
assert first.action == "allow"
|
|
assert [second.action, third.action, fourth.action] == ["warn", "warn", "warn"]
|
|
assert {second.code, third.code, fourth.code} == {"same_tool_failure_warning"}
|
|
assert "Do not switch to text-only replies" in second.message
|
|
assert "keep using tools" in second.message
|
|
assert "diagnose before retrying" in second.message
|
|
assert "different tool" in second.message
|
|
assert controller.halt_decision is None
|
|
|
|
|
|
def test_hard_stop_enabled_halts_same_tool_varying_args_failure_streak():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(
|
|
hard_stop_enabled=True,
|
|
exact_failure_block_after=99,
|
|
same_tool_failure_warn_after=2,
|
|
same_tool_failure_halt_after=3,
|
|
)
|
|
)
|
|
|
|
first = controller.after_call("terminal", {"command": "cmd-1"}, '{"exit_code":1}', failed=True)
|
|
assert first.action == "allow"
|
|
second = controller.after_call("terminal", {"command": "cmd-2"}, '{"exit_code":1}', failed=True)
|
|
assert second.action == "warn"
|
|
assert second.code == "same_tool_failure_warning"
|
|
third = controller.after_call("terminal", {"command": "cmd-3"}, '{"exit_code":1}', failed=True)
|
|
assert third.action == "halt"
|
|
assert third.code == "same_tool_failure_halt"
|
|
assert third.count == 3
|
|
|
|
|
|
def test_idempotent_no_progress_repeated_result_warns_without_blocking_by_default():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(no_progress_warn_after=2, no_progress_block_after=2)
|
|
)
|
|
args = {"path": "/tmp/same.txt"}
|
|
result = "same file contents"
|
|
|
|
for _ in range(4):
|
|
assert controller.before_call("read_file", args).action == "allow"
|
|
decision = controller.after_call("read_file", args, result, failed=False)
|
|
|
|
assert decision.action == "warn"
|
|
assert decision.code == "idempotent_no_progress_warning"
|
|
assert controller.before_call("read_file", args).action == "allow"
|
|
assert controller.halt_decision is None
|
|
|
|
|
|
def test_hard_stop_enabled_blocks_idempotent_no_progress_future_repeat():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(
|
|
hard_stop_enabled=True,
|
|
no_progress_warn_after=2,
|
|
no_progress_block_after=2,
|
|
)
|
|
)
|
|
args = {"path": "/tmp/same.txt"}
|
|
result = "same file contents"
|
|
|
|
assert controller.before_call("read_file", args).action == "allow"
|
|
assert controller.after_call("read_file", args, result, failed=False).action == "allow"
|
|
assert controller.before_call("read_file", args).action == "allow"
|
|
warn = controller.after_call("read_file", args, result, failed=False)
|
|
assert warn.action == "warn"
|
|
assert warn.code == "idempotent_no_progress_warning"
|
|
|
|
blocked = controller.before_call("read_file", args)
|
|
assert blocked.action == "block"
|
|
assert blocked.code == "idempotent_no_progress_block"
|
|
|
|
|
|
def test_mutating_or_unknown_tools_are_not_blocked_for_repeated_identical_success_output_by_default():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(no_progress_warn_after=2, no_progress_block_after=2)
|
|
)
|
|
|
|
for _ in range(3):
|
|
assert controller.before_call("write_file", {"path": "/tmp/x", "content": "x"}).action == "allow"
|
|
assert controller.after_call("write_file", {"path": "/tmp/x", "content": "x"}, "ok", failed=False).action == "allow"
|
|
assert controller.before_call("custom_tool", {"x": 1}).action == "allow"
|
|
assert controller.after_call("custom_tool", {"x": 1}, "ok", failed=False).action == "allow"
|
|
|
|
|
|
def test_reset_for_turn_clears_bounded_guardrail_state():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(hard_stop_enabled=True, exact_failure_block_after=2, no_progress_block_after=2)
|
|
)
|
|
controller.after_call("web_search", {"query": "same"}, '{"error":"boom"}', failed=True)
|
|
controller.after_call("web_search", {"query": "same"}, '{"error":"boom"}', failed=True)
|
|
controller.after_call("read_file", {"path": "/tmp/x"}, "same", failed=False)
|
|
controller.after_call("read_file", {"path": "/tmp/x"}, "same", failed=False)
|
|
|
|
assert controller.before_call("web_search", {"query": "same"}).action == "block"
|
|
assert controller.before_call("read_file", {"path": "/tmp/x"}).action == "block"
|
|
|
|
controller.reset_for_turn()
|
|
|
|
assert controller.before_call("web_search", {"query": "same"}).action == "allow"
|
|
assert controller.before_call("read_file", {"path": "/tmp/x"}).action == "allow"
|
|
|
|
|
|
def test_after_call_survives_lone_surrogates_in_result_and_args():
|
|
# Scraped web/social text can contain unpaired UTF-16 surrogates (e.g. the
|
|
# first half of a mathematical-bold pair, '\ud835'). str.encode('utf-8')
|
|
# rejects them, and the result hasher crashed the whole conversation loop
|
|
# (live outage: "Outer loop error in API call #34 ... surrogates not
|
|
# allowed"). Weird text must never take down the loop.
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(hard_stop_enabled=True, exact_failure_block_after=2, no_progress_block_after=2)
|
|
)
|
|
dirty = "price \ud835 update"
|
|
|
|
decision = controller.after_call("web_search", {"query": dirty}, dirty, failed=False)
|
|
assert decision.action in {"allow", "warn"}
|
|
|
|
# hashing stays deterministic: the same dirty failure twice still trips
|
|
# the exact-failure guard, proving the hash is stable across calls
|
|
controller.after_call("web_search", {"query": dirty}, '{"error":"\ud835 boom"}', failed=True)
|
|
controller.after_call("web_search", {"query": dirty}, '{"error":"\ud835 boom"}', failed=True)
|
|
assert controller.before_call("web_search", {"query": dirty}).action == "block"
|
|
|
|
|
|
# ── Per-turn runaway-loop caps (Claude Code v2.1.212, Week 29) ──────────────
|
|
|
|
from agent.tool_guardrails import LoopCapConfig # noqa: E402
|
|
|
|
|
|
def test_loop_cap_defaults():
|
|
caps = ToolCallGuardrailConfig().loop_caps
|
|
assert caps.max_web_searches == 50
|
|
assert caps.max_subagents == 50
|
|
|
|
|
|
def test_loop_cap_config_parses_nested_section():
|
|
cfg = ToolCallGuardrailConfig.from_mapping(
|
|
{"loop_caps": {"max_web_searches": 3, "max_subagents": 0}}
|
|
)
|
|
assert cfg.loop_caps.max_web_searches == 3
|
|
assert cfg.loop_caps.max_subagents == 0
|
|
|
|
|
|
def test_loop_cap_zero_disables_and_junk_falls_back():
|
|
# 0 is a legitimate "unlimited" value; negatives / junk fall back to default.
|
|
assert LoopCapConfig.from_mapping({"max_web_searches": 0}).max_web_searches == 0
|
|
assert LoopCapConfig.from_mapping({"max_web_searches": -5}).max_web_searches == 50
|
|
assert LoopCapConfig.from_mapping({"max_subagents": "nope"}).max_subagents == 50
|
|
|
|
|
|
def test_web_search_cap_blocks_after_limit_regardless_of_hard_stop():
|
|
# Loop caps fire even with hard_stop_enabled=False (the per-turn loop
|
|
# detector's flag). Each distinct query avoids the loop detector so we know
|
|
# the block came from the loop cap, not exact-failure repetition.
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(
|
|
hard_stop_enabled=False,
|
|
loop_caps=LoopCapConfig(max_web_searches=3),
|
|
)
|
|
)
|
|
for i in range(3):
|
|
assert controller.before_call("web_search", {"query": f"q{i}"}).action == "allow"
|
|
decision = controller.before_call("web_search", {"query": "q4"})
|
|
assert decision.action == "block"
|
|
assert decision.code == "loop_web_search_cap"
|
|
assert decision.should_halt is True
|
|
|
|
|
|
def test_web_search_cap_resets_each_turn():
|
|
# The cap bounds a single turn: reset_for_turn clears the counter so a
|
|
# legitimate multi-turn session is never starved.
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_web_searches=2))
|
|
)
|
|
# Turn 1: two searches allowed, the third would block within the turn.
|
|
assert controller.before_call("web_search", {"query": "a"}).action == "allow"
|
|
assert controller.before_call("web_search", {"query": "b"}).action == "allow"
|
|
assert controller.before_call("web_search", {"query": "c"}).action == "block"
|
|
# New turn: the counter resets, so the budget is fresh again.
|
|
controller.reset_for_turn()
|
|
assert controller.before_call("web_search", {"query": "d"}).action == "allow"
|
|
assert controller.before_call("web_search", {"query": "e"}).action == "allow"
|
|
assert controller.before_call("web_search", {"query": "f"}).action == "block"
|
|
|
|
|
|
def test_subagent_cap_counts_batch_task_spawns():
|
|
# A single delegate_task batch of N tasks spends N of the subagent budget.
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_subagents=5))
|
|
)
|
|
# First call spawns 3 (batch) → count 3, allowed.
|
|
assert controller.before_call(
|
|
"delegate_task", {"tasks": [{"goal": "a"}, {"goal": "b"}, {"goal": "c"}]}
|
|
).action == "allow"
|
|
# Second call spawns 1 (goal) → count 4, allowed.
|
|
assert controller.before_call("delegate_task", {"goal": "d"}).action == "allow"
|
|
# Count is 4 (< 5) so this is allowed and bumps to 5.
|
|
assert controller.before_call("delegate_task", {"goal": "e"}).action == "allow"
|
|
# Now count is 5 (>= 5) so the next call is blocked.
|
|
decision = controller.before_call("delegate_task", {"goal": "f"})
|
|
assert decision.action == "block"
|
|
assert decision.code == "loop_subagent_cap"
|
|
|
|
|
|
def test_subagent_cap_resets_each_turn():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_subagents=1))
|
|
)
|
|
assert controller.before_call("delegate_task", {"goal": "a"}).action == "allow"
|
|
assert controller.before_call("delegate_task", {"goal": "b"}).action == "block"
|
|
controller.reset_for_turn()
|
|
assert controller.before_call("delegate_task", {"goal": "c"}).action == "allow"
|
|
|
|
|
|
def test_loop_caps_disabled_when_zero():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(
|
|
loop_caps=LoopCapConfig(max_web_searches=0, max_subagents=0)
|
|
)
|
|
)
|
|
for i in range(60):
|
|
assert controller.before_call("web_search", {"query": f"q{i}"}).action == "allow"
|
|
assert controller.before_call("delegate_task", {"goal": f"g{i}"}).action == "allow"
|
|
|
|
|
|
def test_other_tools_never_touched_by_loop_caps():
|
|
controller = ToolCallGuardrailController(
|
|
ToolCallGuardrailConfig(loop_caps=LoopCapConfig(max_web_searches=1))
|
|
)
|
|
# read_file / terminal / etc. are unaffected regardless of the web cap.
|
|
for _ in range(10):
|
|
assert controller.before_call("read_file", {"path": "/tmp/x"}).action == "allow"
|