debug_helpers.py
1 """Shared debug session infrastructure for Hermes tools. 2 3 Replaces the identical DEBUG_MODE / _log_debug_call / _save_debug_log / 4 get_debug_session_info boilerplate previously duplicated across web_tools, 5 vision_tools, mixture_of_agents_tool, and image_generation_tool. 6 7 Usage in a tool module: 8 9 from tools.debug_helpers import DebugSession 10 11 _debug = DebugSession("web_tools", env_var="WEB_TOOLS_DEBUG") 12 13 # Log a call (no-op when debug mode is off) 14 _debug.log_call("web_search", {"query": q, "results": len(r)}) 15 16 # Save the debug log (no-op when debug mode is off) 17 _debug.save() 18 19 # Expose debug info to external callers 20 def get_debug_session_info(): 21 return _debug.get_session_info() 22 """ 23 24 import datetime 25 import json 26 import logging 27 import os 28 import uuid 29 from typing import Any, Dict 30 31 from hermes_constants import get_hermes_home 32 33 logger = logging.getLogger(__name__) 34 35 36 class DebugSession: 37 """Per-tool debug session that records tool calls to a JSON log file. 38 39 Activated by a tool-specific environment variable (e.g. WEB_TOOLS_DEBUG=true). 40 When disabled, all methods are cheap no-ops. 41 """ 42 43 def __init__(self, tool_name: str, *, env_var: str) -> None: 44 self.tool_name = tool_name 45 self.enabled = os.getenv(env_var, "false").lower() == "true" 46 self.session_id = str(uuid.uuid4()) if self.enabled else "" 47 self.log_dir = get_hermes_home() / "logs" 48 self._calls: list[Dict[str, Any]] = [] 49 self._start_time = datetime.datetime.now().isoformat() if self.enabled else "" 50 51 if self.enabled: 52 self.log_dir.mkdir(parents=True, exist_ok=True) 53 logger.debug("%s debug mode enabled - Session ID: %s", 54 tool_name, self.session_id) 55 56 @property 57 def active(self) -> bool: 58 return self.enabled 59 60 def log_call(self, call_name: str, call_data: Dict[str, Any]) -> None: 61 """Append a tool-call entry to the in-memory log.""" 62 if not self.enabled: 63 return 64 self._calls.append({ 65 "timestamp": datetime.datetime.now().isoformat(), 66 "tool_name": call_name, 67 **call_data, 68 }) 69 70 def save(self) -> None: 71 """Flush the in-memory log to a JSON file in the logs directory.""" 72 if not self.enabled: 73 return 74 try: 75 filename = f"{self.tool_name}_debug_{self.session_id}.json" 76 filepath = self.log_dir / filename 77 payload = { 78 "session_id": self.session_id, 79 "start_time": self._start_time, 80 "end_time": datetime.datetime.now().isoformat(), 81 "debug_enabled": True, 82 "total_calls": len(self._calls), 83 "tool_calls": self._calls, 84 } 85 with open(filepath, "w", encoding="utf-8") as f: 86 json.dump(payload, f, indent=2, ensure_ascii=False) 87 logger.debug("%s debug log saved: %s", self.tool_name, filepath) 88 except Exception as e: 89 logger.error("Error saving %s debug log: %s", self.tool_name, e) 90 91 def get_session_info(self) -> Dict[str, Any]: 92 """Return a summary dict suitable for returning from get_debug_session_info().""" 93 if not self.enabled: 94 return { 95 "enabled": False, 96 "session_id": None, 97 "log_path": None, 98 "total_calls": 0, 99 } 100 return { 101 "enabled": True, 102 "session_id": self.session_id, 103 "log_path": str(self.log_dir / f"{self.tool_name}_debug_{self.session_id}.json"), 104 "total_calls": len(self._calls), 105 }