Coverage for src / osiris_cli / activity_monitor.py: 0%

1#!/usr/bin/env python3 

2""" 

3Tool Activity Monitor for Osiris CLI 

4Shows real-time tool usage, file operations, and agent activity 

5""" 

6 

7from rich.console import Console 

8from rich.panel import Panel 

9from rich.table import Table 

10from rich.live import Live 

11from datetime import datetime 

12from typing import Dict, List, Optional 

13import time 

14 

15console = Console() 

16 

17 

18class ActivityMonitor: 

19 """Monitor and display agent tool activity in real-time""" 

20 

21 def __init__(self): 

22 self.activities: List[Dict] = [] 

23 self.current_tool: Optional[str] = None 

24 self.start_time: Optional[float] = None 

25 

26 def start_tool(self, tool_name: str, args: Dict): 

27 """Record tool execution start""" 

28 self.current_tool = tool_name 

29 self.start_time = time.time() 

30 

31 activity = { 

32 "tool": tool_name, 

33 "args": args, 

34 "start_time": datetime.now(), 

35 "status": "running" 

36 } 

37 self.activities.append(activity) 

38 

39 # Display tool start 

40 self._display_tool_start(tool_name, args) 

41 

42 def end_tool(self, result: str, success: bool = True): 

43 """Record tool execution end""" 

44 if not self.current_tool: 

45 return 

46 

47 elapsed = time.time() - self.start_time if self.start_time else 0 

48 

49 # Update last activity 

50 if self.activities: 

51 self.activities[-1].update({ 

52 "status": "success" if success else "failed", 

53 "result": result, 

54 "elapsed": elapsed 

55 }) 

56 

57 # Display tool end 

58 self._display_tool_end(self.current_tool, result, elapsed, success) 

59 

60 self.current_tool = None 

61 self.start_time = None 

62 

63 def _display_tool_start(self, tool_name: str, args: Dict): 

64 """Display tool execution start""" 

65 # Disabled - display is handled in minimal_cli.py 

66 pass 

67 

68 def _display_tool_end(self, tool_name: str, result: str, elapsed: float, success: bool): 

69 """Display tool execution end""" 

70 # Disabled - display is handled in minimal_cli.py 

71 pass 

72 

73 

74 def show_summary(self): 

75 """Show summary of all activities""" 

76 if not self.activities: 

77 return 

78 

79 console.print("\n[bold]Activity Summary:[/bold]") 

80 

81 table = Table(show_header=True, box=None, padding=(0, 1)) 

82 table.add_column("Tool", style="cyan") 

83 table.add_column("Status") 

84 table.add_column("Time", style="dim") 

85 table.add_column("Result", style="dim") 

86 

87 for activity in self.activities: 

88 status_icon = "✓" if activity["status"] == "success" else "✗" 

89 status_color = "green" if activity["status"] == "success" else "red" 

90 elapsed = activity.get("elapsed", 0) 

91 result = activity.get("result", "")[:50] 

92 

93 table.add_row( 

94 activity["tool"], 

95 f"[{status_color}]{status_icon}[/{status_color}]", 

96 f"{elapsed:.2f}s", 

97 result 

98 ) 

99 

100 console.print(table) 

101 console.print() 

102 

103 

104# Global monitor instance 

105activity_monitor = ActivityMonitor() 

106 

107 

108def monitor_tool_call(func): 

109 """Decorator to monitor tool calls""" 

110 def wrapper(tool_name: str, *args, **kwargs): 

111 # Start monitoring 

112 activity_monitor.start_tool(tool_name, kwargs) 

113 

114 try: 

115 # Execute tool 

116 result = func(tool_name, *args, **kwargs) 

117 from .tools import normalize_tool_result 

118 

119 normalized = normalize_tool_result(tool_name, result) 

120 summary = ( 

121 normalized.get("message") 

122 or normalized.get("stdout") 

123 or normalized.get("stderr") 

124 or "" 

125 ) 

126 success = normalized.get("status") == "success" 

127 activity_monitor.end_tool(summary, success=success) 

128 

129 return result 

130 except Exception as e: 

131 # End monitoring (failure) 

132 activity_monitor.end_tool(str(e), success=False) 

133 raise 

134 

135 return wrapper