Coverage for src / osiris_cli / live_menus.py: 0%
227 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-27 17:41 +0200
1#!/usr/bin/env python3
2"""
3Live Interactive Submenus for Osiris CLI v4.0
4All submenus are real-time and live with beautiful UX
5"""
7import questionary
8from questionary import Style
9from rich.console import Console
10from typing import Optional, List, Dict
11from pathlib import Path
13console = Console()
15# Completely backgroundless minimal style
16MENU_STYLE = Style([
17 ('qmark', 'fg:#60a5fa'),
18 ('question', 'fg:#e5e7eb'),
19 ('answer', 'fg:#e5e7eb'),
20 ('pointer', 'fg:#60a5fa'),
21 ('highlighted', 'fg:#ffffff bold'), # No background
22 ('selected', 'fg:#e5e7eb'),
23 ('separator', 'fg:#6b7280'),
24 ('instruction', 'fg:#9ca3af'),
25 ('text', 'fg:#d1d5db'),
26 ('disabled', 'fg:#6b7280 italic')
27])
30async def select_model():
31 """Live model selection menu with global search and auto-provider config"""
32 from .config import settings, KNOWN_PROVIDERS
33 from .model_fetcher import model_fetcher
35 # 1. Ask for Provider FIRST (or 'Search All')
36 provider_choices = [
37 questionary.Choice(f"🔍 Search All Providers", "all"),
38 questionary.Separator()
39 ]
40 for p_id, p_info in KNOWN_PROVIDERS.items():
41 provider_choices.append(questionary.Choice(p_info["name"], p_id))
43 selected_provider = await questionary.select(
44 "Select provider or search all:",
45 choices=provider_choices,
46 style=MENU_STYLE
47 ).ask_async()
49 if not selected_provider:
50 return None
52 all_models = []
54 async def fetch_for_provider(p_id):
55 p_info = KNOWN_PROVIDERS[p_id]
56 with console.status(f"[dim]Fetching from {p_info['name']}...[/dim]", spinner="dots"):
57 try:
58 if p_id == "openrouter":
59 m_list = await model_fetcher.fetch_openrouter_models()
60 elif p_id == "openai":
61 key = settings.openai_api_key.get_secret_value() if settings.openai_api_key else ""
62 m_list = await model_fetcher.fetch_openai_models(key)
63 elif p_id == "gemini":
64 key = settings.gemini_api_key.get_secret_value() if settings.gemini_api_key else ""
65 m_list = await model_fetcher.fetch_gemini_models(key)
66 elif p_id == "ollama":
67 m_list = await model_fetcher.fetch_ollama_models()
68 else:
69 key_val = getattr(settings, p_info["api_key_name"], None) if p_info.get("api_key_name") else None
70 key = key_val.get_secret_value() if hasattr(key_val, "get_secret_value") else key_val
71 m_list = await model_fetcher.fetch_generic_models(p_info["base_url"], key)
73 for m in m_list:
74 m["_provider"] = p_id
75 return m_list
76 except Exception:
77 return []
79 if selected_provider == "all":
80 import asyncio
81 tasks = [fetch_for_provider(p_id) for p_id in KNOWN_PROVIDERS.keys()]
82 results = await asyncio.gather(*tasks)
83 for r in results:
84 all_models.extend(r)
85 else:
86 all_models = await fetch_for_provider(selected_provider)
88 if not all_models:
89 console.print("[yellow]No models found.[/yellow]")
90 return None
92 # Sorting: Current model first, then alphabetically
93 all_models.sort(key=lambda x: (x["id"] != settings.default_model, x["id"]))
95 choices = []
96 for m in all_models:
97 m_id = m["id"]
98 p_name = KNOWN_PROVIDERS[m["_provider"]]["name"]
99 display_name = f"{m.get('name', m_id)} ({p_name})"
100 if m_id == settings.default_model:
101 display_name = f"* {display_name} (current)"
102 choices.append(questionary.Choice(display_name, value=m))
104 selected_m_obj = await questionary.select(
105 f"Select model ({len(all_models)} available):",
106 choices=choices,
107 style=MENU_STYLE
108 ).ask_async()
110 if selected_m_obj:
111 # AUTOMATIC CONFIGURATION
112 m_id = selected_m_obj["id"]
113 p_id = selected_m_obj["_provider"]
115 settings.provider = p_id
116 settings.default_model = m_id
118 # Reset base URL override unless it's a special one
119 settings.provider_base_url = None
121 settings.save()
122 console.print(f"[green]✓[/green] Configured: [bold]{p_id}/{m_id}[/bold]")
123 return m_id
125 return None
128async def select_provider() -> Optional[str]:
129 """Enhanced provider selection with API key management and auto model loading"""
130 from .config import settings
131 from .key_manager import key_manager
132 import questionary
134 providers = [
135 {"name": "openrouter", "display": "OpenRouter (100+ models, free options)", "needs_key": True},
136 {"name": "openai", "display": "OpenAI (GPT-4, GPT-3.5)", "needs_key": True},
137 {"name": "gemini", "display": "Google Gemini", "needs_key": True},
138 {"name": "deepseek", "display": "DeepSeek (Reasoning)", "needs_key": True},
139 {"name": "together", "display": "Together AI", "needs_key": True},
140 {"name": "mistral", "display": "Mistral AI", "needs_key": True},
141 {"name": "perplexity", "display": "Perplexity (Sonar, Research)", "needs_key": True},
142 {"name": "fireworks", "display": "Fireworks AI", "needs_key": True},
143 {"name": "groq", "display": "Groq (Fast Inference)", "needs_key": True},
144 {"name": "ollama", "display": "Ollama (Local Models)", "needs_key": False},
145 ]
147 current = settings.provider
149 choices = []
150 for p in providers:
151 # Check if we have a saved key
152 has_saved_key = key_manager.has_key(p["name"])
153 key_indicator = " 🔑" if has_saved_key else ""
155 if p["name"] == current:
156 choices.append(questionary.Choice(
157 title=f"* {p['display']}{key_indicator} (current)",
158 value=p["name"]
159 ))
160 else:
161 choices.append(questionary.Choice(
162 title=f" {p['display']}{key_indicator}",
163 value=p["name"]
164 ))
166 choices.append(questionary.Separator())
167 choices.append(questionary.Choice(title="Cancel", value=None))
169 try:
170 selected_provider = await questionary.select(
171 "Select AI Provider:",
172 choices=choices,
173 style=MENU_STYLE,
174 instruction="(🔑 = saved key | arrow keys, type to filter, Enter to select, Esc to cancel)"
175 ).ask_async()
177 if not selected_provider:
178 return None
180 # Find provider info
181 provider_info = next((p for p in providers if p["name"] == selected_provider), None)
183 # Handle API key if needed
184 if provider_info and provider_info["needs_key"]:
185 api_key = await handle_api_key(selected_provider)
186 if not api_key:
187 console.print("[yellow]Provider selection cancelled - no API key provided[/yellow]")
188 return None
190 # Save key to settings
191 key_attr = f"{selected_provider}_api_key"
192 if hasattr(settings, key_attr):
193 from pydantic import SecretStr
194 setattr(settings, key_attr, SecretStr(api_key))
196 # Update provider in settings
197 settings.provider = selected_provider
199 # Save all settings to config file
200 settings.save()
202 # Reload settings to ensure consistency
203 settings.reload()
205 console.print(f"[green]✓[/green] Provider changed to: {selected_provider}")
206 console.print(f"[dim]API key configured and saved[/dim]")
208 # Auto-load models for the new provider
209 console.print()
210 console.print(f"[cyan]Loading models for {selected_provider}...[/cyan]")
212 # Call select_model to show available models
213 model = await select_model()
214 if model and model != "Cancel":
215 settings.default_model = model
216 settings.save()
217 settings.reload()
218 console.print(f"[green]✓[/green] Model changed to: {model}")
220 return selected_provider
221 except (KeyboardInterrupt, EOFError):
222 return None
225async def handle_api_key(provider: str) -> Optional[str]:
226 """Handle API key for provider - use saved or add new"""
227 from .key_manager import key_manager
228 import questionary
230 # Check if we have a saved key
231 saved_key = key_manager.load_key(provider)
233 if saved_key:
234 # Show options: use saved or add new
235 masked_key = key_manager.mask_key(saved_key)
237 choices = [
238 questionary.Choice(title=f"Use saved key ({masked_key})", value="use_saved"),
239 questionary.Choice(title="Add new API key", value="add_new"),
240 questionary.Separator(),
241 questionary.Choice(title="Cancel", value="cancel"),
242 ]
244 try:
245 choice = await questionary.select(
246 f"API Key for {provider}:",
247 choices=choices,
248 style=MENU_STYLE,
249 instruction="(arrow keys, Enter to select, Esc to cancel)"
250 ).ask_async()
252 if choice == "use_saved":
253 return saved_key
254 elif choice == "add_new":
255 return await prompt_for_new_key(provider)
256 else:
257 return None
258 except (KeyboardInterrupt, EOFError):
259 return None
260 else:
261 # No saved key - prompt for new one
262 return await prompt_for_new_key(provider)
265async def prompt_for_new_key(provider: str) -> Optional[str]:
266 """Prompt user to enter new API key"""
267 from .key_manager import key_manager
268 import questionary
270 console.print()
271 console.print(f"[bold]API Key for {provider}[/bold]")
272 console.print("[dim]Get your API key from the provider's website[/dim]")
273 console.print()
275 try:
276 api_key = await questionary.password(
277 "Enter API key:",
278 style=MENU_STYLE
279 ).ask_async()
281 if not api_key:
282 return None
284 # Ask if user wants to save the key
285 save_key = await questionary.confirm(
286 "Save this API key securely for future use?",
287 default=True,
288 style=MENU_STYLE
289 ).ask_async()
291 if save_key:
292 key_manager.save_key(provider, api_key)
294 return api_key
295 except (KeyboardInterrupt, EOFError):
296 return None
299async def select_session() -> Optional[str]:
300 """Live session selection menu"""
301 sessions_dir = Path.home() / ".osiris" / "sessions"
303 if not sessions_dir.exists():
304 console.print("[yellow]No saved sessions found[/yellow]")
305 return None
307 sessions = list(sessions_dir.glob("*.json"))
309 if not sessions:
310 console.print("[yellow]No saved sessions found[/yellow]")
311 return None
313 choices = []
314 for session_file in sorted(sessions, key=lambda x: x.stat().st_mtime, reverse=True):
315 # Get file modification time
316 import datetime
317 mtime = datetime.datetime.fromtimestamp(session_file.stat().st_mtime)
318 time_str = mtime.strftime("%Y-%m-%d %H:%M")
320 choices.append(questionary.Choice(
321 title=f"📂 {session_file.stem} ({time_str})",
322 value=session_file.stem
323 ))
325 choices.append(questionary.Separator())
326 choices.append(questionary.Choice(title="🗑️ Delete a session", value="__delete__"))
327 choices.append(questionary.Choice(title="❌ Cancel", value=None))
329 try:
330 result = await questionary.select(
331 "📂 Select Session:",
332 choices=choices,
333 style=MENU_STYLE,
334 instruction="(Use arrow keys, type to filter, Enter to select, Esc to cancel)"
335 ).ask_async()
337 if result == "__delete__":
338 # Show delete submenu
339 return await delete_session(sessions)
341 return result
342 except (KeyboardInterrupt, EOFError):
343 return None
346async def delete_session(sessions: List[Path]) -> Optional[str]:
347 """Live session deletion menu"""
348 choices = []
349 for session_file in sorted(sessions, key=lambda x: x.stat().st_mtime, reverse=True):
350 choices.append(questionary.Choice(
351 title=f"🗑️ {session_file.stem}",
352 value=str(session_file)
353 ))
355 choices.append(questionary.Separator())
356 choices.append(questionary.Choice(title="❌ Cancel", value=None))
358 try:
359 result = await questionary.select(
360 "🗑️ Select session to delete:",
361 choices=choices,
362 style=MENU_STYLE,
363 instruction="(Use arrow keys, Enter to select, Esc to cancel)"
364 ).ask_async()
366 if result:
367 # Confirm deletion
368 confirm = await questionary.confirm(
369 f"Are you sure you want to delete this session?",
370 default=False,
371 style=MENU_STYLE
372 ).ask_async()
374 if confirm:
375 Path(result).unlink()
376 console.print(f"[green]✓[/green] Session deleted")
378 return None
379 except (KeyboardInterrupt, EOFError):
380 return None
383async def show_tools_menu():
384 """Live tools display with categorization"""
385 from .tools import tools
387 tool_defs = tools.get_definitions()
389 # Categorize tools
390 categories = {
391 "File Operations": ["read_file", "write_file", "list_dir", "get_tree"],
392 "Code Analysis": ["search_codebase", "get_symbols", "check_syntax"],
393 "System": ["run_shell"],
394 "Web": ["web_search", "read_url"],
395 "Advanced": ["mcp_call", "lsp_diagnostics"],
396 }
398 choices = []
400 for category, tool_names in categories.items():
401 choices.append(questionary.Separator(f"\n═══ {category} ═══"))
402 for tool_name in tool_names:
403 if tool_name in tool_defs:
404 tool = tool_defs[tool_name]
405 choices.append(questionary.Choice(
406 title=f"🛠️ {tool_name:<20} │ {tool['description']}",
407 value=tool_name
408 ))
410 choices.append(questionary.Separator())
411 choices.append(questionary.Choice(title="❌ Close", value=None))
413 try:
414 result = await questionary.select(
415 f"🛠️ Available Tools ({len(tool_defs)}):",
416 choices=choices,
417 style=MENU_STYLE,
418 instruction="(Use arrow keys to browse, Enter to see details, Esc to close)"
419 ).ask_async()
421 if result:
422 # Show tool details
423 tool = tool_defs[result]
424 console.print(f"\n[bold cyan]🛠️ {result}[/bold cyan]")
425 console.print(f"[dim]{tool['description']}[/dim]\n")
427 if 'parameters' in tool:
428 console.print("[bold]Parameters:[/bold]")
429 for param, details in tool['parameters'].items():
430 # Handle both dict and string parameter descriptions
431 if isinstance(details, dict):
432 desc = details.get('description', 'N/A')
433 else:
434 desc = str(details)
435 console.print(f" • [cyan]{param}[/cyan]: {desc}")
437 console.print()
439 except (KeyboardInterrupt, EOFError):
440 pass
443async def show_memory_menu(agent_name: str = "main"):
444 """Live memory management menu"""
445 from .minimal_cli import AgentMemory
447 memory = AgentMemory(agent_name)
449 choices = [
450 questionary.Choice(title="View All Memories", value="view"),
451 questionary.Choice(title="Add Memory", value="add"),
452 questionary.Choice(title="Remove Memory", value="remove"),
453 questionary.Choice(title="Initialize Memory", value="init"),
454 questionary.Choice(title="Add Skill", value="skill"),
455 questionary.Separator(),
456 questionary.Choice(title="Cancel", value=None),
457 ]
459 try:
460 result = await questionary.select(
461 "Memory Management:",
462 choices=choices,
463 style=MENU_STYLE,
464 instruction="(arrow keys, Enter to select, Esc to cancel)"
465 ).ask_async()
467 return result
468 except (KeyboardInterrupt, EOFError):
469 return None