Coverage for src/osiris_cli/dashboard

1"""

2Dashboard UI System for Osiris CLI

4Professional terminal interface with:

5- Persistent HUD (fixed top bar)

6- Collapsible blocks for tool outputs

7- Nerd Fonts/Unicode iconography

8- Stateful in-place updates

9- Clean, minimal design

10"""

12from typing import Optional, List, Dict, Any, Callable

13from dataclasses import dataclass, field

14from datetime import datetime

15from enum import Enum

17from rich.console import Console, Group

18from rich.layout import Layout

19from rich.panel import Panel

20from rich.table import Table

21from rich.live import Live

22from rich.text import Text

23from rich.tree import Tree

24from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn

25from rich.align import Align

26from rich import box

28from .logger import get_logger

30logger = get_logger()

31console = Console()

34# Professional Icons (Nerd Fonts / Unicode)

35class Icons:

36 """Professional iconography using Nerd Fonts and Unicode"""

37 # System Status

38 SUCCESS = "✔"

39 ERROR = "✖"

40 WARNING = "⚠"

41 INFO = "ℹ"

43 # Operations

44 RUNNING = "⟳"

45 WAITING = "⏸"

46 COMPLETED = "✓"

47 FAILED = "✗"

48 SKIPPED = "⊘"

50 # Resources

51 DATABASE = ""

52 CLOUD = "☁"

53 FILE = ""

54 FOLDER = ""

55 CODE = ""

57 # Tools

58 TOOL = ""

59 API = ""

60 SEARCH = ""

61 TERMINAL = ""

63 # Cost & Analytics

64 COST = "$"

65 TOKENS = "#"

66 TIME = ""

67 CHART = ""

69 # UI Elements

70 COLLAPSED = "▶"

71 EXPANDED = "▼"

72 BULLET = "•"

73 ARROW = "→"

74 CHECK = "☑"

75 UNCHECK = "☐"

78class BlockState(Enum):

79 """State of a collapsible block"""

80 COLLAPSED = "collapsed"

81 EXPANDED = "expanded"

82 LOADING = "loading"

83 COMPLETE = "complete"

84 ERROR = "error"

87@dataclass

88class CollapsibleBlock:

89 """A collapsible UI block for tool outputs"""

90 title: str

91 content: str

92 state: BlockState = BlockState.COLLAPSED

93 icon: str = Icons.FILE

94 timestamp: Optional[str] = None

95 metadata: Dict[str, Any] = field(default_factory=dict)

97 def render(self, show_content: bool = None) -> Panel:

98 """Render the block as a Rich Panel"""

99 if show_content is None:

100 show_content = self.state == BlockState.EXPANDED

101

102 # Title with icon and state indicator

103 state_icon = Icons.EXPANDED if show_content else Icons.COLLAPSED

104

105 # Color based on state

106 color_map = {

107 BlockState.COLLAPSED: "dim",

108 BlockState.EXPANDED: "cyan",

109 BlockState.LOADING: "yellow",

110 BlockState.COMPLETE: "green",

111 BlockState.ERROR: "red"

112 }

113 color = color_map.get(self.state, "white")

114

115 title_text = f"{state_icon} {self.icon} {self.title}"

116

117 if self.state == BlockState.LOADING:

118 title_text += f" {Icons.RUNNING}"

119 elif self.state == BlockState.COMPLETE:

120 title_text += f" {Icons.SUCCESS}"

121 elif self.state == BlockState.ERROR:

122 title_text += f" {Icons.ERROR}"

123

124 # Content (only show if expanded)

125 if show_content:

126 content_text = self.content

127 else:

128 # Show preview/summary

129 lines = self.content.split('\n')

130 if len(lines) > 3:

131 preview = '\n'.join(lines[:2]) + f"\n[dim]... ({len(lines)} lines total)[/dim]"

132 else:

133 preview = self.content

134 content_text = preview

135

136 # Add metadata footer

137 if self.metadata:

138 meta_parts = []

139 if 'size' in self.metadata:

140 meta_parts.append(f"Size: {self.metadata['size']}")

141 if 'duration' in self.metadata:

142 meta_parts.append(f"Duration: {self.metadata['duration']}")

143 if meta_parts:

144 content_text += f"\n[dim]{' | '.join(meta_parts)}[/dim]"

145

146 return Panel(

147 content_text,

148 title=title_text,

149 title_align="left",

150 border_style=color,

151 box=box.ROUNDED,

152 padding=(0, 1)

153 )

154

155

156@dataclass

157class StatusItem:

158 """A status item for the HUD"""

159 label: str

160 value: str

161 icon: str = Icons.INFO

162 color: str = "white"

163

164 def render(self) -> str:

165 """Render as formatted string"""

166 return f"[{self.color}]{self.icon} {self.label}:[/{self.color}] {self.value}"

167

168

169class DashboardHUD:

170 """

171 Persistent HUD (Heads-Up Display) at top of terminal.

172

173 Shows:

174 - Context (files loaded, active tools)

175 - System Status (provider, model, tokens)

176 - Autonomy Budget (used/total tools)

177 - Cost (session cost)

178 """

179

180 def __init__(self):

181 self.context_items: List[str] = []

182 self.status_items: List[StatusItem] = []

183 self.cost_total: float = 0.0

184 self.tokens_used: int = 0

185 self.session_id: Optional[str] = None

186 self.provider: str = "openai"

187 self.model: str = "gpt-4"

188 self.budget_used: int = 0

189 self.budget_total: int = 30

190 self.progress_score: int = 0

191 self.context_percent: float = 0.0

192 self.active_modules: set = set() # {FILES, NET, SHELL, MEM}

193 self.branch: str = "main" # Swarm tracking

194 self.active_strategy: str = "" # Current hypothesis

195 self.active_focus: List[str] = [] # Primary files being manipulated

196 self.mission_lock: str = "" # Current top-level goal

197 self.active_mission: str = "GENERAL" # Current workflow type

198 self.origin_branch: Optional[str] = None # For swarm tracking

199 self.engine_status: str = "" # Real-time micro-action

200 self.last_trace: str = "" # Last few tokens received

201 self._pulse_state: bool = False # For flickering neural pulse

202 self._entropy_pulse: int = 0 # For variable border intensity

203 self.last_module: Optional[str] = None # For 'Heartbeat' effect

204 self.active_swarm: List[Dict] = [] # Running sub-agent branches

205 self.last_latency: float = 0.0 # RTT in ms

206 self.tps: float = 0.0 # Tokens Per Second

207 self._scan_pos: int = 0 # For scanning animation

208

209 def update_context(self, items: List[str], percent: float = 0.0):

210 """Update context items and capacity"""

211 self.context_items = items

212 self.context_percent = percent

213

214 def update_status(self, provider: str, model: str, tokens: int, active: set = None, branch: str = "main"):

215 """Update system status and active modules"""

216 self.provider = provider

217 self.model = model

218 self.tokens_used = tokens

219 self.branch = branch

220 if active is not None:

221 self.active_modules = active

222

223 def update_cost(self, cost: float):

224 """Update session cost"""

225 self.cost_total = cost

226

227 def update_budget(self, used: int, total: int, progress: int, strategy: str = ""):

228 """Update autonomy budget stats and current strategy"""

229 self.budget_used = used

230 self.budget_total = total

231 self.progress_score = progress

232 if strategy:

233 self.active_strategy = strategy

234

235 def render(self, compact: bool = False) -> Group:

236 """Render HUD content as high-end Brutalist Group (raw components only)"""

237 renderables = []

238

239 # 0. Mission Lock (Ground Truth)

240 if self.mission_lock:

241 mission_text = Text()

242

243 # 2025 Tier-3: Scanning Pulse

244 is_scanning = any(x in self.engine_status.lower() for x in ["scan", "prep", "Handshake"])

245 if is_scanning:

246 self._scan_pos = (self._scan_pos + 1) % 10

247 scan_bar = ["░"] * 10

248 scan_bar[self._scan_pos] = "▓"

249 mission_text.append(f"[{''.join(scan_bar)}] ", style="bright_cyan")

250

251 mission_text.append(f" MISSION: {self.active_mission.upper()} ", style="reverse bold blue")

252 mission_text.append(" ")

253 mission_text.append(" LOCK ", style="reverse bold red" if self._pulse_state else "reverse bold white")

254 mission_text.append(f" › {self.mission_lock}", style="dim")

255 renderables.append(mission_text)

256 if not compact:

257 renderables.append("") # spacer

258

259 main_table = Table.grid(padding=(0, 1))

260 if compact:

261 # Compact table layout

262 main_table.add_column(justify="left", min_width=20)

263 main_table.add_column(justify="center", ratio=1)

264 main_table.add_column(justify="right", min_width=20)

265 else:

266 main_table.add_column(justify="left", min_width=32)

267 main_table.add_column(justify="center", min_width=50, ratio=1)

268 main_table.add_column(justify="right", min_width=35)

269

270 # 1. Context Slot & Subsystems

271 context_text = Text()

272 context_text.append(" CONTEXT ", style="reverse bold cyan")

273 context_text.append(" ")

274

275 # [W] = Waterfall (Active Stream), [T] = Test, [F] = Files

276 modules = [("F", "FILES"), ("N", "NET"), ("S", "SHELL"), ("M", "MEM"), ("T", "TEST"), ("W", "WATERFALL")]

277 for char, full in modules:

278 is_active = full in self.active_modules

279 is_heartbeat = full == self.last_module

280

281 if is_heartbeat:

282 style = "reverse bold bright_green"

283 elif is_active:

284 style = "bold bright_green"

285 else:

286 style = "dim"

287

288 context_text.append(f"[{char}]", style=style)

289 context_text.append(" ")

290

291 # 2. Status & Swarm Slot (Center)

292 status_text = Text()

293

294 # Neural Pulse (flickering indicator)

295 self._pulse_state = not self._pulse_state

296 pulse_char = "~" if self._pulse_state else "•"

297 # 2025 Swarm Pulse logic

298 pulse_color = "bright_magenta" if "." in self.branch else "cyan"

299 status_text.append(pulse_char, style=f"bold {pulse_color}" if self._pulse_state else "bold bright_white")

300 status_text.append(" ")

301

302 status_text.append(f"{self.provider.upper()}", style="bold green")

303 status_text.append(" │ ", style="dim")

304

305 branch_parts = self.branch.split(".")

306 if len(branch_parts) > 1:

307 # 2025 Sovereign Swarm: Render breadcrumbs (MAIN » BACKEND » FIX)

308 status_text.append(" » ".join(branch_parts).upper(), style="bold bright_magenta")

309 if self.origin_branch:

310 status_text.append(f" [ORIGIN:{self.origin_branch.upper()}]", style="dim")

311 else:

312 status_text.append(self.model.upper(), style="bold yellow")

313

314 status_text.append(" │ ", style="dim")

315 status_text.append("MANA ", style="bold white")

316 budget_pct = min(1.0, self.budget_used / self.budget_total) if self.budget_total > 0 else 0

317 mana_color = "bright_green" if budget_pct < 0.4 else "bright_yellow" if budget_pct < 0.7 else "bright_red"

318 status_text.append(f"[{self.budget_used}/{self.budget_total}]", style=mana_color)

319

320 # 3. Financials & Load Slot (Cognitive Load Bar)

321 fin_text = Text()

322 load_val = int(self.context_percent * 10)

323 load_bar = "█" * load_val + "░" * (10 - load_val)

324 load_color = "red" if load_val > 8 else "green"

325

326 fin_text.append("LOAD ", style="bold white")

327 fin_text.append(f"[{load_bar}]", style=load_color)

328 fin_text.append(" │ ", style="dim")

329 fin_text.append(f"{self.tokens_used:,}", style="bold white")

330 fin_text.append(" TOK │ ", style="dim")

331

332 # Latency & TPS Slot

333 lat_color = "green" if self.last_latency < 500 else "yellow" if self.last_latency < 1500 else "red"

334 fin_text.append(f"{int(self.last_latency)}", style=f"bold {lat_color}")

335 fin_text.append("ms", style="dim")

336 fin_text.append(" │ ", style="dim")

337

338 tps_color = "bright_green" if self.tps > 20 else "yellow" if self.tps > 5 else "dim"

339 fin_text.append(f"{self.tps:.1f}", style=f"bold {tps_color}")

340 fin_text.append(" TPS", style="dim")

341 fin_text.append(" │ ", style="dim")

342

343 fin_text.append(f"${self.cost_total:.4f}", style="bold magenta")

344

345 main_table.add_row(context_text, status_text, fin_text)

346 renderables.append(main_table)

347

348 if self.active_strategy and not compact:

349 strat_text = Text()

350 strat_text.append(" STRATEGY ", style="reverse bold yellow")

351 strat_text.append(f" › {self.active_strategy}", style="bright_white")

352 renderables.append(strat_text)

353

354 # 4.5 Active Focus (Primary Targets)

355 if self.active_focus and not compact:

356 focus_text = Text()

357 focus_text.append(" FOCUS ", style="reverse bold magenta")

358 focus_text.append(f" › {', '.join(self.active_focus[:2])}", style="bold white")

359 renderables.append(focus_text)

360

361 # Dynamic Border Color & Entropy Flicker

362 border_color = "bright_blue"

363 if "NET" in self.active_modules or "FILES" in self.active_modules:

364 border_color = "cyan"

365 if "SHELL" in self.active_modules or self.branch != "main":

366 border_color = "bright_magenta"

367

368 # 5. Engine Status (Micro-actions)

369 if self.engine_status and not compact:

370 eng_text = Text()

371 eng_text.append(f" ↳ {self.engine_status}", style="dim cyan")

372 renderables.append(eng_text)

373

374 # 6. Real-time Trace (Last tokens)

375 if self.last_trace and not compact:

376 trace_text = Text()

377 trace_color = "cyan" if border_color == "cyan" else "magenta"

378 trace_text.append(f" [TRACE] {self.last_trace}", style=f"dim {trace_color}")

379 renderables.append(trace_text)

380

381 # 7. Active Swarm Tray (New for Tier-3)

382 if self.active_swarm:

383 swarm_text = Text()

384 swarm_text.append(" SWARM ", style="reverse bold magenta")

385 for branch in self.active_swarm:

386 name = branch.get("name", "unknown")

387 status = branch.get("status", "working")

388 r = branch.get("reads", 0)

389 w = branch.get("writes", 0)

390

391 swarm_text.append(f" » {name.upper()} ", style="bold bright_magenta")

392 swarm_text.append(f"[", style="dim")

393 swarm_text.append(f"R:{r}", style="cyan" if r > 0 else "dim")

394 swarm_text.append(f"|", style="dim")

395 swarm_text.append(f"W:{w}", style="bright_green" if w > 0 else "dim")

396 swarm_text.append(f"] ", style="dim")

397 renderables.append(swarm_text)

398

399 # Neural Entropy Gauge: Flicker border during flow

400 if self._pulse_state:

401 self._entropy_pulse = (self._entropy_pulse + 1) % 3

402 entropy_colors = {

403 "cyan": ["cyan", "bright_cyan", "deep_sky_blue1"],

404 "bright_magenta": ["bright_magenta", "magenta", "plum1"],

405 "bright_blue": ["bright_blue", "dodger_blue1", "deep_sky_blue1"]

406 }

407 colors = entropy_colors.get(border_color, [border_color])

408 border_color = colors[self._entropy_pulse % len(colors)]

409

410 return Group(*renderables)

411

412

413class TaskProgress:

414 """

415 Stateful task progress tracker.

416 Updates in-place rather than printing new lines.

417 """

418

419 def __init__(self):

420 self.tasks: Dict[str, Dict[str, Any]] = {}

421 self.progress = Progress(

422 SpinnerColumn(),

423 TextColumn("[progress.description]{task.description}"),

424 BarColumn(),

425 TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),

426 )

427

428 def add_task(self, task_id: str, description: str, total: int = 100):

429 """Add a new task"""

430 task = self.progress.add_task(description, total=total)

431 self.tasks[task_id] = {

432 "task": task,

433 "description": description,

434 "total": total,

435 "completed": 0

436 }

437

438 def update_task(self, task_id: str, completed: int = None, description: str = None):

439 """Update task progress"""

440 if task_id not in self.tasks:

441 return

442

443 task_info = self.tasks[task_id]

444 task = task_info["task"]

445

446 if completed is not None:

447 task_info["completed"] = completed

448 self.progress.update(task, completed=completed)

449

450 if description is not None:

451 task_info["description"] = description

452 self.progress.update(task, description=description)

453

454 def complete_task(self, task_id: str):

455 """Mark task as complete"""

456 if task_id in self.tasks:

457 task_info = self.tasks[task_id]

458 task = task_info["task"]

459 self.progress.update(task, completed=task_info["total"])

460

461 def render(self) -> Progress:

462 """Get progress for rendering"""

463 return self.progress

464

465

466class ChecklistItem:

467 """A checklist item that updates in-place"""

468

469 def __init__(self, description: str, checked: bool = False):

470 self.description = description

471 self.checked = checked

472 self.icon = Icons.CHECK if checked else Icons.UNCHECK

473

474 def check(self):

475 """Mark as checked"""

476 self.checked = True

477 self.icon = Icons.CHECK

478

479 def uncheck(self):

480 """Mark as unchecked"""

481 self.checked = False

482 self.icon = Icons.UNCHECK

483

484 def render(self) -> str:

485 """Render as string"""

486 style = "green" if self.checked else "dim"

487 return f"[{style}]{self.icon} {self.description}[/{style}]"

488

489

490class DashboardUI:

491 """

492 Main Dashboard UI Manager.

493

494 Combines:

495 - Persistent HUD

496 - Collapsible blocks

497 - Task progress

498 - Checklists

499 """

500

501 def __init__(self):

502 self.hud = DashboardHUD()

503 self.blocks: List[CollapsibleBlock] = []

504 self.task_progress = TaskProgress()

505 self.checklist: List[ChecklistItem] = []

506 self.messages: List[str] = []

507 self.live: Optional[Live] = None

508

509 def start(self):

510 """Start live dashboard"""

511 self.live = Live(self._render(), console=console, refresh_per_second=4)

512 self.live.start()

513

514 def stop(self):

515 """Stop live dashboard"""

516 if self.live:

517 self.live.stop()

518

519 def _render(self) -> Layout:

520 """Render complete dashboard"""

521 layout = Layout()

522

523 # Split into HUD and content

524 layout.split_column(

525 Layout(name="hud", size=3),

526 Layout(name="content")

527 )

528

529 # HUD at top

530 layout["hud"].update(self.hud.render())

531

532 # Content area

533 content_parts = []

534

535 # Blocks

536 for block in self.blocks[-5:]: # Show last 5 blocks

537 content_parts.append(block.render())

538

539 # Task progress

540 if self.task_progress.tasks:

541 content_parts.append(self.task_progress.render())

542

543 # Checklist

544 if self.checklist:

545 checklist_text = "\n".join(item.render() for item in self.checklist)

546 content_parts.append(Panel(checklist_text, title="Tasks", border_style="cyan"))

547

548 # Recent messages

549 if self.messages:

550 messages_text = "\n".join(self.messages[-10:]) # Last 10 messages

551 content_parts.append(Panel(messages_text, title="Chat", border_style="green"))

552

553 if content_parts:

554 layout["content"].update(Group(*content_parts))

555 else:

556 layout["content"].update(Panel("[dim]No content yet[/dim]"))

557

558 return layout

559

560 def add_block(self, title: str, content: str, icon: str = Icons.FILE, state: BlockState = BlockState.COLLAPSED) -> CollapsibleBlock:

561 """Add a collapsible block"""

562 block = CollapsibleBlock(

563 title=title,

564 content=content,

565 icon=icon,

566 state=state,

567 timestamp=datetime.now().isoformat()

568 )

569 self.blocks.append(block)

570

571 if self.live:

572 self.live.update(self._render())

573

574 return block

575

576 def expand_block(self, block: CollapsibleBlock):

577 """Expand a block"""

578 block.state = BlockState.EXPANDED

579 if self.live:

580 self.live.update(self._render())

581

582 def collapse_block(self, block: CollapsibleBlock):

583 """Collapse a block"""

584 block.state = BlockState.COLLAPSED

585 if self.live:

586 self.live.update(self._render())

587

588 def add_message(self, message: str):

589 """Add a chat message"""

590 self.messages.append(message)

591 if self.live:

592 self.live.update(self._render())

593

594 def add_checklist_item(self, description: str, checked: bool = False) -> ChecklistItem:

595 """Add a checklist item"""

596 item = ChecklistItem(description, checked)

597 self.checklist.append(item)

598 if self.live:

599 self.live.update(self._render())

600 return item

601

602 def update(self):

603 """Force update of display"""

604 if self.live:

605 self.live.update(self._render())

606

607

608# Global dashboard instance

609dashboard = DashboardUI()

610

611

612# Helper functions for common operations

613def show_database_operation(operation: str, database: str, query: str, result: Dict[str, Any]):

614 """Display database operation in clean format"""

615 # Add collapsed block for the operation

616 content = f"Database: {database}\n"

617 content += f"Query: {query}\n\n"

618

619 if "rows" in result:

620 content += f"Results: {len(result['rows'])} rows\n"

621 # Format first few rows as table

622 if result['rows']:

623 content += "\nSample Data:\n"

624 for row in result['rows'][:5]:

625 content += f" {row}\n"

626 elif "documents" in result:

627 content += f"Documents: {len(result['documents'])}\n"

628

629 dashboard.add_block(

630 title=f"Database Query: {operation}",

631 content=content,

632 icon=Icons.DATABASE,

633 state=BlockState.COMPLETE

634 )

635

636

637def show_cloud_operation(operation: str, provider: str, resource: str, result: Any):

638 """Display cloud operation in clean format"""

639 content = f"Provider: {provider}\n"

640 content += f"Resource: {resource}\n\n"

641 content += f"Result: {result}\n"

642

643 dashboard.add_block(

644 title=f"Cloud Operation: {operation}",

645 content=content,

646 icon=Icons.CLOUD,

647 state=BlockState.COMPLETE

648 )

649

650

651def show_file_operation(operation: str, file_path: str, content: str):

652 """Display file operation in collapsed block"""

653 dashboard.add_block(

654 title=f"File {operation}: {file_path}",

655 content=content,

656 icon=Icons.FILE,

657 state=BlockState.COLLAPSED

658 )

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

386 statements