feat(approvals): consecutive-denial circuit breaker for smart approvals

After N consecutive guardian denials in a session the deny message escalates to a hard-stop instruction. Inspired by ChatGPT Work auto-review circuit breaker.

hermes_cli/config.py

@@ -2747,6 +2747,13 @@ DEFAULT_CONFIG = {
         # 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

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)

tools/approval.py

@@ -2018,6 +2018,81 @@ _session_approved: dict[str, set] = {}
 _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}
 

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: