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

1#!/usr/bin/env python3 

2""" 

3Agent Management System for Osiris CLI 

4Handles agent selection, creation, and persistence (Letta-inspired) 

5""" 

6 

7import json 

8from pathlib import Path 

9from typing import Optional, List, Dict 

10from datetime import datetime 

11from rich.console import Console 

12from rich.table import Table 

13from rich.panel import Panel 

14from rich.prompt import Prompt, Confirm 

15 

16console = Console() 

17 

18 

19class AgentManager: 

20 """ 

21 Manages multiple agents with persistent memory (Letta-style) 

22 """ 

23 

24 def __init__(self): 

25 self.agents_dir = Path.home() / ".osiris" / "agents" 

26 self.agents_dir.mkdir(parents=True, exist_ok=True) 

27 

28 def list_agents(self) -> List[Dict]: 

29 """List all available agents""" 

30 agents = [] 

31 

32 for agent_file in self.agents_dir.glob("*.json"): 

33 try: 

34 with open(agent_file, 'r') as f: 

35 data = json.load(f) 

36 

37 # Calculate last used 

38 last_used = data.get("last_used", "never") 

39 last_used_dt = None 

40 if last_used != "never": 

41 try: 

42 dt = datetime.fromisoformat(last_used) 

43 last_used_dt = dt 

44 now = datetime.now() 

45 delta = now - dt 

46 

47 if delta.seconds < 60: 

48 last_used = "just now" 

49 elif delta.seconds < 3600: 

50 last_used = f"{delta.seconds // 60} min ago" 

51 elif delta.days == 0: 

52 last_used = f"{delta.seconds // 3600} hours ago" 

53 elif delta.days == 1: 

54 last_used = "yesterday" 

55 else: 

56 last_used = f"{delta.days} days ago" 

57 except: 

58 pass 

59 

60 agents.append({ 

61 "name": agent_file.stem, 

62 "persona": data.get("persona", "General assistant"), 

63 "skills": data.get("skills", []), 

64 "facts": data.get("core_memory", {}), 

65 "last_used": last_used, 

66 "last_used_dt": last_used_dt, 

67 "file": agent_file 

68 }) 

69 except: 

70 continue 

71 

72 # Sort by last used 

73 agents.sort(key=lambda x: x.get("last_used_dt") or datetime.min, reverse=True) 

74 return agents 

75 

76 def select_agent(self) -> Optional[str]: 

77 """Interactive agent selector (Letta-style)""" 

78 agents = self.list_agents() 

79 

80 console.print("\n[bold cyan]Select Agent[/bold cyan]\n") 

81 

82 # Create table 

83 table = Table(show_header=True, header_style="bold cyan") 

84 table.add_column("#", style="dim", width=3) 

85 table.add_column("Name", style="cyan") 

86 table.add_column("Persona", style="dim") 

87 table.add_column("Skills", style="green") 

88 table.add_column("Last Used", style="yellow") 

89 

90 for i, agent in enumerate(agents, 1): 

91 skills_str = ", ".join(agent["skills"][:2]) 

92 if len(agent["skills"]) > 2: 

93 skills_str += f" +{len(agent['skills']) - 2}" 

94 

95 table.add_row( 

96 str(i), 

97 agent["name"], 

98 agent["persona"][:30] + "..." if len(agent["persona"]) > 30 else agent["persona"], 

99 skills_str or "none", 

100 agent["last_used"] 

101 ) 

102 

103 # Add "create new" option 

104 table.add_row( 

105 str(len(agents) + 1), 

106 "[cyan]+ Create new agent[/cyan]", 

107 "", 

108 "", 

109 "" 

110 ) 

111 

112 console.print(table) 

113 console.print() 

114 

115 # Get selection 

116 try: 

117 choice = Prompt.ask( 

118 "Choose agent", 

119 default="1" if agents else str(len(agents) + 1) 

120 ) 

121 

122 if choice.isdigit(): 

123 idx = int(choice) - 1 

124 if 0 <= idx < len(agents): 

125 return agents[idx]["name"] 

126 elif idx == len(agents): 

127 return self.create_agent() 

128 except (ValueError, KeyboardInterrupt): 

129 pass 

130 

131 return None 

132 

133 def create_agent(self) -> Optional[str]: 

134 """Create a new agent interactively""" 

135 console.print("\n[bold cyan]Create New Agent[/bold cyan]\n") 

136 

137 try: 

138 # Get agent name 

139 name = Prompt.ask("Agent name", default="assistant") 

140 

141 # Get persona 

142 console.print("\n[dim]Choose a persona:[/dim]") 

143 personas = { 

144 "1": ("General Assistant", "A helpful general-purpose AI assistant"), 

145 "2": ("Code Expert", "Senior software engineer specializing in code review and debugging"), 

146 "3": ("DevOps Engineer", "Infrastructure and deployment specialist"), 

147 "4": ("Data Analyst", "Data science and analytics expert"), 

148 "5": ("Custom", "Define your own persona") 

149 } 

150 

151 for key, (title, desc) in personas.items(): 

152 console.print(f" {key}. [cyan]{title}[/cyan] - [dim]{desc}[/dim]") 

153 

154 persona_choice = Prompt.ask("\nChoose persona", default="1") 

155 

156 if persona_choice in personas: 

157 if persona_choice == "5": 

158 persona = Prompt.ask("Enter custom persona") 

159 else: 

160 persona = personas[persona_choice][1] 

161 else: 

162 persona = personas["1"][1] 

163 

164 # Get skills 

165 console.print("\n[dim]Add skills (comma-separated, or press Enter to skip):[/dim]") 

166 skills_input = Prompt.ask("Skills", default="") 

167 skills = [s.strip() for s in skills_input.split(",") if s.strip()] 

168 

169 # Create agent data 

170 agent_data = { 

171 "name": name, 

172 "persona": persona, 

173 "skills": skills, 

174 "core_memory": {}, 

175 "created": datetime.now().isoformat(), 

176 "last_used": datetime.now().isoformat() 

177 } 

178 

179 # Save agent 

180 agent_file = self.agents_dir / f"{name}.json" 

181 with open(agent_file, 'w') as f: 

182 json.dump(agent_data, f, indent=2) 

183 

184 console.print(f"\n[green]✓[/green] Created agent: {name}\n") 

185 return name 

186 

187 except KeyboardInterrupt: 

188 console.print("\n[yellow]Cancelled[/yellow]\n") 

189 return None 

190 

191 def update_last_used(self, agent_name: str): 

192 """Update agent's last used timestamp""" 

193 agent_file = self.agents_dir / f"{agent_name}.json" 

194 if agent_file.exists(): 

195 try: 

196 with open(agent_file, 'r') as f: 

197 data = json.load(f) 

198 

199 data["last_used"] = datetime.now().isoformat() 

200 

201 with open(agent_file, 'w') as f: 

202 json.dump(data, f, indent=2) 

203 except: 

204 pass 

205 

206 def get_agent_data(self, agent_name: str) -> Dict: 

207 """Get agent data""" 

208 agent_file = self.agents_dir / f"{agent_name}.json" 

209 if agent_file.exists(): 

210 try: 

211 with open(agent_file, 'r') as f: 

212 return json.load(f) 

213 except: 

214 pass 

215 

216 # Return default agent data 

217 return { 

218 "name": agent_name, 

219 "persona": "A helpful AI assistant", 

220 "skills": [], 

221 "core_memory": {}, 

222 "created": datetime.now().isoformat(), 

223 "last_used": datetime.now().isoformat() 

224 } 

225 

226 def save_agent_data(self, agent_name: str, data: Dict): 

227 """Save agent data""" 

228 agent_file = self.agents_dir / f"{agent_name}.json" 

229 with open(agent_file, 'w') as f: 

230 json.dump(data, f, indent=2) 

231 

232 

233# Global instance 

234agent_manager = AgentManager()