butler/tools/tool_audit.py
Rama main · réplica de lectura del repositorio original.
"""Tool audit recording, result finalization, and observation tracking."""
from __future__ import annotations
import json
import logging
import os
import threading
import time
from collections import deque
from typing import Any
from butler.contracts.tool_registry_registry import get_tool_registry_read
from butler.core.session_transcript import transcript_path
from butler.tools.tool_audit_ops import get_audit_session_key_safe
from butler.tools.tool_audit_ops import run_observation_side_effects_safe
from butler.execution_context import get_current_session_key
from butler.tools.tool_audit_ops import enqueue_tool_observation_safe, record_tool_observation_safe, resolve_observation_workspace_safe
from butler.execution_context import get_current_orchestrator
from butler.tools.tool_audit_ops import run_permission_denied_hooks_safe
from butler.defaults.env_defaults import TOOL_AUDIT_PERSIST_DEFAULT
logger = logging.getLogger(__name__)
_TOOL_AUDIT_EVENTS: deque[dict[str, Any]] = deque(maxlen=200)
_TOOL_AUDIT_EVENTS_BY_SESSION: dict[str, deque[dict[str, Any]]] = {}
_TOOL_AUDIT_LOCK = threading.RLock()
_MAX_SESSION_BUCKETS = 512
def _evict_oldest_session_bucket_locked() -> None:
while len(_TOOL_AUDIT_EVENTS_BY_SESSION) >= _MAX_SESSION_BUCKETS:
oldest = next(iter(_TOOL_AUDIT_EVENTS_BY_SESSION))
_TOOL_AUDIT_EVENTS_BY_SESSION.pop(oldest, None)
def _parse_json_object(text: str) -> dict[str, Any] | None:
try:
parsed = json.loads(text)
except (TypeError, json.JSONDecodeError):
return None
return parsed if isinstance(parsed, dict) else None
def _tool_result_ok(payload: dict[str, Any]) -> bool:
if payload.get("ok") is False:
return False
if "error" in payload:
return False
if payload.get("success") is False:
return False
exit_code = payload.get("exit_code")
if isinstance(exit_code, int) and exit_code != 0:
return False
return True
def _tool_result_code(name: str, payload: dict[str, Any], *, ok: bool) -> str:
if ok:
return "TOOL_OK"
guardrail = payload.get("guardrail")
if isinstance(guardrail, dict):
action = guardrail.get("action")
if action == "halt":
return "TOOL_GUARDRAIL_HALT"
if action == "block":
return "TOOL_GUARDRAIL_BLOCKED"
if payload.get("code"):
return str(payload["code"])
error = str(payload.get("error") or "")
if error == "interrupted":
return "TOOL_INTERRUPTED"
if isinstance(payload.get("exit_code"), int) and payload["exit_code"] != 0:
return "TOOL_EXIT_NONZERO"
read_port = get_tool_registry_read()
if read_port is not None and not read_port.is_tool_registered(name):
return "TOOL_NOT_FOUND"
lowered = error.lower()
if (
"access denied" in lowered
or "outside workspace" in lowered
or "路径在工作区外" in error
or "sensitive" in lowered
or "symlink" in lowered
or "hardlinked" in lowered
):
return "TOOL_SECURITY_DENIED"
if lowered.startswith("file not found") or lowered.startswith("unknown tool"):
return "TOOL_NOT_FOUND"
if "timeout" in lowered or "timed out" in lowered:
return "TOOL_TIMEOUT"
if "too large" in lowered or "limit" in lowered:
return "TOOL_RESOURCE_LIMIT"
return "TOOL_ERROR"
def _tool_audit_persist_enabled() -> bool:
return os.getenv("BUTLER_TOOL_AUDIT_PERSIST", TOOL_AUDIT_PERSIST_DEFAULT).strip().lower() in (
"1",
"true",
"yes",
"on",
)
def _persist_tool_audit_event(event: dict[str, Any]) -> None:
if not _tool_audit_persist_enabled():
return
sk = str(event.get("session_key") or "").strip()
if not sk:
return
try:
path = transcript_path(sk).parent / "tool_audit.jsonl"
path.parent.mkdir(parents=True, exist_ok=True)
import json
line = json.dumps(event, ensure_ascii=False) + "\n"
with open(path, "a", encoding="utf-8") as fh:
fh.write(line)
except OSError as exc:
logger.debug("tool audit persist skipped: %s", exc)
def _record_tool_audit(
name: str,
args: dict[str, Any],
*,
ok: bool,
code: str,
started_at: float,
) -> None:
session_key = get_audit_session_key_safe()
event = {
"tool": name,
"ok": ok,
"code": code,
"session_key": session_key or "",
"elapsed_ms": round((time.monotonic() - started_at) * 1000, 2),
"arg_keys": sorted(str(key) for key in (args or {}).keys()),
}
with _TOOL_AUDIT_LOCK:
_TOOL_AUDIT_EVENTS.append(event)
sk = str(event.get("session_key") or "")
bucket = _TOOL_AUDIT_EVENTS_BY_SESSION.get(sk)
if bucket is None:
_evict_oldest_session_bucket_locked()
bucket = deque(maxlen=200)
_TOOL_AUDIT_EVENTS_BY_SESSION[sk] = bucket
bucket.append(event)
_persist_tool_audit_event(event)
def get_tool_audit_events(
limit: int | None = None,
*,
session_key: str | None = None,
) -> list[dict[str, Any]]:
"""Return recent tool audit events with redacted arguments."""
with _TOOL_AUDIT_LOCK:
if session_key is None:
events = list(_TOOL_AUDIT_EVENTS)
else:
events = list(_TOOL_AUDIT_EVENTS_BY_SESSION.get(session_key, ()))
if limit is not None:
events = events[-max(0, int(limit)):]
return [dict(event) for event in events]
def reset_tool_audit_events(session_key: str | None = None) -> None:
"""Clear in-memory tool audit events. Intended for tests and diagnostics reset."""
with _TOOL_AUDIT_LOCK:
if session_key is None:
_TOOL_AUDIT_EVENTS.clear()
_TOOL_AUDIT_EVENTS_BY_SESSION.clear()
return
_TOOL_AUDIT_EVENTS_BY_SESSION.pop(session_key, None)
retained = [
event for event in _TOOL_AUDIT_EVENTS
if event.get("session_key") != session_key
]
_TOOL_AUDIT_EVENTS.clear()
maxlen = _TOOL_AUDIT_EVENTS.maxlen or 200
_TOOL_AUDIT_EVENTS.extend(retained[-maxlen:])
def pop_last_tool_audit_for_tool(name: str) -> None:
"""Remove the latest audit event for a tool when replacing it with guardrail halt."""
with _TOOL_AUDIT_LOCK:
if not _TOOL_AUDIT_EVENTS or _TOOL_AUDIT_EVENTS[-1].get("tool") != name:
return
last = _TOOL_AUDIT_EVENTS.pop()
session_key = str(last.get("session_key") or "")
bucket = _TOOL_AUDIT_EVENTS_BY_SESSION.get(session_key)
if bucket and bucket[-1].get("tool") == name:
bucket.pop()
def _maybe_record_tool_observation(
name: str,
args: dict[str, Any],
payload: dict[str, Any],
) -> None:
def _run() -> None:
_record_tool_observation_side_effects(name, args, payload)
run_observation_side_effects_safe(_run)
def _record_tool_observation_side_effects(
name: str,
args: dict[str, Any],
payload: dict[str, Any],
) -> None:
sk = str(get_current_session_key() or "").strip()
if not sk:
return
preview = ""
if payload.get("error"):
preview = str(payload.get("error") or "")[:200]
elif payload.get("mode") == "summary":
preview = f"summary lines={payload.get('total_lines', '?')}"
elif payload.get("preview"):
preview = str(payload.get("preview") or "")[:200]
else:
for key in ("content", "result", "output", "message"):
if payload.get(key):
preview = str(payload[key])[:200]
break
if not preview and name == "read_file":
preview = str(args.get("path") or "")[:120]
record_tool_observation_safe(
sk,
tool=name,
ok=_tool_result_ok(payload),
preview=preview,
)
path_hint = str(args.get("path") or args.get("file") or "")
workspace = None
orch = get_current_orchestrator()
if orch is not None:
workspace = resolve_observation_workspace_safe(orch, sk)
enqueue_tool_observation_safe(
session_key=sk,
tool=name,
ok=_tool_result_ok(payload),
preview=preview,
path=path_hint,
workspace=workspace,
)
def _finalize_tool_result(
name: str,
args: dict[str, Any],
result: Any,
*,
started_at: float,
) -> str:
if isinstance(result, str):
payload = _parse_json_object(result)
if payload is None:
ok = True
code = "TOOL_OK"
_record_tool_audit(name, args, ok=ok, code=code, started_at=started_at)
parsed = _parse_json_object(result)
if isinstance(parsed, dict):
_maybe_record_tool_observation(name, args, parsed)
elif name:
_maybe_record_tool_observation(
name,
args,
{"preview": str(result)[:200]},
)
return result
elif isinstance(result, dict):
payload = dict(result)
else:
payload = result
if isinstance(payload, dict):
ok = _tool_result_ok(payload)
code = _tool_result_code(name, payload, ok=ok)
if not ok:
payload.setdefault("ok", False)
payload.setdefault("tool", name)
payload.setdefault("code", code)
err = str(payload.get("error") or "")
if err.startswith("Access denied"):
hint = run_permission_denied_hooks_safe(name, args, err)
if hint:
payload["permission_denied_hint"] = hint
_record_tool_audit(name, args, ok=ok, code=code, started_at=started_at)
_maybe_record_tool_observation(name, args, payload)
return json.dumps(payload, ensure_ascii=False, default=str)
_record_tool_audit(name, args, ok=True, code="TOOL_OK", started_at=started_at)
if isinstance(payload, dict):
_maybe_record_tool_observation(name, args, payload)
return json.dumps(payload, ensure_ascii=False, default=str)
def finalize_tool_result(
name: str,
args: dict[str, Any],
result: Any,
*,
started_at: float | None = None,
) -> str:
"""Apply Butler's tool result envelope and audit record to fallback paths."""
return _finalize_tool_result(
name,
args,
result,
started_at=started_at if started_at is not None else time.monotonic(),
)