butler/tool_guardrails.py
Rama main · réplica de lectura del repositorio original.
"""Tool-call loop guardrail primitives for Butler v3.
Two-phase (before_call + after_call) loop detector that tracks repeated
failures and non-progressing idempotent reads within a single agent turn.
Actions: allow, warn, block, halt.
Ported from Butler v1, adapted for Hermes tool naming.
"""
from __future__ import annotations
import hashlib
import json
import os
import threading
from collections import deque
from dataclasses import dataclass
from typing import Any, Mapping, cast
import logging
from butler.defaults.env_defaults import DOOM_LOOP_MODE_DEFAULT
logger = logging.getLogger(__name__)
IDEMPOTENT_TOOLS = frozenset({
"read_file",
"search_code",
"search_files",
"list_directory",
"git_status",
"git_diff",
"git_log",
"skills_list",
"skill_view",
"butler_recall",
"list_runtime_jobs", # 仍为 registry 工具名（非仅 CLI `butler runtime list`）
})
MUTATING_TOOLS = frozenset({
"run_shell",
"terminal",
"write_file",
"edit_file",
"patch",
"git_add",
"git_commit",
"run_runtime_job", # registry 工具；mutating job 须 /批准运行
"download_file",
})
@dataclass(frozen=True)
class GuardrailConfig:
warnings_enabled: bool = True
hard_stop_enabled: bool = True
exact_failure_warn_after: int = 2
exact_failure_block_after: int = 5
same_tool_failure_warn_after: int = 3
same_tool_failure_halt_after: int = 8
no_progress_warn_after: int = 2
no_progress_block_after: int = 5
@dataclass(frozen=True)
class ToolCallSignature:
tool_name: str
args_hash: str
@classmethod
def from_call(cls, tool_name: str, args: Mapping[str, Any] | None) -> ToolCallSignature:
canonical = _canonical_tool_args(_coerce_args(args))
return cls(tool_name=tool_name, args_hash=_sha256(canonical))
@dataclass(frozen=True)
class GuardrailDecision:
action: str = "allow"
code: str = "allow"
message: str = ""
tool_name: str = ""
count: int = 0
@property
def allows_execution(self) -> bool:
return self.action in {"allow", "warn"}
@property
def should_halt(self) -> bool:
return self.action in {"block", "halt"}
def _canonical_tool_args(args: Mapping[str, Any]) -> str:
return json.dumps(args, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
def _coerce_args(args: Mapping[str, Any] | None) -> Mapping[str, Any]:
return args if isinstance(args, Mapping) else {}
def _sha256(value: str) -> str:
return hashlib.sha256(value.encode("utf-8")).hexdigest()
def doom_loop_threshold() -> int:
"""Consecutive identical tool calls before block (OpenCode processor; 0=off)."""
from butler.utilities.env_parse import int_env
return int(int_env("BUTLER_DOOM_LOOP_THRESHOLD", 3, min=0))
def doom_loop_mode() -> str:
"""``block`` (default, D3) or ``ask`` (Owner /批准一次 via permission cache)."""
mode = os.getenv("BUTLER_DOOM_LOOP_MODE", DOOM_LOOP_MODE_DEFAULT).strip().lower()
return mode if mode in ("block", "ask") else "block"
def doom_loop_soft_nudge_enabled() -> bool:
from butler.utilities.env_parse import env_truthy
return bool(env_truthy("BUTLER_DOOM_LOOP_SOFT_NUDGE", default=True))
def _safe_json_loads(text: str) -> Any:
try:
return json.loads(text)
except (json.JSONDecodeError, TypeError, ValueError):
return None
def classify_tool_failure(tool_name: str, result: str | None) -> tuple[bool, str]:
"""Return (failed, suffix_hint) inferred from serialized tool output."""
if result is None:
return False, ""
from butler.tool_guardrails_ops import mutation_classification_hint
hint = mutation_classification_hint(tool_name, result or "")
if hint is not None:
return cast(tuple[bool, str], hint)
if tool_name in {"run_shell", "terminal"}:
data = _safe_json_loads(result)
if isinstance(data, dict):
if data.get("error"):
return True, " [error]"
exit_code = data.get("exit_code")
if exit_code is not None and exit_code != 0:
return True, f" [exit {exit_code}]"
return False, ""
if tool_name.startswith("git_"):
data = _safe_json_loads(result)
if isinstance(data, dict):
if data.get("error"):
return True, " [error]"
exit_code = data.get("exit_code")
if exit_code is not None and exit_code != 0:
return True, f" [exit {exit_code}]"
return False, ""
data = _safe_json_loads(result)
if isinstance(data, dict):
if data.get("success") is False:
return True, " [error]"
if data.get("error"):
return True, " [error]"
lower = result[:500].lower()
if '"error"' in lower or '"failed"' in lower or result.startswith("Error"):
return True, " [error]"
return False, ""
def _result_hash(result: str | None) -> str:
parsed = _safe_json_loads(result or "")
if parsed is not None:
try:
canonical = json.dumps(parsed, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
except TypeError:
canonical = str(parsed)
else:
canonical = result or ""
return _sha256(canonical)
class ToolCallGuardrailController:
"""Per-turn controller for repeated failed or non-progressing tool calls."""
def __init__(self, config: GuardrailConfig | None = None):
self.config = config or GuardrailConfig()
self._lock = threading.RLock()
self.reset_for_turn()
def reset_for_turn(self) -> None:
from butler.tool_guardrails_ops import reset_tool_loop_detector_safe
reset_tool_loop_detector_safe()
with self._lock:
self._exact_failure_counts: dict[ToolCallSignature, int] = {}
self._same_tool_failure_counts: dict[str, int] = {}
self._no_progress: dict[ToolCallSignature, tuple[str, int]] = {}
self._halt_decision: GuardrailDecision | None = None
self._call_history: deque[ToolCallSignature] = deque(maxlen=16)
self._doom_soft_nudged: set[ToolCallSignature] = set()
@property
def halt_decision(self) -> GuardrailDecision | None:
with self._lock:
return self._halt_decision
def set_halt_decision(self, decision: GuardrailDecision) -> None:
"""Record a halt/block decision from tool execution (thread-safe)."""
with self._lock:
if self._halt_decision is None:
self._halt_decision = decision
def before_call(self, tool_name: str, args: Mapping[str, Any] | None) -> GuardrailDecision:
with self._lock:
return self._before_call_locked(tool_name, args)
def _before_call_locked(
self, tool_name: str, args: Mapping[str, Any] | None
) -> GuardrailDecision:
signature = ToolCallSignature.from_call(tool_name, args)
from butler.tool_guardrails_ops import check_tool_loop_before_call_safe
loop_dec = check_tool_loop_before_call_safe(tool_name, args)
if loop_dec is not None and loop_dec.stuck:
decision = GuardrailDecision(
action="block" if loop_dec.level == "critical" else "warn",
code=loop_dec.detector,
message=loop_dec.message,
tool_name=tool_name,
count=loop_dec.count,
)
if loop_dec.level == "critical":
self._halt_decision = decision
return decision
if not self.config.hard_stop_enabled:
return GuardrailDecision(tool_name=tool_name)
threshold = doom_loop_threshold()
if threshold > 0:
self._call_history.append(signature)
tail = list(self._call_history)[-threshold:]
soft_at = max(1, threshold - 1)
if (
doom_loop_soft_nudge_enabled()
and threshold > 1
and len(tail) >= soft_at
and all(s == tail[0] for s in tail)
and signature not in self._doom_soft_nudged
):
self._doom_soft_nudged.add(signature)
return GuardrailDecision(
action="warn",
code="doom_loop_soft_nudge",
message=(
f"{tool_name} 已连续 {len(tail)} 次相同参数调用"
f"（硬拦截阈值 {threshold}）。请换策略或先向用户确认，"
"避免无效重复。"
),
tool_name=tool_name,
count=len(tail),
)
if len(tail) >= threshold and all(s == tail[0] for s in tail):
mode = doom_loop_mode()
if mode == "ask":
decision = GuardrailDecision(
action="ask",
code="doom_loop",
message=(
f"Doom loop detected for {tool_name} "
f"({threshold} identical calls). "
"Owner: /批准一次 或 /始终允许 doom_loop 后重试。"
),
tool_name=tool_name,
count=threshold,
)
return decision
decision = GuardrailDecision(
action="block",
code="doom_loop",
message=(
f"Blocked {tool_name}: same tool+arguments repeated "
f"{threshold} times in a row. Change strategy or ask the user."
),
tool_name=tool_name,
count=threshold,
)
self._halt_decision = decision
return decision
exact_count = self._exact_failure_counts.get(signature, 0)
if exact_count >= self.config.exact_failure_block_after:
decision = GuardrailDecision(
action="block",
code="repeated_exact_failure_block",
message=(
f"Blocked {tool_name}: same call failed {exact_count} times "
"with identical arguments. Change strategy."
),
tool_name=tool_name,
count=exact_count,
)
self._halt_decision = decision
return decision
if self._is_idempotent(tool_name):
record = self._no_progress.get(signature)
if record is not None:
_rh, repeat_count = record
if repeat_count >= self.config.no_progress_block_after:
decision = GuardrailDecision(
action="block",
code="idempotent_no_progress_block",
message=(
f"Blocked {tool_name}: returned same result {repeat_count} "
"times. Use the result already provided."
),
tool_name=tool_name,
count=repeat_count,
)
self._halt_decision = decision
return decision
return GuardrailDecision(tool_name=tool_name)
def after_call(
self,
tool_name: str,
args: Mapping[str, Any] | None,
result: str | None,
*,
failed: bool | None = None,
) -> GuardrailDecision:
with self._lock:
return self._after_call_locked(tool_name, args, result, failed=failed)
def _after_call_locked(
self,
tool_name: str,
args: Mapping[str, Any] | None,
result: str | None,
*,
failed: bool | None = None,
) -> GuardrailDecision:
args = _coerce_args(args)
signature = ToolCallSignature.from_call(tool_name, args)
if failed is None:
failed, _ = classify_tool_failure(tool_name, result)
from butler.tool_guardrails_ops import record_tool_loop_after_call_safe
record_tool_loop_after_call_safe(
tool_name,
signature.args_hash,
result or "",
)
if failed:
exact_count = self._exact_failure_counts.get(signature, 0) + 1
self._exact_failure_counts[signature] = exact_count
self._no_progress.pop(signature, None)
same_count = self._same_tool_failure_counts.get(tool_name, 0) + 1
self._same_tool_failure_counts[tool_name] = same_count
if self.config.hard_stop_enabled and same_count >= self.config.same_tool_failure_halt_after:
decision = GuardrailDecision(
action="halt",
code="same_tool_failure_halt",
message=(
f"Stopped {tool_name}: failed {same_count} times this turn. "
"Choose a different approach."
),
tool_name=tool_name,
count=same_count,
)
self._halt_decision = decision
return decision
if self.config.warnings_enabled and exact_count >= self.config.exact_failure_warn_after:
return GuardrailDecision(
action="warn",
code="repeated_exact_failure_warning",
message=f"{tool_name} failed {exact_count} times with identical args.",
tool_name=tool_name,
count=exact_count,
)
if self.config.warnings_enabled and same_count >= self.config.same_tool_failure_warn_after:
return GuardrailDecision(
action="warn",
code="same_tool_failure_warning",
message=f"{tool_name} failed {same_count} times this turn.",
tool_name=tool_name,
count=same_count,
)
return GuardrailDecision(tool_name=tool_name, count=exact_count)
self._exact_failure_counts.pop(signature, None)
self._same_tool_failure_counts.pop(tool_name, None)
if not self._is_idempotent(tool_name):
self._no_progress.pop(signature, None)
return GuardrailDecision(tool_name=tool_name)
rh = _result_hash(result)
previous = self._no_progress.get(signature)
repeat_count = 1
if previous is not None and previous[0] == rh:
repeat_count = previous[1] + 1
self._no_progress[signature] = (rh, repeat_count)
if self.config.warnings_enabled and repeat_count >= self.config.no_progress_warn_after:
return GuardrailDecision(
action="warn",
code="idempotent_no_progress_warning",
message=f"{tool_name} returned the same result {repeat_count} times.",
tool_name=tool_name,
count=repeat_count,
)
return GuardrailDecision(tool_name=tool_name, count=repeat_count)
def _is_idempotent(self, tool_name: str) -> bool:
if tool_name in MUTATING_TOOLS:
return False
return tool_name in IDEMPOTENT_TOOLS
def guardrail_metadata(decision: GuardrailDecision) -> dict[str, Any]:
"""Structured guardrail payload for tool result envelopes."""
payload: dict[str, Any] = {
"action": decision.action,
"code": decision.code,
"count": decision.count,
"message": decision.message,
}
if decision.tool_name:
payload["tool_name"] = decision.tool_name
return payload
def synthetic_result(decision: GuardrailDecision) -> str:
"""Build a synthetic tool result JSON for blocked calls."""
return json.dumps({
"error": decision.message,
"guardrail": guardrail_metadata(decision),
}, ensure_ascii=False)
def append_guidance(result: str, decision: GuardrailDecision) -> str:
"""Merge guardrail guidance into a tool result without breaking JSON envelopes."""
if decision.action not in {"warn", "block", "halt"} or not decision.message:
return result
parsed = _safe_json_loads(result or "")
if isinstance(parsed, dict):
payload = dict(parsed)
payload["guardrail"] = guardrail_metadata(decision)
if decision.action in {"block", "halt"}:
payload.setdefault("error", decision.message)
return json.dumps(payload, ensure_ascii=False, default=str)
label = "Tool loop hard stop" if decision.action in {"block", "halt"} else "Tool loop warning"
return (result or "") + f"\n\n[{label}: {decision.code}; count={decision.count}; {decision.message}]"