"""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(), )