diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 2f31fef19b65..7ecb2e0a4db0 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -2747,6 +2747,13 @@ def _ensure_hermes_home_managed(home: Path): # compose restarts under ~/deploys". Inspired by ChatGPT Work's # customizable auto-review guardian policy. "smart_policy": "", + # Consecutive-denial circuit breaker for smart approvals: after this + # many guardian DENY verdicts in a row within one session, the deny + # message returned to the model escalates to a hard-stop instruction + # (report to the user / ask for manual run or /approve) instead of a + # plain "Do NOT retry". Any approval resets the count. 0 disables. + # Inspired by ChatGPT Work's auto-review circuit breaker. + "denial_breaker_threshold": 3, # User-defined deny rules: fnmatch globs matched against terminal # commands. A match blocks the command unconditionally — BEFORE the # --yolo / /yolo / mode=off bypass — making this the user-editable diff --git a/tests/tools/test_denial_circuit_breaker.py b/tests/tools/test_denial_circuit_breaker.py new file mode 100644 index 000000000000..919df114cafb --- /dev/null +++ b/tests/tools/test_denial_circuit_breaker.py @@ -0,0 +1,276 @@ +"""Tests for the consecutive-denial circuit breaker in smart approvals. + +After ``approvals.denial_breaker_threshold`` consecutive guardian DENY +verdicts in one session, the deny message returned to the model escalates +from "Do NOT retry" to a hard-stop CIRCUIT BREAKER instruction. Any +approval resets the tally. State is per-session and capped in size. + +Follows the existing smart-approval mocking patterns from +tests/tools/test_execute_code_approval_cluster.py: monkeypatch +``_smart_approve`` / ``_get_approval_mode`` on the module and drive the +public guard entry points. +""" + +from __future__ import annotations + +import pytest + +from tools import approval as A + +BREAKER_MARKER = "CIRCUIT BREAKER:" + + +@pytest.fixture +def breaker_session(monkeypatch): + """A clean gateway smart-mode session with the guardian forced to DENY. + + Uses the gateway path with a notify callback that resolves 'deny' + (user denies the smart-DENY override) so the guard returns a definitive + BLOCKED message — the channel the breaker text rides on. + """ + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.delenv("HERMES_EXEC_ASK", raising=False) + monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False) + monkeypatch.setattr(A, "_smart_approve", lambda _c, _d: "deny") + monkeypatch.setattr(A, "_get_denial_breaker_threshold", lambda: 3) + monkeypatch.setattr( + A, "detect_dangerous_command", + lambda command: (True, "breaker-test-danger", f"risk:{command}"), + ) + monkeypatch.setattr( + "tools.tirith_security.check_command_security", + lambda _command: {"action": "allow", "findings": [], "summary": ""}, + raising=False, + ) + + session_key = "breaker-test-session" + token = A.set_current_session_key(session_key) + A._reset_denials(session_key) + with A._lock: + A._permanent_approved.discard("breaker-test-danger") + A._permanent_approved.discard("execute_code") + A._session_approved.get(session_key, set()).discard("breaker-test-danger") + A._session_approved.get(session_key, set()).discard("execute_code") + A._gateway_queues.pop(session_key, None) + A._gateway_notify_cbs.pop(session_key, None) + try: + yield session_key + finally: + A.reset_current_session_key(token) + A._reset_denials(session_key) + with A._lock: + A._gateway_queues.pop(session_key, None) + A._gateway_notify_cbs.pop(session_key, None) + + +def _register_resolver(session_key: str, result): + """Notify callback resolving the newest queued approval with *result*.""" + def cb(_approval_data): + with A._lock: + entries = A._gateway_queues.get(session_key, []) + if entries: + entries[-1].result = result + entries[-1].event.set() + with A._lock: + A._gateway_notify_cbs[session_key] = cb + + +def _denied_terminal(command="dangerous thing"): + return A.check_all_command_guards(command, "local") + + +def _denied_execute_code(code="print('x')"): + return A.check_execute_code_guard(code, "local") + + +# --------------------------------------------------------------------------- +# (a) Two denials -> normal message; third -> breaker text present +# --------------------------------------------------------------------------- + +def test_breaker_trips_on_third_consecutive_denial(breaker_session): + _register_resolver(breaker_session, "deny") + + first = _denied_terminal("dangerous one") + second = _denied_terminal("dangerous two") + third = _denied_terminal("dangerous three") + + assert first["approved"] is False + assert BREAKER_MARKER not in first["message"] + assert second["approved"] is False + assert BREAKER_MARKER not in second["message"] + assert third["approved"] is False + assert BREAKER_MARKER in third["message"] + assert "3 consecutive commands were blocked" in third["message"] + assert "STOP attempting variations" in third["message"] + + +# --------------------------------------------------------------------------- +# (b) An approval resets the tally +# --------------------------------------------------------------------------- + +def test_approval_resets_tally(breaker_session, monkeypatch): + _register_resolver(breaker_session, "deny") + _denied_terminal("dangerous one") + _denied_terminal("dangerous two") + + # Guardian approves the next command → tally resets. + monkeypatch.setattr(A, "_smart_approve", lambda _c, _d: "approve") + ok = _denied_terminal("benign command") + assert ok["approved"] is True and ok.get("smart_approved") is True + + # Back to denials: the count restarts, so the next deny is #1, not #3. + monkeypatch.setattr(A, "_smart_approve", lambda _c, _d: "deny") + after = _denied_terminal("dangerous again") + assert after["approved"] is False + assert BREAKER_MARKER not in after["message"] + + +def test_human_approval_resets_tally(breaker_session): + _register_resolver(breaker_session, "deny") + _denied_terminal("dangerous one") + _denied_terminal("dangerous two") + + # User overrides the smart DENY (one-operation approval) → tally resets. + _register_resolver(breaker_session, "once") + ok = _denied_terminal("dangerous but user says yes") + assert ok["approved"] is True and ok.get("user_approved") is True + + _register_resolver(breaker_session, "deny") + after = _denied_terminal("dangerous again") + assert after["approved"] is False + assert BREAKER_MARKER not in after["message"] + + +# --------------------------------------------------------------------------- +# (c) Threshold 0 disables the breaker +# --------------------------------------------------------------------------- + +def test_threshold_zero_disables_breaker(breaker_session, monkeypatch): + monkeypatch.setattr(A, "_get_denial_breaker_threshold", lambda: 0) + _register_resolver(breaker_session, "deny") + for i in range(5): + res = _denied_terminal(f"dangerous {i}") + assert res["approved"] is False + assert BREAKER_MARKER not in res["message"] + + +# --------------------------------------------------------------------------- +# (d) Tally is per-session — two session keys are independent +# --------------------------------------------------------------------------- + +def test_tally_is_per_session(breaker_session): + other = "breaker-other-session" + A._reset_denials(other) + try: + assert A._record_denial(breaker_session) == 1 + assert A._record_denial(breaker_session) == 2 + # A different session starts from zero. + assert A._record_denial(other) == 1 + # And its denial did not advance the first session's count. + assert A._record_denial(breaker_session) == 3 + # Resetting one session leaves the other intact. + A._reset_denials(breaker_session) + assert A._record_denial(other) == 2 + assert A._record_denial(breaker_session) == 1 + finally: + A._reset_denials(other) + + +# --------------------------------------------------------------------------- +# (e) BOTH call paths increment: terminal guard and execute_code guard +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("deny_call", [_denied_terminal, _denied_execute_code], + ids=["terminal", "execute_code"]) +def test_both_paths_increment_and_trip(breaker_session, deny_call): + _register_resolver(breaker_session, "deny") + for _ in range(2): + res = deny_call() + assert res["approved"] is False + assert BREAKER_MARKER not in res["message"] + tripped = deny_call() + assert tripped["approved"] is False + assert BREAKER_MARKER in tripped["message"] + + +def test_paths_share_one_session_tally(breaker_session): + """Denials from the terminal and execute_code paths accumulate together.""" + _register_resolver(breaker_session, "deny") + assert BREAKER_MARKER not in _denied_terminal("dangerous one")["message"] + assert BREAKER_MARKER not in _denied_execute_code()["message"] + tripped = _denied_terminal("dangerous three") + assert BREAKER_MARKER in tripped["message"] + + +# --------------------------------------------------------------------------- +# Headless hard-deny path (no cli/gateway/ask override) also increments +# --------------------------------------------------------------------------- + +def test_headless_smart_deny_increments_and_trips(monkeypatch): + monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False) + monkeypatch.delenv("HERMES_INTERACTIVE", raising=False) + monkeypatch.delenv("HERMES_CRON_SESSION", raising=False) + monkeypatch.setenv("HERMES_EXEC_ASK", "0") + monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart") + monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False) + monkeypatch.setattr(A, "_smart_approve", lambda _c, _d: "deny") + monkeypatch.setattr(A, "_get_denial_breaker_threshold", lambda: 3) + monkeypatch.setattr(A, "_is_interactive_cli", lambda: True) + monkeypatch.setattr( + A, "detect_dangerous_command", + lambda command: (True, "headless-breaker-danger", f"risk:{command}"), + ) + monkeypatch.setattr( + "tools.tirith_security.check_command_security", + lambda _command: {"action": "allow", "findings": [], "summary": ""}, + raising=False, + ) + # CLI-interactive path: the owner denies via the prompt callback. + monkeypatch.setattr(A, "prompt_dangerous_approval", + lambda *args, **kwargs: "deny") + + session_key = "headless-breaker-session" + token = A.set_current_session_key(session_key) + A._reset_denials(session_key) + with A._lock: + A._permanent_approved.discard("headless-breaker-danger") + A._session_approved.get(session_key, set()).discard( + "headless-breaker-danger") + try: + first = A.check_all_command_guards("dangerous h1", "local") + second = A.check_all_command_guards("dangerous h2", "local") + third = A.check_all_command_guards("dangerous h3", "local") + assert BREAKER_MARKER not in first["message"] + assert BREAKER_MARKER not in second["message"] + assert BREAKER_MARKER in third["message"] + finally: + A.reset_current_session_key(token) + A._reset_denials(session_key) + + +# --------------------------------------------------------------------------- +# Eviction cap: the tally dict never grows past _DENIAL_TALLY_MAX_SESSIONS +# --------------------------------------------------------------------------- + +def test_tally_evicts_oldest_sessions(): + with A._lock: + saved = dict(A._denial_tally) + A._denial_tally.clear() + try: + for i in range(A._DENIAL_TALLY_MAX_SESSIONS + 10): + A._record_denial(f"evict-session-{i}") + with A._lock: + assert len(A._denial_tally) == A._DENIAL_TALLY_MAX_SESSIONS + # Oldest entries were evicted, newest survive. + assert "evict-session-0" not in A._denial_tally + assert ( + f"evict-session-{A._DENIAL_TALLY_MAX_SESSIONS + 9}" + in A._denial_tally + ) + finally: + with A._lock: + A._denial_tally.clear() + A._denial_tally.update(saved) diff --git a/tools/approval.py b/tools/approval.py index 7afd0edf8969..f7d15b8ab2a6 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -2018,6 +2018,81 @@ def detect_dangerous_command(command: str) -> tuple: _session_yolo: set[str] = set() _permanent_approved: set = set() +# ========================================================================= +# Consecutive-denial circuit breaker for smart approvals +# ========================================================================= +# Nothing stops the model from retrying variants of a smart-denied command — +# each retry burns another guardian LLM call and agent iteration. After +# ``approvals.denial_breaker_threshold`` consecutive guardian DENY verdicts +# in one session (default 3; 0 disables), the deny message returned to the +# model escalates to a hard-stop instruction. Any approval resets the tally. +# This changes only the TOOL RESULT text — no message-history surgery, no +# interrupts — so it is prompt-cache-invariant by construction. Inspired by +# ChatGPT Work's auto-review circuit breaker (3 consecutive denials). +_denial_tally: dict[str, int] = {} +# Plain dict with a small cap so an army of short-lived session keys cannot +# grow it without bound; oldest (least recently denied) entries are evicted. +_DENIAL_TALLY_MAX_SESSIONS = 256 + + +def _get_denial_breaker_threshold() -> int: + """Read ``approvals.denial_breaker_threshold`` from config. + + Defaults to 3 consecutive guardian denials; 0 (or negative) disables + the breaker entirely. + """ + try: + return int(_get_approval_config().get("denial_breaker_threshold", 3)) + except (ValueError, TypeError): + return 3 + + +def _record_denial(session_key: str) -> int: + """Increment and return the session's consecutive guardian-denial count. + + Pop-and-reinsert keeps actively-denying sessions at the most-recent end + of the dict so eviction (insertion-ordered) drops genuinely idle keys. + """ + with _lock: + count = _denial_tally.pop(session_key, 0) + 1 + _denial_tally[session_key] = count + while len(_denial_tally) > _DENIAL_TALLY_MAX_SESSIONS: + _denial_tally.pop(next(iter(_denial_tally))) + return count + + +def _reset_denials(session_key: str) -> None: + """Clear the session's consecutive-denial tally (an approval happened).""" + with _lock: + _denial_tally.pop(session_key, None) + + +def _denial_breaker_addendum(session_key: str) -> str: + """Return the escalated hard-stop text when the breaker has tripped. + + Read-only: callers increment via :func:`_record_denial` on the guardian + DENY verdict; this just checks the session's tally against the + configured threshold. Returns '' below the threshold (or when + disabled), otherwise a leading-space addendum the caller appends + verbatim to the deny message returned to the model. + """ + with _lock: + count = _denial_tally.get(session_key, 0) + threshold = _get_denial_breaker_threshold() + if threshold <= 0 or count < threshold: + return "" + logger.warning( + "Smart-approval circuit breaker tripped for session %s: " + "%d consecutive denials (threshold %d)", + session_key, count, threshold, + ) + return ( + f" CIRCUIT BREAKER: {count} consecutive commands were blocked by " + "the security reviewer. STOP attempting variations of this " + "operation. Report the blocked operation to the user and either " + "ask them to run it manually or use /approve." + ) + # ========================================================================= # Blocking gateway approval (mirrors CLI's synchronous input() flow) # ========================================================================= @@ -3430,19 +3505,27 @@ def check_all_command_guards(command: str, env_type: str, # Approve this command only. Pattern-level persistence would let one # benign command suppress review of later commands that happen to # match the same broad detector category. + _reset_denials(session_key) logger.debug("Smart approval: auto-approved '%s' (%s)", command[:60], combined_desc_for_llm) return {"approved": True, "message": None, "smart_approved": True, "description": combined_desc_for_llm} elif verdict == "deny" and not (is_cli or is_gateway or is_ask): + _record_denial(session_key) + breaker_addendum = _denial_breaker_addendum(session_key) return { "approved": False, "message": f"BLOCKED by smart approval: {combined_desc_for_llm}. " - "The command was assessed as genuinely dangerous. Do NOT retry.", + "The command was assessed as genuinely dangerous. " + f"Do NOT retry.{breaker_addendum}", "smart_denied": True, } elif verdict == "deny": + # Guardian DENY that falls through to a one-operation human + # override still counts toward the consecutive-denial breaker; + # a subsequent human approval resets the tally below. + _record_denial(session_key) smart_denied_for_owner = True # An interactive owner may override DENY for this operation only. # ESCALATE follows the normal, potentially persistent manual behavior. @@ -3537,6 +3620,7 @@ def check_all_command_guards(command: str, env_type: str, reason_addendum = "" if outcome == "denied" and deny_reason: reason_addendum = f' Reason given by the user: "{deny_reason}".' + breaker_addendum = _denial_breaker_addendum(session_key) return { "approved": False, "message": ( @@ -3546,7 +3630,7 @@ def check_all_command_guards(command: str, env_type: str, f"same outcome via a different command. Stop the " f"current workflow and wait for the user to respond " f"before taking any further destructive or " - f"irreversible action.{timeout_addendum}" + f"irreversible action.{timeout_addendum}{breaker_addendum}" ), "pattern_key": primary_key, "description": combined_desc, @@ -3567,6 +3651,9 @@ def check_all_command_guards(command: str, env_type: str, approve_permanent(key) save_permanent_allowlist(_permanent_approved) + # A human approval (including an ESCALATE-then-approve or a + # smart-DENY owner override) resets the consecutive-denial tally. + _reset_denials(session_key) return {"approved": True, "message": None, "user_approved": True, "description": combined_desc} @@ -3631,6 +3718,7 @@ def check_all_command_guards(command: str, env_type: str, ) if choice == "deny": + breaker_addendum = _denial_breaker_addendum(session_key) return { "approved": False, "message": ( @@ -3638,8 +3726,8 @@ def check_all_command_guards(command: str, env_type: str, "to this action. Do NOT retry this command, do NOT rephrase " "it, and do NOT attempt the same outcome via a different " "command. Stop the current workflow and wait for the user " - "to respond before taking any further destructive or " - "irreversible action." + f"to respond before taking any further destructive or " + f"irreversible action.{breaker_addendum}" ), "pattern_key": primary_key, "description": combined_desc, @@ -3660,6 +3748,8 @@ def check_all_command_guards(command: str, env_type: str, approve_permanent(key) save_permanent_allowlist(_permanent_approved) + # A human approval resets the consecutive-denial tally. + _reset_denials(session_key) return {"approved": True, "message": None, "user_approved": True, "description": combined_desc} @@ -3761,16 +3851,19 @@ def check_execute_code_guard(code: str, env_type: str, verdict = _smart_approve(command, description) _observe_smart_approval_verdict(observer_payload, verdict) if verdict == "approve": + _reset_denials(session_key) logger.debug("Smart approval: auto-approved execute_code for session %s", session_key) return {"approved": True, "message": None, "smart_approved": True, "description": description} if verdict == "deny" and not (is_gateway or is_ask): + _record_denial(session_key) + breaker_addendum = _denial_breaker_addendum(session_key) return { "approved": False, "message": ("BLOCKED by smart approval: execute_code script " "execution was assessed as genuinely dangerous. " - "Do NOT retry."), + f"Do NOT retry.{breaker_addendum}"), "smart_denied": True, "pattern_key": pattern_key, "description": description, @@ -3778,6 +3871,10 @@ def check_execute_code_guard(code: str, env_type: str, "user_consent": False, } if verdict == "deny": + # Guardian DENY that falls through to a one-operation human + # override still counts toward the consecutive-denial breaker; + # a subsequent human approval resets the tally below. + _record_denial(session_key) smart_denied_for_owner = True # Interactive DENY falls through to one-operation human approval; # ESCALATE retains the normal manual approval behavior. @@ -3859,13 +3956,14 @@ def check_execute_code_guard(code: str, env_type: str, reason_addendum = "" if resolved and choice == "deny" and deny_reason: reason_addendum = f' Reason given by the user: "{deny_reason}".' + breaker_addendum = _denial_breaker_addendum(session_key) return { "approved": False, "message": ( f"BLOCKED: execute_code script {reason}.{reason_addendum} The " f"user has NOT consented to running this code. Do NOT retry, " f"do NOT rephrase the script, and do NOT attempt the same " - f"outcome via a different tool.{addendum}" + f"outcome via a different tool.{addendum}{breaker_addendum}" ), "pattern_key": pattern_key, "description": description, @@ -3886,6 +3984,8 @@ def check_execute_code_guard(code: str, env_type: str, save_permanent_allowlist(_permanent_approved) # choice == "once": no persistence — approval lasts this single call only. + # A human approval resets the consecutive-denial tally. + _reset_denials(session_key) return {"approved": True, "message": None, "user_approved": True, "description": description} diff --git a/website/docs/user-guide/configuration.md b/website/docs/user-guide/configuration.md index 2f03c6cb8641..977fa6be2605 100644 --- a/website/docs/user-guide/configuration.md +++ b/website/docs/user-guide/configuration.md @@ -2040,6 +2040,15 @@ Smart mode is particularly useful for reducing approval fatigue — it lets the Setting `approvals.mode: off` disables all safety checks for terminal commands. Only use this in trusted, sandboxed environments. ::: +### Denial circuit breaker + +`approvals.denial_breaker_threshold` (default `3`) guards against the agent retrying variations of a command the smart-approval reviewer keeps denying — each retry burns another guardian LLM call. After that many consecutive denials in a session, the deny message escalates to a hard-stop instruction telling the agent to stop, report the blocked operation, and ask you to run it manually or `/approve`. Any approval resets the count; set `0` to disable: + +```yaml +approvals: + denial_breaker_threshold: 3 # 0 disables the breaker +``` + ### Deny rules `approvals.deny` is a list of glob patterns that block matching terminal commands unconditionally — even under `--yolo`, `/yolo`, or `mode: off`. It's the user-editable counterpart to the built-in hardline blocklist: