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

1#!/usr/bin/env python3 

2""" 

3Enhanced Display System for Osiris CLI v5.1 

4Blends Gemini CLI's clean streaming with Letta Code's structured output 

5""" 

6 

7import asyncio 

8from typing import Optional, AsyncIterator 

9from rich.console import Console 

10from rich.live import Live 

11from rich.spinner import Spinner 

12from rich.panel import Panel 

13from rich.text import Text 

14from rich.table import Table 

15from rich.markdown import Markdown 

16import time 

17 

18console = Console() 

19 

20 

21class StreamingDisplay: 

22 """ 

23 Handles real-time streaming display like Gemini CLI 

24 """ 

25 

26 def __init__(self): 

27 self.current_line = "" 

28 self.buffer = [] 

29 

30 async def stream_text(self, text: str, delay: float = 0.01): 

31 """Stream text character-by-character""" 

32 for char in text: 

33 console.print(char, end="", markup=False) 

34 await asyncio.sleep(delay) 

35 console.print() # Newline at end 

36 

37 async def stream_markdown(self, text: str, delay: float = 0.01): 

38 """Stream markdown with syntax highlighting""" 

39 # For now, just stream as text 

40 # TODO: Implement incremental markdown rendering 

41 await self.stream_text(text, delay) 

42 

43 

44class ToolDisplay: 

45 """ 

46 Clean tool execution display (Gemini-inspired) 

47 """ 

48 

49 @staticmethod 

50 def show_tool_call(tool_name: str, args: dict, compact: bool = True): 

51 """Display tool being called""" 

52 if compact: 

53 # Compact format (Gemini-style) 

54 args_str = ", ".join(f"{k}={repr(v)[:50]}" for k, v in args.items()) 

55 console.print(f"\n[cyan]→[/cyan] {tool_name}") 

56 if args_str: 

57 console.print(f" [dim]{args_str}[/dim]") 

58 else: 

59 # Detailed format 

60 console.print(Panel( 

61 f"[bold]{tool_name}[/bold]\n" + 

62 "\n".join(f"{k}: {v}" for k, v in args.items()), 

63 border_style="cyan", 

64 padding=(0, 1) 

65 )) 

66 

67 @staticmethod 

68 def show_tool_result(result: str, success: bool = True, duration: Optional[float] = None): 

69 """Display tool result""" 

70 # Truncate long results 

71 display_result = result[:200] + ("..." if len(result) > 200 else "") 

72 

73 if success: 

74 status = "[green]✓[/green]" 

75 if duration: 

76 status += f" [dim]{duration:.2f}s[/dim]" 

77 else: 

78 status = "[red]✗[/red]" 

79 

80 console.print(f" {status} [dim]{display_result}[/dim]") 

81 

82 @staticmethod 

83 def show_tool_progress(message: str, spinner_type: str = "dots"): 

84 """Show progress spinner""" 

85 return Spinner(spinner_type, text=message) 

86 

87 

88class ThoughtDisplay: 

89 """ 

90 Display model's reasoning process (Gemini-style) 

91 """ 

92 

93 @staticmethod 

94 def show_thought(thought: str): 

95 """Display a thought bubble""" 

96 console.print(f"\n[dim]💭 {thought}[/dim]") 

97 

98 @staticmethod 

99 def show_planning(steps: list): 

100 """Display planning steps""" 

101 console.print("\n[bold]Planning:[/bold]") 

102 for i, step in enumerate(steps, 1): 

103 console.print(f" {i}. [dim]{step}[/dim]") 

104 console.print() 

105 

106 

107class MemoryDisplay: 

108 """ 

109 Display agent memory (Letta-inspired) 

110 """ 

111 

112 @staticmethod 

113 def show_memory_card(agent_name: str, memory: dict): 

114 """Display memory as a card""" 

115 table = Table(show_header=False, box=None, padding=(0, 1)) 

116 table.add_column(style="cyan") 

117 table.add_column() 

118 

119 table.add_row("Agent:", agent_name) 

120 

121 if "persona" in memory: 

122 persona = memory["persona"][:50] + "..." if len(memory["persona"]) > 50 else memory["persona"] 

123 table.add_row("Persona:", persona) 

124 

125 if "skills" in memory: 

126 skills = ", ".join(memory["skills"][:3]) 

127 if len(memory["skills"]) > 3: 

128 skills += f" +{len(memory['skills']) - 3} more" 

129 table.add_row("Skills:", skills) 

130 

131 if "facts" in memory: 

132 table.add_row("Facts:", f"{len(memory['facts'])} stored") 

133 

134 console.print(Panel( 

135 table, 

136 title="[bold]Agent Memory[/bold]", 

137 border_style="blue", 

138 padding=(0, 1) 

139 )) 

140 

141 @staticmethod 

142 def show_memory_update(key: str, value: str, operation: str = "add"): 

143 """Show memory being updated""" 

144 if operation == "add": 

145 console.print(f"[dim]💾 Remembered: {key} = {value[:50]}...[/dim]") 

146 elif operation == "remove": 

147 console.print(f"[dim]🗑️ Forgot: {key}[/dim]") 

148 

149 

150class SessionDisplay: 

151 """ 

152 Session summary display (Gemini-inspired) 

153 """ 

154 

155 @staticmethod 

156 def show_summary( 

157 duration: str, 

158 tools_used: int, 

159 files_modified: int, 

160 key_actions: list 

161 ): 

162 """Display session summary""" 

163 console.print("\n" + "─" * 50) 

164 console.print("[bold]Session Summary[/bold]") 

165 console.print("─" * 50) 

166 console.print(f"Duration: {duration}") 

167 console.print(f"Tools used: {tools_used}") 

168 console.print(f"Files modified: {files_modified}") 

169 

170 if key_actions: 

171 console.print("\nKey actions:") 

172 for action in key_actions: 

173 console.print(f" • {action}") 

174 

175 console.print() 

176 

177 

178class AgentSelector: 

179 """ 

180 Agent selection interface (Letta-inspired) 

181 """ 

182 

183 @staticmethod 

184 def show_selector(agents: list) -> Optional[str]: 

185 """Display agent selector and return selected agent""" 

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

187 

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

189 name = agent["name"] 

190 last_used = agent.get("last_used", "never") 

191 console.print(f" {i}. {name} [dim](last used {last_used})[/dim]") 

192 

193 console.print(f" {len(agents) + 1}. [cyan]+ Create new agent[/cyan]") 

194 

195 try: 

196 choice = input("\n> ").strip() 

197 if choice.isdigit(): 

198 idx = int(choice) - 1 

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

200 return agents[idx]["name"] 

201 elif idx == len(agents): 

202 return "__new__" 

203 except (ValueError, KeyboardInterrupt): 

204 pass 

205 

206 return None 

207 

208 

209class ProgressTracker: 

210 """ 

211 Track and display progress for long-running operations 

212 """ 

213 

214 def __init__(self, total: int, description: str = "Processing"): 

215 self.total = total 

216 self.current = 0 

217 self.description = description 

218 self.start_time = time.time() 

219 

220 def update(self, increment: int = 1): 

221 """Update progress""" 

222 self.current += increment 

223 elapsed = time.time() - self.start_time 

224 

225 if self.total > 0: 

226 percent = (self.current / self.total) * 100 

227 console.print( 

228 f"\r{self.description}... [{self.current}/{self.total}] " 

229 f"{percent:.0f}% ({elapsed:.1f}s)", 

230 end="" 

231 ) 

232 else: 

233 console.print( 

234 f"\r{self.description}... {self.current} items ({elapsed:.1f}s)", 

235 end="" 

236 ) 

237 

238 def complete(self): 

239 """Mark as complete""" 

240 elapsed = time.time() - self.start_time 

241 console.print( 

242 f"\r{self.description}... [green]✓[/green] " 

243 f"({self.current} items, {elapsed:.1f}s)" 

244 ) 

245 

246 

247# Convenience functions 

248def stream_text(text: str): 

249 """Quick function to stream text""" 

250 display = StreamingDisplay() 

251 asyncio.run(display.stream_text(text)) 

252 

253 

254def show_tool(name: str, args: dict, result: str = None, success: bool = True): 

255 """Quick function to show tool execution""" 

256 ToolDisplay.show_tool_call(name, args) 

257 if result: 

258 ToolDisplay.show_tool_result(result, success) 

259 

260 

261def show_thought(thought: str): 

262 """Quick function to show thought""" 

263 ThoughtDisplay.show_thought(thought) 

264 

265 

266def show_memory(agent_name: str, memory: dict): 

267 """Quick function to show memory""" 

268 MemoryDisplay.show_memory_card(agent_name, memory)