Coverage for src / osiris_cli / letta_formatter.py: 0%
148 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-27 17:41 +0200
1"""
2Letta Visual Standard Formatter
4Implements the exact visual style preferred by the user:
5- Action-Result indentation with ● and ⎿ glyphs
6- Thinking blocks with ✻ symbol
7- Persistent bottom status bar
8- No walls of text - always summarize
9"""
11import os
12import sys
13from typing import Optional, List, Dict, Any
14from datetime import datetime
15from pathlib import Path
17from rich.console import Console
18from rich.text import Text
19from rich.panel import Panel
20from rich.layout import Layout
21from rich.live import Live
22from rich import box
24from .logger import get_logger
25from .config import settings
27logger = get_logger()
28console = Console()
31class LettaGlyphs:
32 """Official glyphs for Letta Visual Standard"""
33 ACTION = "●" # Circle for actions/tool calls
34 RESULT = "⎿" # Branch for results
35 THINKING = "✻" # Asterisk for thinking/reasoning
36 SUCCESS = "✓" # Checkmark for success
37 ERROR = "✗" # X for errors
38 WARNING = "⚠" # Warning triangle
39 ARROW = "→" # Arrow for flow
40 BULLET = "•" # Bullet for lists
43class LettaFormatter:
44 """
45 Letta Visual Standard Formatter.
47 Formats all CLI output according to Letta Code aesthetic:
48 - Clean action-result trees
49 - Summarized outputs (no walls of text)
50 - Thinking blocks
51 - Persistent status bar
52 """
54 def __init__(self):
55 self.status_mode = "standard"
56 self.status_version = "v4.0.0"
57 self.status_context = "0 files"
58 self.live: Optional[Live] = None
59 self.messages: List[str] = []
60 self.indent_level = 0
62 def format_action(
63 self,
64 tool_name: str,
65 args: Dict[str, Any],
66 indent: int = 0
67 ) -> str:
68 """
69 Format tool action in Letta style.
71 Example:
72 ● db_connect(connection_id="mydb", db_type="postgresql")
73 """
74 # Format arguments
75 arg_parts = []
76 for key, value in args.items():
77 if isinstance(value, str):
78 arg_parts.append(f'{key}="{value}"')
79 else:
80 arg_parts.append(f'{key}={value}')
82 args_str = ", ".join(arg_parts) if arg_parts else ""
84 indent_str = " " * indent
85 return f"{indent_str}{LettaGlyphs.ACTION} {tool_name}({args_str})"
87 def format_result(
88 self,
89 summary: str,
90 indent: int = 1,
91 success: bool = True
92 ) -> str:
93 """
94 Format tool result in Letta style.
96 Example:
97 ⎿ Successfully connected to postgresql database
98 """
99 indent_str = " " * indent
100 status_glyph = LettaGlyphs.SUCCESS if success else LettaGlyphs.ERROR
101 return f"{indent_str}{LettaGlyphs.RESULT} {summary} {status_glyph}"
103 def format_thinking(self, reasoning: str) -> str:
104 """
105 Format thinking block in Letta style.
107 Example:
108 ✻ Thinking...
109 I need to connect to the database first, then query the users table.
110 """
111 lines = reasoning.strip().split('\n')
112 formatted = [f"{LettaGlyphs.THINKING} Thinking..."]
113 for line in lines:
114 formatted.append(f" {line}")
115 return '\n'.join(formatted)
117 def format_status_bar(self) -> str:
118 """
119 Format persistent bottom status bar.
121 Example:
122 >> yolo (allow all) mode | Osiris v4.0.0 | 3 files loaded
123 """
124 mode_display = f"{self.status_mode}"
125 if settings.yolo_mode:
126 mode_display += " (allow all)"
128 return f">> {mode_display} mode | Osiris {self.status_version} | {self.status_context}"
130 def summarize_file_content(
131 self,
132 file_path: str,
133 content: str,
134 max_lines: int = 3
135 ) -> str:
136 """
137 Summarize file content instead of dumping it.
139 Example:
140 ⎿ Read 500 lines from config.py
141 """
142 lines = content.split('\n')
143 line_count = len(lines)
144 size_kb = len(content) / 1024
146 return f"Read {line_count} lines from {Path(file_path).name} ({size_kb:.1f}KB)"
148 def summarize_query_results(
149 self,
150 result: Dict[str, Any]
151 ) -> str:
152 """
153 Summarize database query results.
155 Example:
156 ⎿ Found 150 rows in users table ✓
157 """
158 if "rows" in result:
159 row_count = len(result["rows"])
160 columns = result.get("columns", [])
161 return f"Found {row_count} rows with {len(columns)} columns"
162 elif "documents" in result:
163 doc_count = len(result["documents"])
164 return f"Found {doc_count} documents"
165 elif "rows_affected" in result:
166 affected = result["rows_affected"]
167 return f"Modified {affected} rows"
168 else:
169 return "Query completed successfully"
171 def summarize_schema(
172 self,
173 schema: Dict[str, Any]
174 ) -> str:
175 """
176 Summarize database schema.
178 Example:
179 ⎿ Schema: 12 tables, 85 columns total ✓
180 """
181 table_count = schema.get("table_count", 0)
183 # Count total columns
184 total_columns = 0
185 tables = schema.get("tables", {})
186 for table_name, columns in tables.items():
187 total_columns += len(columns)
189 return f"Schema: {table_count} tables, {total_columns} columns total"
191 def print_action_result(
192 self,
193 tool_name: str,
194 args: Dict[str, Any],
195 result_summary: str,
196 success: bool = True
197 ):
198 """
199 Print complete action-result pair.
201 Example:
202 ● db_connect(connection_id="mydb", db_type="postgresql")
203 ⎿ Successfully connected to postgresql database ✓
204 """
205 # Action line
206 action_line = self.format_action(tool_name, args, indent=0)
207 console.print(action_line, style="cyan")
209 # Result line
210 result_line = self.format_result(result_summary, indent=1, success=success)
211 result_style = "green" if success else "red"
212 console.print(result_line, style=result_style)
214 # Add to message history
215 self.messages.append(action_line)
216 self.messages.append(result_line)
218 def print_thinking(self, reasoning: str):
219 """
220 Print thinking block.
222 Example:
223 ✻ Thinking...
224 I need to query the database to get the user count.
225 """
226 thinking_block = self.format_thinking(reasoning)
227 console.print(thinking_block, style="dim cyan")
229 self.messages.append(thinking_block)
231 def print_error(
232 self,
233 tool_name: str,
234 args: Dict[str, Any],
235 error: str
236 ):
237 """
238 Print error in Letta style.
240 Example:
241 ● db_connect(connection_id="bad", db_type="postgresql")
242 ⎿ Connection failed: Invalid credentials ✗
243 """
244 self.print_action_result(tool_name, args, f"Error: {error}", success=False)
246 def update_status(
247 self,
248 mode: Optional[str] = None,
249 version: Optional[str] = None,
250 context: Optional[str] = None
251 ):
252 """Update status bar values"""
253 if mode is not None:
254 self.status_mode = mode
255 if version is not None:
256 self.status_version = version
257 if context is not None:
258 self.status_context = context
260 def render_with_status_bar(self):
261 """
262 Render interface with persistent bottom status bar.
264 Layout:
265 [Chat messages - scrollable]
266 ────────────────────────────
267 >> mode | version | context
268 """
269 # Get terminal size
270 terminal_height = console.height
272 # Chat area (all but last 2 lines)
273 chat_lines = self.messages[-(terminal_height - 3):]
274 chat_content = '\n'.join(chat_lines)
276 # Status bar
277 status_bar = self.format_status_bar()
279 # Create layout
280 layout = Layout()
281 layout.split_column(
282 Layout(name="chat", ratio=1),
283 Layout(name="status", size=1)
284 )
286 layout["chat"].update(Text(chat_content))
287 layout["status"].update(Panel(
288 status_bar,
289 style="bold blue",
290 box=box.SIMPLE,
291 padding=(0, 1)
292 ))
294 return layout
296 def start_live(self):
297 """Start live rendering with status bar"""
298 self.live = Live(
299 self.render_with_status_bar(),
300 console=console,
301 refresh_per_second=2
302 )
303 self.live.start()
305 def stop_live(self):
306 """Stop live rendering"""
307 if self.live:
308 self.live.stop()
310 def update_live(self):
311 """Update live display"""
312 if self.live:
313 self.live.update(self.render_with_status_bar())
316# Global formatter instance
317letta_formatter = LettaFormatter()
320# Convenience functions
321def action_result(
322 tool_name: str,
323 args: Dict[str, Any],
324 result_summary: str,
325 success: bool = True
326):
327 """Print action-result pair in Letta style"""
328 letta_formatter.print_action_result(tool_name, args, result_summary, success)
331def thinking(reasoning: str):
332 """Print thinking block in Letta style"""
333 letta_formatter.print_thinking(reasoning)
336def update_status(mode: str = None, context: str = None):
337 """Update status bar"""
338 letta_formatter.update_status(mode=mode, context=context)
341def summarize_file(file_path: str, content: str) -> str:
342 """Summarize file instead of dumping content"""
343 return letta_formatter.summarize_file_content(file_path, content)
346def summarize_query(result: Dict[str, Any]) -> str:
347 """Summarize query results"""
348 return letta_formatter.summarize_query_results(result)
351def summarize_schema(schema: Dict[str, Any]) -> str:
352 """Summarize database schema"""
353 return letta_formatter.summarize_schema(schema)
356# Examples of usage
357def example_database_operation():
358 """Example of database operation with Letta formatting"""
360 # Show thinking
361 thinking("I need to connect to the database first, then query the users table to count how many users exist.")
363 # Connect action
364 action_result(
365 "db_connect",
366 {
367 "connection_id": "mydb",
368 "db_type": "postgresql",
369 "host": "localhost",
370 "database": "users_db"
371 },
372 "Successfully connected to postgresql database",
373 success=True
374 )
376 # Query action
377 action_result(
378 "db_query",
379 {
380 "connection_id": "mydb",
381 "query": "SELECT COUNT(*) FROM users"
382 },
383 "Found 150 rows in users table",
384 success=True
385 )
387 # Schema action
388 action_result(
389 "db_schema",
390 {"connection_id": "mydb"},
391 "Schema: 12 tables, 85 columns total",
392 success=True
393 )
396def example_file_operation():
397 """Example of file operation with Letta formatting"""
399 thinking("I'll read the configuration file to understand the current settings.")
401 # Simulate reading a large file
402 large_content = "import os\nfrom pathlib import Path\n" + ("# More code...\n" * 100)
404 action_result(
405 "read_file",
406 {"file_path": "config.py"},
407 summarize_file("config.py", large_content),
408 success=True
409 )
412def example_cloud_operation():
413 """Example of cloud operation with Letta formatting"""
415 thinking("I need to list the EC2 instances to see what's currently running.")
417 action_result(
418 "aws_ec2_list",
419 {"region": "us-east-1"},
420 "Found 3 instances: 2 running, 1 stopped",
421 success=True
422 )
425def example_error():
426 """Example of error handling with Letta formatting"""
428 thinking("Let me try to connect to the database.")
430 action_result(
431 "db_connect",
432 {
433 "connection_id": "baddb",
434 "db_type": "postgresql",
435 "host": "invalid.example.com"
436 },
437 "Connection failed: Connection refused",
438 success=False
439 )