Coverage for src / osiris_cli / cli_v5_2.py: 0%
1958 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-31 05:01 +0200
1#!/usr/bin/env python3
2"""
3Osiris CLI v5.2 - Truly Polished & Feature-Complete
4Beautiful visuals + Full command system + Dynamic menus
5"""
7import asyncio
8import hashlib
9import json
10import time
11import re
12import sys
13import os
14from datetime import datetime
15from urllib.parse import urlparse
16from pathlib import Path
17from typing import Optional, List, Dict, Any, Tuple
18from types import SimpleNamespace
19from rich.console import Console, RenderableType, Group
20from rich.panel import Panel
21from rich.live import Live
22from rich.spinner import Spinner
23from rich.syntax import Syntax
24from rich.text import Text
25from rich.table import Table
26from rich.markdown import Markdown
27from rich.prompt import Prompt
28from rich.align import Align
29from rich import box
30from pydantic import SecretStr
32from prompt_toolkit import PromptSession
33from prompt_toolkit.completion import Completer, Completion
34from prompt_toolkit.formatted_text import HTML
35import html
36from prompt_toolkit.styles import Style as PTStyle
38from .dashboard_ui import DashboardHUD, Icons, BlockState
39from .checkpoint import CheckpointManager
40from .logger import get_logger, console
41from .chat_engine import ChatEngine, StreamingState, ToolCallStatus
42from .tools_registry import tools_registry
43from .config import settings
44from .client import get_async_client
45from .cost_tracker import cost_tracker
46from .plan_executor import plan_executor
47from .plan_mode import plan_mode, Plan, StepStatus
48from .profiles import list_profiles, save_profile, load_profile, delete_profile
49from .tools import execute_tool
50from .session import session
51from .activity_monitor import activity_monitor
52from .input_normalizer import normalize_user_input
54console = Console()
56# Enhanced OSIRIS-CLI Logo
57OSIRIS_LOGO = """
58 ██████╗ ███████╗██╗██████╗ ██╗███████╗ ██████╗██╗ ██╗
59██╔═══██╗██╔════╝██║██╔══██╗██║██╔════╝ ██╔════╝██║ ██║
60██║ ██║███████╗██║██████╔╝██║███████╗█████╗██║ ██║ ██║
61██║ ██║╚════██║██║██╔══██╗██║╚════██║╚════╝██║ ██║ ██║
62╚██████╔╝███████║██║██║ ██║██║███████║ ╚██████╗███████╗██║
63 ╚═════╝ ╚══════╝╚═╝╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝╚══════╝╚═╝
64"""
66# Comprehensive slash commands
67SLASH_COMMANDS = [
68 {"cmd": "/help", "desc": "Show command reference", "category": "Core"},
69 {"cmd": "/exit", "desc": "Exit Osiris", "category": "Core"},
70 {"cmd": "/version", "desc": "Show Osiris version", "category": "Core"},
71 {"cmd": "/shortcuts", "desc": "Show keyboard shortcuts", "category": "Core"},
73 {"cmd": "/model", "desc": "Change AI model", "category": "Model & Provider"},
74 {"cmd": "/provider", "desc": "Switch provider", "category": "Model & Provider"},
75 {"cmd": "/yolo", "desc": "Toggle autonomous mode", "category": "Model & Provider"},
76 {"cmd": "/system", "desc": "Change system prompt (codex, review, debug, etc.)", "category": "Model & Provider"},
77 {"cmd": "/setup", "desc": "Run setup wizard", "category": "Model & Provider"},
78 {"cmd": "/status", "desc": "Show config status", "category": "Model & Provider"},
80 {"cmd": "/remember", "desc": "Add to agent memory", "category": "Memory"},
81 {"cmd": "/forget", "desc": "Remove from memory", "category": "Memory"},
82 {"cmd": "/memory", "desc": "Show agent memory (use /memory edit)", "category": "Memory"},
83 {"cmd": "/skill", "desc": "Learn a new skill", "category": "Memory"},
85 {"cmd": "/session", "desc": "Manage sessions", "category": "Session"},
86 {"cmd": "/clear", "desc": "Clear screen", "category": "Session"},
87 {"cmd": "/reset", "desc": "Reset conversation", "category": "Session"},
88 {"cmd": "/summary", "desc": "Show session summary", "category": "Session"},
90 {"cmd": "/context", "desc": "Show context files + budget", "category": "Context & Tools"},
91 {"cmd": "/tools", "desc": "List available tools", "category": "Context & Tools"},
92 {"cmd": "/history", "desc": "Show conversation history", "category": "Context & Tools"},
94 {"cmd": "/profile", "desc": "Save/load config profiles", "category": "Workflow"},
95 {"cmd": "/export", "desc": "Export session (md/json)", "category": "Workflow"},
96 {"cmd": "/usage", "desc": "Show session cost & tokens", "category": "Workflow"},
97 {"cmd": "/tracking", "desc": "Adjust tracking verbosity", "category": "Workflow"},
98 {"cmd": "/report", "desc": "Generate a site report to file", "category": "Workflow"},
99 {"cmd": "/plan", "desc": "Create/load/advance plans", "category": "Workflow"},
100 {"cmd": "/activity", "desc": "Show recent tool activity", "category": "Workflow"},
101 {"cmd": "/demo", "desc": "Run a safe non-destructive demo", "category": "Workflow"},
102]
104# Keyboard shortcuts
105KEYBOARD_SHORTCUTS = [
106 {"key": "Ctrl+C", "action": "Cancel current operation"},
107 {"key": "Ctrl+D", "action": "Exit Osiris (EOF)"},
108 {"key": "Tab", "action": "Auto-complete command"},
109 {"key": "↑/↓", "action": "Navigate command history"},
110 {"key": "/", "action": "Show command menu"},
111]
114class OsirisCompleter(Completer):
117 """Quantum Command Completion: Grouped completions with technical snippets."""
120 def get_completions(self, document, complete_event):
123 text = document.text_before_cursor
126 if text.startswith("/"):
129 word = text[1:].lower()
132 for cmd_info in SLASH_COMMANDS:
135 cmd = cmd_info["cmd"][1:]
138 if cmd.startswith(word):
141 # 2025 High-End Style: Category-prefixed metadata
144 display_meta = f"[{cmd_info.get('category', 'CORE')}] {cmd_info.get('desc', '')}"
147 yield Completion(
150 f"/{cmd}",
153 start_position=-len(text),
156 display_meta=display_meta
159 )
165 # File completion (@)
168 if "@" in text:
171 last_at = text.rfind("@")
174 path_part = text[last_at+1:]
177 # We skip heavy dir scans for speed
180 try:
183 p = Path(".")
186 for f in p.iterdir():
189 if f.name.startswith(path_part):
192 yield Completion(f"@{f.name}", start_position=-len(path_part)-1)
195 except: pass
207class BeautifulCLI:
210 """Truly beautiful CLI with full feature set"""
216 # Enhanced color scheme
219 COLORS = {
222 "primary": "#00D9FF",
225 "success": "#00FF9F",
228 "warning": "#FFB800",
231 "error": "#FF006E",
234 "muted": "#6B7280",
237 "accent": "#A78BFA",
240 "logo": "#00D9FF",
243 }
249 def __init__(self, agent_name: str = "main", context_file: Optional[str] = None):
252 self.agent_name = agent_name
255 self.running = True
258 self.session_start = time.time()
261 self.session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
264 settings.session_id = self.session_id
267 self.tools_used = 0
270 self.files_modified = []
273 from collections import Counter
276 self.file_heat = Counter() # Tier-2: Track activity hotspots
282 self.key_actions = []
285 self.tool_trace: List[Dict[str, str]] = []
288 self.pending_shell_context: List[str] = []
291 self.last_plan_signature: Optional[str] = None
294 self.print_mode = False
297 self.welcome_shown = False
300 self.show_details = False
303 self._plan_dirty = False
306 self._tracker_dirty = False
309 self._in_stream = False
312 self._follow_up_queue: List[str] = []
315 self._follow_up_task: Optional[asyncio.Task[None]] = None
318 self.hud = DashboardHUD()
321 self.checkpoint_manager = CheckpointManager(self.session_id)
324 self.turn_count = 0
327 self._prefetch_task = None
330 self._cached_orientation = None
336 self.swarm_parent = None
339 swarm_ctx_path = os.getenv("OSIRIS_SWARM_CONTEXT")
342 if swarm_ctx_path and Path(swarm_ctx_path).exists():
345 try:
348 with open(swarm_ctx_path, 'r') as f:
351 swarm_data = json.load(f)
354 self.swarm_parent = swarm_data.get("parent_branch")
357 self.agent_name = f"{self.swarm_parent}.{self.agent_name}"
360 self.hud.origin_branch = self.swarm_parent
363 except: pass
369 # 2025 Sovereign Swarm: Parent Context Injection
372 if context_file:
375 self._load_parent_context(context_file)
381 # State tracking
384 # State tracking (initialized early)
390 self.content_buffer = ""
393 self.thought_buffer = ""
396 self.reflection_buffer = ""
399 from collections import deque
402 self.quantum_stream = deque(maxlen=300)
405 self.tool_events = []
408 self.turn_files_modified = []
411 self.tool_names = []
417 # Initialize chat engine with dynamic smart budgeting
420 self.chat_engine = ChatEngine(
423 client=get_async_client(),
426 settings=settings,
429 tools_registry=tools_registry,
432 max_tool_calls_per_turn=30, # Base starting point for smart autonomy
435 enable_auto_approval=settings.yolo_mode,
438 confirm_tool_callback=self._confirm_tool_execution,
441 checkpoint_manager=self.checkpoint_manager # Pass manager
444 )
450 # Load context
453 self.system_context, self.memory = self._load_system_context()
459 # Prompt session with completions
462 self.prompt_session = PromptSession(
465 completer=OsirisCompleter(),
468 complete_while_typing=True,
471 style=PTStyle.from_dict({
474 'completion-menu.completion': 'bg:#1a1a1a #00D9FF',
477 'completion-menu.completion.current': 'bg:#00D9FF #000000',
480 'completion-menu.meta.completion': 'bg:#1a1a1a #6B7280',
483 'completion-menu.meta.completion.current': 'bg:#00D9FF #000000',
486 }),
489 bottom_toolbar=self._get_bottom_toolbar # DYNAMIC FOOTER
492 )
495 self.rich_console = console
500 def _load_parent_context(self, path: str):
501 """2025 Sovereign Swarm: Inherit context from parent."""
502 p = Path(path)
503 if not p.exists(): return
505 try:
506 with open(p, 'r') as f:
507 data = json.load(f)
509 # Inherit history
510 if data.get("messages"):
511 # We skip the first system message if it's identical
512 self.chat_engine.conversation = data["messages"]
514 # Inherit stats
515 if data.get("total_tokens"):
516 self.hud.tokens_used = data["total_tokens"]
518 # Track parent
519 if data.get("parent_session_id"):
520 self.swarm_parent = data["parent_session_id"]
521 self.hud.origin_branch = self.swarm_parent
523 console.print(f" [dim]Neural context inherited from parent session: {self.swarm_parent}[/dim]")
524 except Exception as e:
525 console.print(f" [error]Failed to inherit parent context: {e}[/error]")
527 def _load_system_context(self) -> tuple[str, dict]:
528 """Load system context - MINIMAL approach for speed"""
529 context_parts = []
530 memory = {}
532 # Only load custom system prompt if set
533 if settings.system_prompt:
534 context_parts.append(settings.system_prompt)
535 else:
536 # Use optimized prompt by default
537 from .optimized_prompt import get_optimized_prompt
538 context_parts.append(get_optimized_prompt())
540 # Only load OSIRIS.md if it exists AND is small (< 50KB)
541 osiris_md = Path.cwd() / "OSIRIS.md"
542 if osiris_md.exists():
543 try:
544 size = osiris_md.stat().st_size
545 if size < 50000: # Only load if < 50KB
546 with open(osiris_md, 'r') as f:
547 context_parts.append(f"# Project Context\n{f.read()}")
548 except:
549 pass # Skip if any error
551 # Load agent memory (lightweight)
552 memory_file = Path.home() / ".osiris" / "agents" / f"{self.agent_name}.json"
553 if memory_file.exists():
554 import json
555 try:
556 with open(memory_file, 'r') as f:
557 memory = json.load(f)
558 # Only add memory summary, not full content
559 if memory.get("persona"):
560 context_parts.append(f"# Agent Persona\n{memory['persona'][:200]}")
561 except:
562 pass # Skip if any error
564 return "\n\n".join(context_parts), memory
566 async def show_welcome(self):
567 """High-end Brutalist Boot Sequence (Snappy Optimization)"""
568 if self.welcome_shown:
569 return
571 console.clear()
573 # 1. Logo Line-by-Line Typewriter
574 logo_lines = OSIRIS_LOGO.strip("\n").split("\n")
575 for line in logo_lines:
576 console.print(Text(line, style=self.COLORS["logo"]))
577 await asyncio.sleep(0.02) # Faster
579 console.print()
580 await asyncio.sleep(0.05)
582 # 2. Subsystem Initialization Narrative
583 boot_steps = [
584 ("Sensory", "FILES, NET, LINTERS"),
585 ("Motor", "SHELL, WRITER, BUILDER"),
586 ("Cognitive", "MEM, ARCHIVE, LOGIC"),
587 ]
589 for name, modules in boot_steps:
590 status = Text()
591 status.append("[ BOOT ] ", style="bold green")
592 status.append(f"Initializing {name} Subsystem... ", style="white")
593 status.append(f"({modules})", style="dim")
594 console.print(status)
595 await asyncio.sleep(0.05) # Faster
597 # Swarm Relay Confirmation
598 if self.swarm_parent:
599 relay = Text()
600 relay.append("[ SWARM ] ", style="bold bright_magenta")
601 relay.append(f"Neural Handshake Verified: {self.swarm_parent} ➔ {self.agent_name}", style="white")
602 console.print(relay)
603 await asyncio.sleep(0.1)
605 console.print()
606 await asyncio.sleep(0.1)
608 # 3. Session Info Panel (Heavy Brutalist)
609 info_table = Table.grid(padding=(0, 2))
610 info_table.add_column(style=self.COLORS["muted"], width=12)
611 info_table.add_column()
613 info_table.add_row("Agent", Text(self.agent_name.upper(), style=f"bold {self.COLORS['primary']}"))
614 info_table.add_row("Model", Text(settings.default_model, style=self.COLORS["accent"]))
615 info_table.add_row("Mode", Text("YOLO (autonomous)" if settings.yolo_mode else "Interactive",
616 style=self.COLORS["success"] if settings.yolo_mode else self.COLORS["warning"]))
618 console.print(Panel(
619 info_table,
620 border_style=self.COLORS["primary"],
621 padding=(1, 2),
622 title="[b] SYSTEM READY [/b]",
623 title_align="left",
624 box=box.HEAVY
625 ))
627 # 4. Quick Help
628 help_text = Text()
629 help_text.append("\n💡 ", style="bold")
630 help_text.append("Type ", style=self.COLORS["muted"])
631 help_text.append("/", style=f"bold {self.COLORS['primary']}")
632 help_text.append(" for commands • ", style=self.COLORS["muted"])
633 help_text.append("!cmd", style=f"bold {self.COLORS['primary']}")
634 help_text.append(" run shell • ", style=self.COLORS["muted"])
635 help_text.append("@file", style=f"bold {self.COLORS['primary']}")
636 help_text.append(" attach\n", style=self.COLORS["muted"])
637 console.print(help_text)
638 self.welcome_shown = True
640 def _print_section(self, title: str, body: RenderableType, style: str) -> None:
641 panel = Panel(
642 body,
643 title=f"[bold]{title}[/bold]",
644 title_align="left",
645 border_style=style,
646 padding=(0, 1),
647 )
648 console.print(panel)
650 def _render_user_panel(self, user_input: str):
651 if self.print_mode or not user_input.strip():
652 return
653 console.print(self._build_user_panel(user_input))
655 def _build_user_panel(self, user_input: str) -> Panel:
656 user_body = Syntax(user_input, "text", theme="ansi_dark", word_wrap=True)
657 return Panel(
658 user_body,
659 title="[bold]You[/bold]",
660 title_align="left",
661 border_style=self.COLORS["primary"],
662 padding=(1, 2),
663 box=box.HEAVY
664 )
666 def _render_thought_panel(self, content: str) -> Panel:
667 """Ghost Stream: Dimmed panel for inner monologue."""
668 body = Text()
669 body.append(content.strip(), style="dim white")
671 return Panel(
672 body,
673 title="[bold]THOUGHT[/bold]",
674 title_align="left",
675 border_style="dim blue",
676 padding=(0, 1),
677 box=box.ASCII2 # Consistent with waterfall
678 )
680 def _render_response_panel(
681 self,
682 content: str,
683 thinking: bool = False,
684 elapsed: Optional[float] = None,
685 footer_hint: Optional[str] = None
686 ) -> Panel:
687 elapsed_label = f"{elapsed:.1f}s" if elapsed is not None else "--"
688 session_stats = cost_tracker.get_session_costs(self.session_id) if self.session_id else {}
689 actual_tokens = session_stats.get("total_tokens", 0) if session_stats else 0
690 tokens_label = f"~{actual_tokens} tok"
692 cost_label = f"${session_stats.get('total_cost', 0.0):.4f}" if session_stats else "$0.0000"
693 pulse_text = f"{settings.provider}/{settings.default_model} \u2022 {tokens_label} \u2022 {elapsed_label} \u2022 {cost_label}"
694 subtitle = Text(pulse_text, style=self.COLORS["muted"])
695 renderables: list[RenderableType] = []
696 if thinking and not content.strip():
697 status_line = f"Osiris working \u2022 {elapsed_label}"
698 renderables.append(Spinner("simpleDots", text=status_line, style=self.COLORS["muted"]))
699 elif content.strip():
700 header = Text(f"Streaming \u2022 {elapsed_label}", style=self.COLORS["muted"])
701 renderables.append(header)
702 renderables.append(Markdown(content.strip(), code_theme="monokai"))
703 else:
704 renderables.append(Text("Neural stream silent. The model returned no text.", style="dim yellow"))
706 if footer_hint:
707 renderables.append(Text(footer_hint, style=self.COLORS["muted"]))
709 body: RenderableType = Group(*renderables)
710 return Panel(
711 body,
712 title="[bold]Assistant[/bold]",
713 title_align="left",
714 border_style=self.COLORS["primary"],
715 padding=(1, 2),
716 subtitle=subtitle,
717 subtitle_align="right",
718 )
720 def _stream_footer_hint(self) -> str:
721 return "Ctrl+C cancels • Follow-up input active"
723 def _enqueue_follow_up(self, text: str) -> bool:
724 normalized = normalize_user_input(text)
725 if not normalized.strip():
726 return False
727 self._follow_up_queue.append(normalized)
728 console.print(Text("Queued follow-up message.", style=self.COLORS["muted"]))
729 return True
731 async def _collect_follow_up_input(self):
732 prompt_text = HTML('<ansigreen>Follow-up</ansigreen> <style fg="#6B7280">\u203a</style> ')
733 while True:
734 from prompt_toolkit.patch_stdout import patch_stdout
735 with patch_stdout():
736 try:
737 follow_up = await self.prompt_session.prompt_async(prompt_text)
738 except (asyncio.CancelledError, EOFError, KeyboardInterrupt):
739 break
740 if not self._enqueue_follow_up(follow_up):
741 continue
742 if self._follow_up_task and self._follow_up_task.done():
743 self._follow_up_task = None
745 def _start_follow_up_listener(self):
746 if self.print_mode or self._follow_up_task:
747 return
748 self._follow_up_task = asyncio.create_task(self._collect_follow_up_input())
750 async def _stop_follow_up_listener(self):
751 if not self._follow_up_task:
752 return
753 task = self._follow_up_task
754 self._follow_up_task = None
755 task.cancel()
756 try:
757 # 2025 Hardening: Explicitly await to ensure prompt_toolkit session is cleared
758 await asyncio.wait_for(task, timeout=1.0)
759 except (asyncio.CancelledError, asyncio.TimeoutError):
760 pass
762 async def _drain_follow_up_queue(self):
763 while self._follow_up_queue:
764 follow_up = self._follow_up_queue.pop(0)
765 if not follow_up.strip():
766 continue
767 if not self.print_mode:
768 console.print(Text("\u2192 Auto-sending queued follow-up...", style=self.COLORS["muted"]))
769 await self.handle_message(follow_up)
771 def _format_tool_args(self, args: dict, limit: int = 3) -> RenderableType:
772 if not args:
773 return Text("No arguments", style=self.COLORS["muted"])
774 table = Table.grid(padding=(0, 1))
775 table.add_column(style=self.COLORS["muted"], width=16)
776 table.add_column()
777 for key, value in list(args.items())[:limit]:
778 value_str = str(value).replace("\n", " ")
779 if len(value_str) > 80:
780 value_str = value_str[:80] + "..."
781 table.add_row(str(key), value_str)
782 if len(args) > limit:
783 table.add_row("...", f"+{len(args) - limit} more")
784 return table
786 def _format_tool_preview(self, result: str, limit: int = 240) -> RenderableType:
787 if not result:
788 return Text("No output", style=self.COLORS["muted"])
789 text = result.replace("\r", "").strip()
790 if text.startswith(("{", "[")) and len(text) < 5000:
791 try:
792 parsed = json.loads(text)
793 pretty = json.dumps(parsed, indent=2, ensure_ascii=True)
794 text = pretty
795 except json.JSONDecodeError:
796 pass
797 if len(text) > limit:
798 text = text[:limit] + "..."
799 return Syntax(text, "text", theme="ansi_dark", word_wrap=True)
801 def _format_tool_inline(self, tool, duration: Optional[float]) -> Text:
802 name = tool.name
803 duration_label = f"{duration:.2f}s" if duration else "--"
804 args = tool.arguments or {}
805 detail = ""
806 if name == "read_file":
807 path = args.get("path", "")
808 start = args.get("start_line")
809 end = args.get("end_line")
810 if start and end:
811 detail = f"{path}:{start}-{end}"
812 else:
813 detail = path
814 elif name == "run_shell_command":
815 detail = args.get("command", "").strip()
816 elif name == "list_directory":
817 detail = args.get("path", ".")
818 elif name == "get_file_tree":
819 detail = args.get("path", ".")
820 elif name in {"write_file", "replace_file_content"}:
821 detail = args.get("path", "")
822 else:
823 detail = ", ".join(f"{k}={v}" for k, v in list(args.items())[:2])
824 if detail:
825 detail = f" — {detail}"
827 line = Text()
828 line.append("• ", style=self.COLORS["muted"])
829 line.append(name, style=f"bold {self.COLORS['primary']}")
830 if duration:
831 line.append(f" {duration_label}", style=self.COLORS["muted"])
832 if detail:
833 line.append(detail, style=self.COLORS["muted"])
834 return line
836 def _format_tool_compact(self, tool, duration: Optional[float]) -> Text:
837 name = tool.name
838 duration_label = f"{duration:.2f}s" if duration else "--"
839 args = tool.arguments or {}
840 detail = ""
841 if name == "read_file":
842 path = args.get("path", "")
843 start = args.get("start_line")
844 end = args.get("end_line")
845 detail = f"{path}:{start}-{end}" if start and end else path
846 elif name == "run_shell_command":
847 detail = args.get("command", "").strip()
848 elif name in {"list_directory", "get_file_tree"}:
849 detail = args.get("path", ".")
850 elif name in {"write_file", "replace_file_content"}:
851 detail = args.get("path", "")
852 else:
853 detail = ", ".join(f"{k}={v}" for k, v in list(args.items())[:2])
854 if detail:
855 detail = f" — {detail}"
857 line = Text()
858 line.append(name, style=f"bold {self.COLORS['primary']}")
859 if duration:
860 line.append(f" {duration_label}", style=self.COLORS["muted"])
861 if detail:
862 line.append(detail, style=self.COLORS["muted"])
863 return line
865 def _format_tool_result_inline(
866 self,
867 result: Optional[str],
868 structured: Optional[Dict[str, Any]] = None,
869 limit: int = 120
870 ) -> Optional[Text]:
871 if not isinstance(structured, dict):
872 structured = {}
874 if not result and not structured:
875 return None
876 line = Text()
877 line.append(" ↳ ", style=self.COLORS["muted"])
878 if structured:
879 status = structured.get("status")
880 exit_code = structured.get("exit_code")
881 meta = structured.get("meta") or {}
882 pid = meta.get("pid")
883 url = meta.get("url")
884 parts = []
885 if status:
886 parts.append(status.capitalize())
887 if exit_code is not None:
888 parts.append(f"exit={exit_code}")
889 if pid:
890 parts.append(f"PID {pid}")
891 if url:
892 parts.append(url)
893 if parts:
894 line.append(" | ".join(parts), style=self.COLORS["muted"])
895 message_text = (structured.get("message") or "").strip()
896 summary_line = ""
897 if message_text:
898 summary_line = message_text.splitlines()[0]
899 if len(summary_line) > limit:
900 summary_line = summary_line[:limit] + "..."
901 line.append("\n ")
902 line.append(summary_line, style=self.COLORS["muted"])
903 upper_summary = summary_line.upper()
904 stdout_text = structured.get("stdout", "").strip()
905 if stdout_text:
906 stdout_line = stdout_text.splitlines()[0]
907 duplicate_stdout = self._is_duplicate_output_line(summary_line, stdout_line)
908 show_stdout = not (
909 duplicate_stdout or "STDOUT:" in upper_summary
910 )
911 if show_stdout:
912 line.append(f"\n STDOUT: {stdout_line[:limit]}", style=self.COLORS["muted"])
913 stderr_text = (structured.get("stderr") or "").strip()
914 if stderr_text:
915 stderr_line = stderr_text.splitlines()[0]
916 duplicate_stderr = self._is_duplicate_output_line(summary_line, stderr_line)
917 show_stderr = not (
918 duplicate_stderr or "STDERR:" in upper_summary
919 )
920 if show_stderr:
921 line.append(f"\n STDERR: {stderr_line[:limit]}", style=self.COLORS["muted"])
922 detach_hint = (meta.get("detach_hint") or "").strip()
923 if detach_hint:
924 line.append(f"\n {detach_hint}", style=self.COLORS["warning"])
925 else:
926 text = (result or "").replace("\r", "").strip().splitlines()[0] if result else ""
927 if len(text) > limit:
928 text = text[:limit] + "..."
929 if not text:
930 return None
931 line.append(text, style=self.COLORS["muted"])
932 return line if line.plain.strip() else None
934 def _is_duplicate_output_line(self, summary_line: str, candidate_line: str) -> bool:
935 primary = summary_line.strip().lower()
936 candidate = candidate_line.strip().lower()
937 return bool(primary and candidate and primary == candidate)
939 def _build_tool_report_panel(
940 self,
941 tool_events: List[Tuple[Any, Optional[float]]],
942 limit: int = 120
943 ) -> Optional[RenderableType]:
944 report_lines: List[Text] = []
945 for index, (tool, _) in enumerate(tool_events, start=1):
946 structured = getattr(tool, "structured_result", None)
947 if not isinstance(structured, dict):
948 structured = {}
949 if not structured:
950 continue
951 status = structured.get("status", "unknown")
952 exit_code = structured.get("exit_code")
953 meta = structured.get("meta") or {}
954 pid = meta.get("pid")
955 url = meta.get("url")
956 stop_hint = ""
957 if pid:
958 stop_hint = f"kill {pid} (PID, not port)"
959 message_text = (structured.get("message") or "").replace("\r", "").strip()
961 line = Text()
962 line.append(f"{index}. ", style=self.COLORS["muted"])
963 line.append(tool.name, style=f"bold {self.COLORS['primary']}")
964 status_parts = []
965 if status:
966 status_parts.append(status.capitalize())
967 if exit_code is not None:
968 status_parts.append(f"exit={exit_code}")
969 if status_parts:
970 line.append(f" ({' | '.join(status_parts)})", style=self.COLORS["muted"])
971 if url:
972 line.append(f"\n URL: {url}", style=self.COLORS["accent"])
973 if pid:
974 line.append(f"\n PID: {pid}", style=self.COLORS["accent"])
975 if stop_hint:
976 line.append(f"\n Stop: {stop_hint}", style=self.COLORS["warning"])
977 summary_text = ""
978 if message_text and message_text.lower() not in {"success", "running"}:
979 summary_text = message_text.splitlines()[0]
980 if len(summary_text) > limit:
981 summary_text = summary_text[:limit] + "..."
982 line.append(f"\n {summary_text}", style=self.COLORS["muted"])
983 stdout_text = (structured.get("stdout") or "").strip()
984 if stdout_text:
985 stdout_line = stdout_text.splitlines()[0]
986 if not self._is_duplicate_output_line(summary_text, stdout_line):
987 line.append(f"\n STDOUT: {stdout_line[:limit]}", style=self.COLORS["muted"])
988 stderr_text = (structured.get("stderr") or "").strip()
989 if stderr_text:
990 stderr_line = stderr_text.splitlines()[0]
991 if not self._is_duplicate_output_line(summary_text, stderr_line):
992 line.append(f"\n STDERR: {stderr_line[:limit]}", style=self.COLORS["error"])
994 report_lines.append(line)
996 if not report_lines:
997 return None
999 return Group(*report_lines)
1001 def _estimate_tokens(self, text: str) -> int:
1002 return max(0, len(text) // 4)
1004 def _build_context_budget(self, user_input: str) -> Text:
1005 conversation = self.chat_engine.get_conversation()
1006 history_text = "\n".join(str(m.get("content", "")) for m in conversation)
1007 combined = f"{self.system_context}\n{history_text}\n{user_input}"
1008 used_tokens = self._estimate_tokens(combined)
1009 max_tokens = max(1, settings.max_context_chars // 4)
1010 pct = min(1.0, used_tokens / max_tokens)
1011 bar_width = 24
1012 filled = int(bar_width * pct)
1013 bar = "[" + ("#" * filled) + ("-" * (bar_width - filled)) + "]"
1014 meter = Text()
1015 meter.append("Context ", style=self.COLORS["muted"])
1016 meter.append(bar, style=self.COLORS["primary"])
1017 meter.append(f" {int(pct * 100)}% ", style=self.COLORS["accent"])
1018 meter.append(f"({used_tokens:,}/{max_tokens:,} tok)", style=self.COLORS["muted"])
1019 return meter
1021 def _is_document_request(self, text: str) -> bool:
1022 lowered = text.lower()
1023 lowered = re.sub(r"\S+\.(md|txt|pdf|docx)\b", "", lowered)
1024 if re.search(r"\b(delete|remove|erase|rm|show|open|read|view|context)\b", lowered):
1025 return False
1026 verb = re.search(r"\b(write|create|generate|draft|produce|make|compile|deliver)\b", lowered)
1027 noun = re.search(r"\b(report|document|brief|overview|summary|write-up|deliverable)\b", lowered)
1028 return bool(verb and noun)
1030 CONSTRAINT_HEADERS = [
1031 "constraint", "constraints", "allowed tools", "blocked tools",
1032 "forbidden", "policy", "rule", "rules", "requirement", "requirements"
1033 ]
1035 def _is_constraint_header(self, header: str) -> bool:
1036 return any(keyword in header for keyword in self.CONSTRAINT_HEADERS)
1038 def _extract_task_steps(self, text: str) -> List[str]:
1039 if not text:
1040 return []
1042 lines = [line.strip() for line in text.splitlines()]
1043 steps: List[str] = []
1044 allow_section = False
1045 constraint_section = False
1047 for line in lines:
1048 if not line:
1049 allow_section = False
1050 constraint_section = False
1051 continue
1053 header_match = re.match(r"^([A-Za-z0-9\s]+):\s*$", line)
1054 if header_match:
1055 header = header_match.group(1).strip().lower()
1056 allow_section = header.startswith("tasks")
1057 constraint_section = self._is_constraint_header(header)
1058 continue
1060 bullet_match = re.match(r"^(\d+[\.\)]|[-*•])\s+(.*)$", line)
1061 if not bullet_match:
1062 continue
1064 marker, content = bullet_match.groups()
1065 content = content.strip(" .,-")
1066 if not content or constraint_section:
1067 continue
1069 is_numbered = bool(re.match(r"\d+[\.\)]", marker.strip()))
1070 is_bullet = marker.strip() in {"-", "*", "•"} or not is_numbered
1071 if is_numbered or allow_section or is_bullet:
1072 steps.append(content)
1073 if len(steps) >= 6:
1074 break
1076 if steps:
1077 return steps
1079 cleaned = re.sub(r"\s+", " ", text.strip())
1080 if len(cleaned) < 12:
1081 return []
1082 parts = re.split(
1083 r"\b(?:and then|then|also|plus|next|after that|afterwards|finally)\b|[;]",
1084 cleaned,
1085 flags=re.I
1086 )
1087 fallback_steps = [part.strip(" ,.-") for part in parts if len(part.strip()) > 10]
1088 return fallback_steps[:6]
1090 def _is_multistep_request(self, text: str) -> bool:
1091 return len(self._extract_task_steps(text)) >= 2
1093 def _plan_title_from_text(self, text: str) -> str:
1094 words = re.findall(r"[a-z0-9]+", text.lower())
1095 title = " ".join(words[:4]) or "Autonomous Plan"
1096 return title.title()
1098 async def _attempt_plan_tool_action(self, step_description: str):
1099 """Neural-Lock: Transition from Plan Step to prioritized LLM action."""
1100 # 2025 Tier-2: Speculative instruction injection
1101 instruction = f"TARGET STEP: {step_description}\nFocus your next tool calls exclusively on completing this objective."
1102 self.hud.active_strategy = step_description
1103 # We don't call handle_message directly to avoid infinite loops,
1104 # but we set the prefetch context or strategy for the next cycle.
1105 self.pending_shell_context.append(f"### CURRENT PLAN TARGET\n{instruction}")
1107 async def _maybe_create_autonomous_plan(self, user_input: str) -> Optional[Plan]:
1108 if self.print_mode:
1109 return False
1110 if self.chat_engine.is_research_request(user_input):
1111 return False
1112 if not self._is_multistep_request(user_input):
1113 return False
1114 signature = hashlib.sha256(user_input.strip().encode()).hexdigest()
1115 if signature == self.last_plan_signature:
1116 return False
1117 self.last_plan_signature = signature
1118 if plan_mode.current_plan and plan_mode.current_plan.status not in {"completed", "failed"}:
1119 return False
1120 steps = self._extract_task_steps(user_input)
1121 if len(steps) < 2:
1122 return False
1123 title = self._plan_title_from_text(user_input)
1124 description = user_input.strip()[:100]
1125 plan = plan_mode.create_plan(title, description, steps)
1126 next_step = plan_mode.get_next_step()
1127 session.add_plan_snapshot(plan.description)
1128 if next_step:
1129 plan_mode.start_step(next_step.id)
1130 await self._attempt_plan_tool_action(next_step.description)
1131 return plan
1133 def _classify_shell_command(self, command: str) -> str:
1134 if not command:
1135 return "unknown"
1136 lowered = command.strip().lower()
1137 validation_tokens = (
1138 "pytest", "ruff", "mypy", "pyright", "flake8",
1139 "npm test", "pnpm test", "yarn test", "go test", "cargo test"
1140 )
1141 if any(token in lowered for token in validation_tokens):
1142 return "validate"
1143 read_only_prefixes = (
1144 "ls", "pwd", "cat ", "rg ", "grep ", "find ",
1145 "head ", "tail ", "sed -n", "awk ", "stat ",
1146 "wc ", "du ", "df ", "ps ", "whoami", "uname",
1147 "git status", "git diff", "git log", "which ", "type "
1148 )
1149 if any(lowered.startswith(prefix) for prefix in read_only_prefixes):
1150 return "inspect"
1151 mutating_tokens = (
1152 "pip install", "npm install", "pnpm install", "yarn add",
1153 "brew install", "apt-get install", "dnf install", "pacman -s",
1154 "mv ", "cp ", "rm ", "mkdir ", "touch ", "sed -i", "perl -pi",
1155 "git apply", "git checkout", "git add", "git commit"
1156 )
1157 if any(token in lowered for token in mutating_tokens):
1158 return "implement"
1159 return "other"
1161 def _should_show_tracking(
1162 self,
1163 user_input: str,
1164 tool_events: List,
1165 files_modified: List[str]
1166 ) -> bool:
1167 mode = getattr(settings, "tracking_mode", "standard")
1168 min_tools = getattr(settings, "tracking_min_tools", 2)
1169 min_steps = getattr(settings, "tracking_min_steps", 2)
1170 if mode == "minimal":
1171 return self._is_document_request(user_input) or bool(files_modified)
1172 if mode == "verbose":
1173 return True
1175 if self.chat_engine.is_research_request(user_input):
1176 return bool(tool_events) or self._is_document_request(user_input)
1177 if self._is_document_request(user_input):
1178 return True
1179 if self._is_multistep_request(user_input) or len(self._extract_task_steps(user_input)) >= min_steps:
1180 return True
1181 if len(tool_events) >= min_tools:
1182 return True
1183 if files_modified:
1184 return True
1185 return False
1187 def _extract_first_url(self, text: str) -> Optional[str]:
1188 if not text:
1189 return None
1190 match = re.search(r"(https?://\S+|www\.[^\s]+)", text)
1191 return match.group(0) if match else None
1193 def _extract_path_mentions(self, text: str) -> List[str]:
1194 """Extract @path mentions, supporting spaces in filenames."""
1195 if not text:
1196 return []
1198 # Look for @ followed by a sequence that ends in a file extension
1199 # This matches things like @WhatsApp Image ... .jpeg
1200 mentions = []
1201 # Pattern: @ + non-greedy chars until a known extension + boundary
1202 pattern = r'@([^\n@]+\.(?:py|js|ts|md|txt|html|css|json|toml|yaml|yml|png|jpg|jpeg|webp|gif|Dockerfile|sh|bash))\b'
1204 tokens = re.findall(pattern, text)
1205 for token in tokens:
1206 cleaned = token.strip().strip(",.;:)")
1207 # Check if this actually exists to avoid false positives
1208 if Path(cleaned).exists() or Path(Path.cwd() / cleaned).exists():
1209 mentions.append(cleaned)
1210 else:
1211 # If it doesn't exist, maybe it's just the first word?
1212 first_word = cleaned.split()[0]
1213 if Path(first_word).exists():
1214 mentions.append(first_word)
1216 return list(dict.fromkeys(mentions))[:5]
1218 def _default_deliverable_name(self, user_input: str, extension: str = "md") -> str:
1219 url = self._extract_first_url(user_input)
1220 if url:
1221 parsed = urlparse(url if url.startswith(("http://", "https://")) else f"https://{url}")
1222 host = parsed.netloc or "site_report"
1223 safe = host.replace(".", "_")
1224 return f"{safe}_report.{extension}"
1225 words = re.findall(r"[a-z0-9]+", user_input.lower())
1226 slug = "_".join(words[:6]) if words else "osiris_deliverable"
1227 return f"{slug}.{extension}"
1229 def _is_task_request(self, text: str, tool_events: List) -> bool:
1230 if tool_events:
1231 return True
1232 lowered = text.lower()
1233 keywords = [
1234 "build", "implement", "fix", "refactor", "improve", "enhance",
1235 "organize", "setup", "install", "debug", "design", "create",
1236 "add", "remove", "update", "change", "migrate", "review"
1237 ]
1238 return len(text.strip()) > 80 or any(key in lowered for key in keywords)
1240 def _is_env_intent(self, text: str) -> bool:
1241 """Sovereign Detection: Is the user asking about their environment?"""
1242 lowered = text.lower()
1243 probes = ["where am i", "where are you", "pwd", "git status", "whoami", "machine", "environment", "system", "os", "platform"]
1244 return any(p in lowered for p in probes)
1246 def _is_capability_request(self, text: str) -> bool:
1247 """Sovereign Detection: Is the user asking about what the agent can do?"""
1248 lowered = text.lower()
1249 probes = ["what can you do", "capabilities", "help", "commands", "tools"]
1250 return any(p in lowered for p in probes)
1252 def _derive_research_flags(self, events: List, files: List, content: str) -> Dict[str, bool]:
1253 """Sovereign Discovery: derive progress flags for research missions."""
1254 lowered = content.lower()
1255 tool_names = [e[0].name for e in events]
1256 return {
1257 "sources_done": any(x in tool_names for x in ["web_search", "site_report"]),
1258 "notes_done": any(x in tool_names for x in ["read_url", "read_file"]),
1259 "synthesis_done": len(content) > 500,
1260 "deliverable_done": len(files) > 0 or "report saved" in lowered
1261 }
1263 def _derive_task_flags(self, events: List, files: List, content: str) -> Dict[str, bool]:
1264 names = [e[0].name for e in events]
1265 return {
1266 "inspected": any(x in names for x in ["read_file", "list_directory", "get_file_tree", "search_codebase", "collect_project_insights"]),
1267 "implemented": any(x in names for x in ["write_file", "replace_file_content", "replace_content", "run_shell_command"]),
1268 "validated": any(x in names for x in ["check_syntax", "lsp_diagnostics", "run_shell_command"]) or "test" in str(names).lower(),
1269 "delivered": len(files) > 0 or "completed" in content.lower()
1270 }
1272 def _build_research_tracker(
1273 self,
1274 user_input: str,
1275 tool_events: List,
1276 content_buffer: str,
1277 files_modified: List[str]
1278 ) -> Optional[RenderableType]:
1279 """High-End Brutalist Research Tracker with progress bars."""
1280 if not self.chat_engine.is_research_request(user_input):
1281 return None
1282 if not self._should_show_tracking(user_input, tool_events, files_modified):
1283 return None
1285 flags = self._derive_research_flags(tool_events, files_modified, content_buffer)
1287 steps = [
1288 ("SOURCES", flags["sources_done"]),
1289 ("EXTRACTION", flags["notes_done"]),
1290 ("SYNTHESIS", flags["synthesis_done"]),
1291 ("DELIVERY", flags["deliverable_done"])
1292 ]
1294 lines = []
1295 for label, done in steps:
1296 style = self.COLORS["success"] if done else "dim white"
1297 icon = "\u2611" if done else "\u2610"
1298 line = Text()
1299 line.append(f" {icon} {label:<15} ", style=style)
1300 bar = "\u2588" * (10 if done else 0) + "\u2591" * (0 if done else 10)
1301 line.append(f"[{bar}]", style=style)
1302 lines.append(line)
1304 return Panel(
1305 Group(*lines),
1306 title="[b] RESEARCH TRACKER [/b]",
1307 title_align="left",
1308 border_style=self.COLORS["primary"],
1309 box=box.HEAVY,
1310 padding=(0, 1)
1311 )
1313 def _build_task_tracker(
1314 self,
1315 user_input: str,
1316 tool_events: List,
1317 content_buffer: str,
1318 files_modified: List[str]
1319 ) -> Optional[RenderableType]:
1320 """High-End Brutalist Task Tracker with progress bars."""
1321 if self.chat_engine.is_research_request(user_input):
1322 return None
1323 if not self._is_task_request(user_input, tool_events):
1324 return None
1325 if not self._should_show_tracking(user_input, tool_events, files_modified):
1326 return None
1328 flags = self._derive_task_flags(tool_events, files_modified, content_buffer)
1330 # 2025 Tier-3: Tactical Progress Logic
1331 steps = [
1332 ("GROUNDING", flags["inspected"]),
1333 ("IMPLEMENTATION", flags["implemented"]),
1334 ("VERIFICATION", flags["validated"]),
1335 ("COMPLETION", flags["delivered"])
1336 ]
1338 lines = []
1339 for label, done in steps:
1340 style = self.COLORS["success"] if done else "dim white"
1341 icon = "\u2611" if done else "\u2610"
1342 line = Text()
1343 line.append(f" {icon} {label:<15} ", style=style)
1344 bar = "\u2588" * (10 if done else 0) + "\u2591" * (0 if done else 10)
1345 line.append(f"[{bar}]", style=style)
1346 lines.append(line)
1348 return Panel(
1349 Group(*lines),
1350 title="[b] MISSION TRACKER [/b]",
1351 title_align="left",
1352 border_style=self.COLORS["primary"],
1353 box=box.HEAVY,
1354 padding=(0, 1)
1355 )
1357 def _is_validation_command(self, command: str) -> bool:
1358 if not command:
1359 return False
1360 tokens = ("pytest", "ruff", "mypy", "pyright", "flake8", "npm test", "pnpm test", "yarn test", "go test", "cargo test")
1361 lowered = command.lower()
1362 return any(token in lowered for token in tokens)
1364 def _build_workings_panel(
1365 self,
1366 user_input: str,
1367 tool_events: List,
1368 files_modified: List[str]
1369 ) -> Optional[RenderableType]:
1370 """High-End Brutalist Workings Overview."""
1371 if not (self.chat_engine.is_research_request(user_input) or self._is_task_request(user_input, tool_events)):
1372 return None
1373 if not self._should_show_tracking(user_input, tool_events, files_modified):
1374 return None
1376 # Stats derivation
1377 tool_names = [e[0].name for e in tool_events]
1378 sensory_count = len([n for n in tool_names if any(x in n.lower() for x in ["read", "list", "tree", "insights", "map"])])
1379 motor_count = len([n for n in tool_names if any(x in n.lower() for x in ["write", "replace", "mkdir", "shell"])])
1381 table = Table.grid(padding=(0, 2))
1382 table.add_column(style="bold white", width=15)
1383 table.add_column()
1385 table.add_row("SENSORY LOAD", Text(f"{sensory_count} reads", style=self.COLORS["primary"]))
1386 table.add_row("MOTOR DRIVE", Text(f"{motor_count} actions", style=self.COLORS["accent"]))
1387 if files_modified:
1388 table.add_row("TARGET FOCUS", Text(", ".join([Path(p).name for p in files_modified[-3:]]), style=self.COLORS["success"]))
1390 return Panel(
1391 table,
1392 title="[b] NEURAL WORKINGS [/b]",
1393 title_align="left",
1394 border_style=self.COLORS["primary"],
1395 box=box.HEAVY,
1396 padding=(0, 1)
1397 )
1399 def _build_plan_panel(self, user_input: str, tool_events: List) -> Optional[RenderableType]:
1400 """High-End Brutalist Plan Tracker with Active Step Highlighting."""
1401 if not (self.chat_engine.is_research_request(user_input) or self._is_task_request(user_input, tool_events)):
1402 return None
1403 if not self._should_show_tracking(user_input, tool_events, []):
1404 return None
1406 # Determine steps and status
1407 if self.chat_engine.is_research_request(user_input):
1408 steps = ["Gather sources", "Extract notes", "Synthesize findings", "Deliver report"]
1409 flags = self._derive_research_flags(tool_events, [], "")
1410 status_map = {
1411 "Gather sources": flags["sources_done"],
1412 "Extract notes": flags["notes_done"],
1413 "Synthesize findings": flags["synthesis_done"],
1414 "Deliver report": flags["deliverable_done"],
1415 }
1416 else:
1417 steps = self._extract_task_steps(user_input)
1418 if not steps:
1419 steps = ["Inspect context", "Implement changes", "Validate", "Deliver output"]
1420 flags = self._derive_task_flags(tool_events, [], "")
1421 status_map = {}
1422 keywords = {
1423 "inspect": flags["inspected"], "review": flags["inspected"], "scan": flags["inspected"], "analyze": flags["inspected"],
1424 "implement": flags["implemented"], "fix": flags["implemented"], "update": flags["implemented"], "refactor": flags["implemented"],
1425 "test": flags["validated"], "validate": flags["validated"], "verify": flags["validated"],
1426 "deliver": flags["delivered"], "document": flags["delivered"], "report": flags["delivered"],
1427 }
1428 for s in steps:
1429 lowered = s.lower()
1430 status_map[s] = any(k in lowered and v for k, v in keywords.items())
1432 lines = []
1433 found_active = False
1435 for step in steps:
1436 done = status_map.get(step, False)
1437 is_active = not done and not found_active
1438 if is_active: found_active = True
1440 style = "bright_green" if done else "bold yellow" if is_active else "dim white"
1441 icon = "\u2713" if done else "\u2794" if is_active else "\u2022"
1443 line = Text()
1444 line.append(f" {icon} ", style=style)
1445 line.append(step, style=style)
1446 lines.append(line)
1448 return Panel(
1449 Group(*lines),
1450 title="[b] TACTICAL PLAN [/b]",
1451 border_style=self.COLORS["warning"],
1452 box=box.HEAVY,
1453 padding=(0, 1)
1454 )
1456 def _build_next_actions_panel(self, gaps: List[str]) -> Optional[RenderableType]:
1457 """High-End Tactical Recommendation Panel."""
1458 if not gaps: return None
1459 suggestions = []
1460 if "validation not run" in gaps: suggestions.append("Run 'lsp_diagnostics' or 'run_tests'")
1461 if "deliverable missing" in gaps: suggestions.append("Use 'write_file' to produce report")
1462 if "no tools used" in gaps: suggestions.append("Ground mission with 'collect_env_grounding'")
1463 if "limited tooling" in gaps: suggestions.append("Explore logic with 'generate_repo_map'")
1465 if not suggestions: return None
1467 table = Table.grid(padding=(0, 1))
1468 table.add_column(style="bold yellow", width=3)
1469 table.add_column()
1470 for s in suggestions[:3]:
1471 table.add_row("\u2794", Text(s, style="white"))
1473 return Panel(
1474 table,
1475 title="[b] TACTICAL PIVOT [/b]",
1476 border_style=self.COLORS["warning"],
1477 box=box.HEAVY,
1478 padding=(0, 1)
1479 )
1481 async def _confirm_tool_execution(self, tool, preview: str) -> bool:
1482 """Sovereign Tool Gate: High-frequency approval prompt."""
1483 # Special style for the prompt
1484 prompt_text = Text()
1485 prompt_text.append("\n \u2794 APPROVE MOTOR ACTION: ", style="bold yellow")
1486 prompt_text.append(tool.name.upper(), style="bold bright_magenta")
1487 prompt_text.append("? (Y/n) ", style="bold white")
1489 choice = Prompt.ask(prompt_text, default="y", choices=["y", "n", "Y", "N"])
1490 return choice.lower().startswith("y")
1492 def _get_bottom_toolbar(self):
1493 """High-end Neural Command Footer."""
1494 branch = self.chat_engine.ctx.branch.upper()
1495 mana = f"{self.chat_engine.tokenizer.count(self.content_buffer):,}" # Simplified for toolbar
1496 load = int(self.hud.context_percent * 100)
1498 # 2025 Best Practice: Context-aware footer
1499 return HTML(
1500 f'<style bg="ansiblue" fg="white"><b> OSIRIS </b></style> '
1501 f'<style fg="ansiwhite"> BRANCH:</style><style fg="ansimagenta"><b>{branch}</b></style> '
1502 f'│ <style fg="ansiwhite">MANA:</style><style fg="ansigreen"><b>{self.chat_engine.tokenizer.count(self.content_buffer)}</b></style> '
1503 f'│ <style fg="ansiwhite">LOAD:</style><style fg="ansiyellow"><b>{load}%</b></style>'
1504 )
1506 def _highlight_quantum_stream(self, raw_text: str) -> Text:
1507 """Apply Chromesthesia highlighting to raw neural throughput with boundary safety."""
1508 # 2025 Hardening: Aggressively strip ALL ANSI escape codes
1509 clean_text = re.sub(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]', '', str(raw_text))
1511 # 2025 Tier-3: Chunked Pulse Engine
1512 # Styling every character creates too many ANSI codes which mangles some terminals.
1513 # We now use a more stable chunk-based highlighting.
1514 highlighted = Text(overflow="crop", no_wrap=True)
1516 # Split into words/tokens to reduce ANSI noise
1517 tokens = re.split(r'(\s+|\W)', clean_text)
1518 for token in tokens:
1519 if not token: continue
1521 style = "white dim"
1522 if any(c in token for c in "{}[]\":"): style = "bold yellow"
1523 elif any(c in token for c in ".,;()"): style = "cyan dim"
1524 elif any(c.isdigit() for c in token): style = "bright_magenta"
1525 elif token == "\u21b5": style = "bold blue"
1526 elif any(c.isalpha() for c in token): style = "cyan dim"
1528 if token.isspace() and token != "\u21b5":
1529 highlighted.append(token)
1530 else:
1531 highlighted.append(token, style=style)
1533 return highlighted
1535 async def handle_message(self, user_input: str):
1536 """Handle message with beautiful display and IMMEDIATE cancellation support"""
1537 # 2025 Sovereign UX: Ensure welcome is shown once
1538 if not self.print_mode:
1539 await self.show_welcome()
1541 user_input = normalize_user_input(user_input)
1542 if not user_input.strip():
1543 return
1545 # State tracking
1546 self.hud.mission_lock = user_input.strip()[:100] # Set MISSION LOCK
1547 self.tool_events: List[Tuple[Any, Optional[float]]] = []
1548 self.turn_files_modified: List[str] = []
1549 self.thought_buffer = ""
1550 self.reflection_buffer = "" # New for Tier-2
1551 from collections import deque
1552 self.quantum_stream = deque(maxlen=1000) # Increased for Waterfall
1553 self.content_buffer = ""
1554 self.tool_trace = []
1555 self.tool_names = []
1556 self.tool_history = []
1557 self.tools_used_turn = 0
1558 self.hud.engine_status = "Neural handshake initiated..."
1560 turn_start = time.time()
1561 response_start: Optional[float] = None
1563 cancelled = False
1564 self._in_stream = False
1565 self._is_prefetching = True # New state for UI
1567 if not self.print_mode:
1568 self._render_user_panel(user_input)
1570 # MAIN RENDER LOOP
1571 def generate_layout() -> Group:
1572 renderables = []
1574 # 0. HUD (Sovereign Dashboard)
1575 stats = cost_tracker.get_session_costs(self.session_id) if self.session_id else {}
1576 tokens = stats.get("total_tokens", 0) or (len(self.content_buffer) // 4)
1578 active_subs = self._get_active_subsystems()
1579 self.hud.update_status(
1580 settings.provider.upper(),
1581 settings.default_model,
1582 tokens,
1583 active=active_subs,
1584 branch=self.chat_engine.ctx.branch
1585 )
1586 self.hud.update_cost(stats.get("total_cost", 0.0))
1588 max_chars = getattr(settings, "max_context_chars", 32000)
1589 context_pct = (tokens * 4) / max_chars if max_chars > 0 else 0
1590 all_mods = list(dict.fromkeys(self.files_modified + self.turn_files_modified))
1591 self.hud.update_context(all_mods, percent=context_pct)
1593 # Update Focus Set
1594 self.hud.active_focus = [Path(p).name for p in self.turn_files_modified[-2:]]
1596 # Dynamic HUD Border Color
1597 hud_border = "bright_blue"
1598 if "NET" in active_subs or "FILES" in active_subs: hud_border = "cyan"
1599 if "SHELL" in active_subs or self.chat_engine.ctx.branch != "main": hud_border = "bright_magenta"
1601 # 2025 Tier-3: High-Resolution Height Calculation
1602 calculated_hud_height = 3
1604 is_compact = getattr(settings, "compact_hud", False) or (self.rich_console.width < 100)
1606 if not is_compact:
1607 if self.hud.mission_lock: calculated_hud_height += 2
1608 if self.hud.active_strategy: calculated_hud_height += 1
1609 if self.hud.active_focus: calculated_hud_height += 1
1610 if self.hud.engine_status: calculated_hud_height += 1
1611 if self.hud.last_trace: calculated_hud_height += 1
1612 if hasattr(self.hud, 'active_swarm') and self.hud.active_swarm: calculated_hud_height += 1
1613 else:
1614 if self.hud.mission_lock: calculated_hud_height += 1
1616 # 1. THE SOVEREIGN HUD (Single Atomic Panel)
1617 # We wrap the raw HUD group in a HEAVY box to prevent "Ghosting"
1618 # Always show HUD for dynamic feedback
1619 renderables.append(Panel(
1620 self.hud.render(compact=is_compact),
1621 box=box.HEAVY,
1622 border_style=hud_border,
1623 title="[b] MISSION: GENERAL LOCK [/b]",
1624 title_align="left",
1625 padding=(0, 1),
1626 height=calculated_hud_height
1627 ))
1629 # 2. THE QUANTUM NEURAL WATERFALL (Restored from Source)
1630 # Displays real-time TPS (Tokens Per Second) and Stream Signal
1631 if self._in_stream and self.quantum_stream:
1632 tps = self.hud.tps
1633 waterfall_frame = f" QUANTUM NEURAL WATERFALL | SIGNAL: {tps:.1f} TPS "
1634 # 2025 Brutalist: Custom ASCII separator
1635 renderables.append(Align.center(
1636 f"[bold cyan]+-{'-'*30} {waterfall_frame} {'-'*30}-+[/]"
1637 ))
1639 # 2. THE QUANTUM NEURAL WATERFALL (Restored from Source)
1640 # Displays real-time TPS (Tokens Per Second) and Stream Signal
1641 if self._in_stream and self.quantum_stream:
1642 tps = self.hud.tps
1643 waterfall_frame = f" QUANTUM NEURAL WATERFALL | SIGNAL: {tps:.1f} TPS "
1644 # 2025 Brutalist: Custom ASCII separator
1645 renderables.append(Align.center(
1646 f"[bold cyan]+-{'-'*30} {waterfall_frame} {'-'*30}-+[/]"
1647 ))
1649 # Always show live panels for dynamic feedback
1650 # 1. Trackers
1651 show_tracking = self._should_show_tracking(user_input, self.tool_events, self.turn_files_modified)
1652 if show_tracking:
1653 tracker = self._build_research_tracker(user_input, self.tool_events, self.content_buffer, self.turn_files_modified)
1654 if tracker: renderables.append(tracker)
1655 task_tracker = self._build_task_tracker(user_input, self.tool_events, self.content_buffer, self.turn_files_modified)
1656 if task_tracker: renderables.append(task_tracker)
1657 plan_panel = self._build_plan_panel(user_input, self.tool_events)
1658 if plan_panel: renderables.append(plan_panel)
1660 # 2. Activity / Workings / Heatmap
1661 if self.tool_events:
1662 workings = self._build_workings_panel(user_input, self.tool_events, self.turn_files_modified)
1663 if workings: renderables.append(workings)
1665 heatmap = self._build_heatmap_panel()
1666 if heatmap: renderables.append(heatmap)
1668 activity_panel = self._build_live_activity_panel(self.tool_events)
1669 if activity_panel: renderables.append(activity_panel)
1671 # 3. Thought & Reflection
1672 if self.thought_buffer.strip():
1673 renderables.append(self._render_thought_panel(self.thought_buffer))
1675 # 3.5 Artifact & Patch Rendering
1676 for tool, _ in self.tool_events:
1677 if tool.name == "create_artifact" and hasattr(tool, "structured_result"):
1678 sr = tool.structured_result
1679 if sr.get("status") == "success":
1680 meta = sr.get("meta", {})
1681 renderables.append(Panel(
1682 Text(f"Artifact '{meta.get('artifact_title')}' created at {meta.get('artifact_path')}", style="bold green"),
1683 title=f"[b] ARTIFACT: {meta.get('artifact_type').upper()} [/b]",
1684 border_style="bright_magenta",
1685 box=box.DOUBLE,
1686 padding=(0, 1)
1687 ))
1689 elif tool.name == "replace_content" and tool.status == ToolCallStatus.SUCCESS:
1690 diff = getattr(tool, "result", "")
1691 if "---" in str(diff):
1692 renderables.append(Panel(
1693 Syntax(str(diff), "diff", theme="monokai"),
1694 title="[b] SURGICAL PATCH APPLIED [/b]",
1695 border_style="bright_green",
1696 box=box.HEAVY,
1697 padding=(0, 1)
1698 ))
1700 # 4. Assistant Response (The Neural Stream)
1701 # Renders actual content inside waterfall flow
1702 elapsed = time.time() - (response_start or turn_start)
1703 is_thinking = (self._in_stream or self._is_prefetching) and not self.content_buffer.strip()
1705 response_panel = self._render_response_panel(
1706 self.content_buffer,
1707 thinking=is_thinking,
1708 elapsed=elapsed,
1709 footer_hint=self._stream_footer_hint() if (self._in_stream or self._is_prefetching) else None
1710 )
1711 # Restore DOUBLE box for response as requested
1712 response_panel.box = box.DOUBLE
1713 response_panel.border_style = "cyan" if self._in_stream else "dim"
1714 renderables.append(response_panel)
1716 # 2025 Restoration: Return raw Group to CLI for atomic outer panel wrapping
1717 return Group(*renderables)
1719 try:
1720 # 2025 Hardening: Use transient=True to prevent 'waterfall' artifacts during active neural stream.
1721 # The final state will be printed permanently after the block exits.
1722 with Live(None, console=self.rich_console, refresh_per_second=8, transient=True, auto_refresh=False) as live:
1723 self.live = live
1724 live.update(generate_layout(), refresh=True)
1726 # PHASE 1: Prefetch / Autonomous Context Gathering
1727 # 2025 Speculative Consumption
1728 if self._prefetch_task and not self._prefetch_task.done():
1729 self.hud.engine_status = "Awaiting speculative prefetch..."
1730 live.update(generate_layout(), refresh=True)
1731 prefetch_context = await self._prefetch_task
1732 elif self._prefetch_task and self._prefetch_task.done():
1733 prefetch_context = self._prefetch_task.result()
1734 else:
1735 prefetch_context = await self._run_prefetch_logic(user_input)
1737 # Clear task after consumption
1738 self._prefetch_task = None
1740 live.update(generate_layout(), refresh=True)
1742 if self.pending_shell_context:
1743 shell_block = "\n\n".join(self.pending_shell_context[-3:])
1744 prefetch_context = f"{prefetch_context}\n\n# Shell Context\n{shell_block}" if prefetch_context else f"# Shell Context\n{shell_block}"
1745 self.pending_shell_context.clear()
1747 file_context = await self._async_handle_path_mentions(user_input)
1748 if file_context:
1749 prefetch_context = f"{prefetch_context}\n\n{file_context}" if prefetch_context else file_context
1751 live.update(generate_layout(), refresh=True)
1752 self._is_prefetching = False
1754 # Autonomous Plan (if any)
1755 self.hud.active_mission = self.chat_engine._select_optimal_persona(user_input).upper()
1756 plan = await self._maybe_create_autonomous_plan(user_input)
1757 if plan:
1758 next_step = plan_mode.get_next_step()
1759 if next_step:
1760 plan_mode.start_step(next_step.id)
1761 await self._attempt_plan_tool_action(next_step.description)
1763 # PHASE 2: Main LLM Loop
1764 full_system_context = f"{self.system_context}\n\n{prefetch_context}" if prefetch_context else self.system_context
1766 async for event in self.chat_engine.send_message(user_input, system_context=full_system_context):
1767 if cancelled: break
1769 event_type = event["type"]
1770 if event_type == "state_change":
1771 if event["state"] == StreamingState.RESPONDING:
1772 self._in_stream = True
1773 if not response_start: response_start = time.time()
1774 if not self.print_mode and not getattr(self, 'prompt_mode', False): self._start_follow_up_listener()
1775 elif event["state"] == StreamingState.IDLE:
1776 self._in_stream = False
1777 await self._stop_follow_up_listener()
1778 elif event_type == "budget_update":
1779 self.hud.update_budget(
1780 used=event.get("used", 0),
1781 total=event.get("limit", 30),
1782 progress=event.get("progress", 0),
1783 strategy=event.get("strategy", "")
1784 )
1786 elif event_type == "status":
1787 self.hud.engine_status = event["message"]
1788 live.update(generate_layout(), refresh=True)
1790 elif event_type == "engine_notification":
1791 # Pulse background events in the HUD
1792 self.hud.engine_status = f"⚡ {event['message']}"
1793 live.update(generate_layout(), refresh=True)
1795 elif event_type == "thought":
1796 self.thought_buffer += event["text"]
1798 elif event_type == "tool_call":
1799 tool = event["tool"]
1800 # TUI Handover: Pause Live for sub-agents to stream output
1801 if tool.name == "delegate_to_subagent":
1802 live.stop()
1804 # 2025 Tier-3: Logic-Gate Flash
1805 # Briefly shift border to signal 'Action'
1806 self.hud.engine_status = f"➔ EXEC: {tool.name.upper()}"
1807 live.update(generate_layout(), refresh=True)
1808 await asyncio.sleep(0.05) # Visual pulse
1810 live.update(generate_layout(), refresh=True)
1812 elif event_type == "cycle_start":
1813 if self.content_buffer.strip():
1814 self.content_buffer += "\n\n---\n\n" # Cycle separator
1816 elif event_type == "tps_update":
1817 self.hud.tps = event["tps"]
1818 live.update(generate_layout(), refresh=True)
1820 elif event_type == "content":
1821 if not response_start:
1822 response_start = time.time()
1823 self.hud.last_latency = (response_start - turn_start) * 1000
1825 text = event["text"]
1827 # 2025 Hardening: Strip ALL potential formatting before quantum highlighting
1828 if hasattr(text, "plain"):
1829 clean_text_for_stream = text.plain
1830 else:
1831 clean_text_for_stream = re.sub(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]', '', str(text))
1833 # Neural Stream for visualization (with Noise Cancellation)
1834 if "🧹" not in clean_text_for_stream and "♻️" not in clean_text_for_stream:
1835 for char in clean_text_for_stream:
1836 # 2025: Density Control - Skip consecutive spaces
1837 if char == " " and self.quantum_stream and self.quantum_stream[-1] == " ":
1838 continue
1839 self.quantum_stream.append(char)
1841 # Real-time Trace for HUD
1842 self.hud.last_trace = (self.hud.last_trace + text.replace("\n", " "))[-40:]
1844 if hasattr(self, "_in_reflection") and self._in_reflection:
1845 self.reflection_buffer += text
1846 if "</reflection>" in text or "[/REFLECTION]" in text:
1847 self._in_reflection = False
1848 else:
1849 self.content_buffer += text
1851 elif event_type == "reflection_start":
1852 self._in_reflection = True
1853 live.update(generate_layout(), refresh=True)
1854 elif event_type == "tool_preview":
1855 tool, preview = event["tool"], event.get("preview", "")
1856 if event.get("requires_confirmation", True):
1857 live.stop()
1858 self._print_section(f"Preview • {tool.name}", Syntax(preview, "text"), self.COLORS["warning"])
1859 await self._confirm_tool_execution(tool, preview)
1860 live.start()
1861 elif event_type == "tool_executing":
1862 tool = event["tool"]
1863 self.tool_events.append((tool, 0.0))
1865 # Trigger HUD Heartbeat
1866 active = self._get_active_subsystems()
1867 if active: self.hud.last_module = list(active)[0]
1869 live.update(generate_layout(), refresh=True)
1871 elif event_type == "tool_result":
1872 tool = event["tool"]
1873 if tool.status == ToolCallStatus.ERROR:
1874 self.chat_engine.plan_needs_revalidation = True
1876 # Find and update existing event if it was 'executing'
1877 found = False
1878 for i in range(len(self.tool_events) - 1, -1, -1):
1879 et, _ = self.tool_events[i]
1880 if et.id == tool.id:
1881 self.tool_events[i] = (tool, 0.0) # Update with result
1882 found = True
1883 break
1885 if not found:
1886 self.tool_events.append((tool, 0.0))
1888 self.tools_used_turn += 1
1889 self.tools_used += 1
1890 self.tool_names.append(tool.name)
1892 # TUI Handover: Resume Live after delegation
1893 if tool.name == "delegate_to_subagent" and not live.is_started:
1894 live.start()
1896 if tool.status == ToolCallStatus.SUCCESS:
1897 self.tool_trace.append({"tool": tool.name, "status": "success"})
1898 self.tool_history.append((tool.name, 0.0, True))
1899 if tool.name in ["write_file", "replace_file_content", "replace_content"]:
1900 path = tool.arguments.get("path")
1901 if path:
1902 self.turn_files_modified.append(path)
1903 self.files_modified.append(path)
1904 self.file_heat[path] += 1 # Record HEAT
1905 else:
1906 self.tool_trace.append({"tool": tool.name, "status": "error", "error": tool.error})
1907 self.tool_history.append((tool.name, 0.0, False))
1908 elif event_type == "error":
1909 live.console.print(f"[{self.COLORS['error']}]Error: {event['message']}[/]")
1911 # Final UI update
1912 live.update(generate_layout(), refresh=True)
1914 # 2025 Sovereign UX: Final permanent render after Live block exits
1915 # This keeps the last state in terminal history since Live was transient.
1916 self.rich_console.print(generate_layout())
1918 # --- DRAIN FOLLOW-UP QUEUE ---
1919 await self._drain_follow_up_queue()
1921 except Exception as e:
1922 console.print(f"\n[{self.COLORS['error']}]✗ Error: {e}[/{self.COLORS['error']}]")
1923 finally:
1924 # 2025 Hardening: Always stop follow-up listener to prevent session reuse errors
1925 self._in_stream = False
1926 await self._stop_follow_up_listener()
1928 async def handle_slash_command(self, command: str) -> bool:
1929 """Handle slash commands with full menu system"""
1930 parts = command.split(maxsplit=1)
1931 cmd = parts[0].lower()
1932 args = parts[1] if len(parts) > 1 else ""
1934 if cmd == "/help":
1935 self.show_help()
1936 return True
1938 elif cmd in ["/exit", "/quit"]:
1939 self.show_session_summary()
1940 self.running = False
1941 return True
1943 elif cmd == "/clear":
1944 await self.show_welcome()
1945 return True
1947 elif cmd == "/reset":
1948 self.chat_engine.reset()
1949 plan_mode.cancel_plan()
1950 console.print(f"[{self.COLORS['success']}]✓ Conversation reset[/{self.COLORS['success']}]\n")
1951 return True
1953 elif cmd == "/summary":
1954 self.show_session_summary()
1955 return True
1957 elif cmd == "/memory":
1958 if args.strip().lower() == "edit":
1959 self.edit_memory_cards()
1960 else:
1961 self.show_memory()
1962 return True
1964 elif cmd == "/tools":
1965 self.show_tools()
1966 return True
1968 elif cmd == "/version":
1969 console.print(f"\n[{self.COLORS['primary']}]Osiris CLI v5.2[/{self.COLORS['primary']}] - Autonomous AI Terminal\n")
1970 return True
1972 elif cmd == "/yolo":
1973 settings.yolo_mode = not settings.yolo_mode
1974 settings.save()
1975 status = "enabled" if settings.yolo_mode else "disabled"
1976 console.print(f"[{self.COLORS['success']}]✓ YOLO mode {status}[/{self.COLORS['success']}]\n")
1977 return True
1979 elif cmd == "/shortcuts":
1980 self.show_shortcuts()
1981 return True
1983 elif cmd == "/system":
1984 if args:
1985 # Apply system prompt
1986 from .system_prompts import get_system_prompt, apply_system_prompt
1988 prompt_id = args.strip()
1989 prompt_data = get_system_prompt(prompt_id)
1991 if prompt_id in ["general", "codex", "review", "debug", "devops", "data", "writer", "architect"]:
1992 # Apply to memory
1993 apply_system_prompt(prompt_id, self.memory)
1995 # Update system context
1996 self.system_context = prompt_data["prompt"]
1998 # Save memory
1999 memory_file = Path.home() / ".osiris" / "agents" / f"{self.agent_name}.json"
2000 memory_file.parent.mkdir(parents=True, exist_ok=True)
2001 import json
2002 with open(memory_file, 'w') as f:
2003 json.dump(self.memory, f, indent=2)
2005 console.print(f"\n[{self.COLORS['success']}]✓ Applied system prompt: {prompt_data['name']}[/{self.COLORS['success']}]")
2006 console.print(f"[{self.COLORS['muted']}]Skills: {', '.join(prompt_data['skills'])}[/{self.COLORS['muted']}]\n")
2007 else:
2008 console.print(f"\n[{self.COLORS['error']}]Unknown system prompt: {prompt_id}[/{self.COLORS['error']}]")
2009 console.print(f"[{self.COLORS['muted']}]Use /system to see available prompts[/{self.COLORS['muted']}]\n")
2010 else:
2011 # Show available prompts
2012 self.show_system_prompts()
2013 return True
2015 elif cmd == "/provider":
2016 from .config import KNOWN_PROVIDERS
2018 providers = list(KNOWN_PROVIDERS.keys())
2019 console.print(f"\n[{self.COLORS['primary']}]Current provider:[/{self.COLORS['primary']}] {settings.provider}")
2020 console.print(f"[{self.COLORS['muted']}]Available providers:[/{self.COLORS['muted']}]")
2021 for idx, provider_id in enumerate(providers, 1):
2022 name = KNOWN_PROVIDERS[provider_id]["name"]
2023 console.print(f" {idx}. {name} ({provider_id})")
2025 choice = args.strip() or Prompt.ask("\nChoose provider (name or number)", default=settings.provider)
2026 if not choice:
2027 console.print()
2028 return True
2030 provider_id = choice.lower()
2031 if provider_id.isdigit():
2032 selection = int(provider_id) - 1
2033 if 0 <= selection < len(providers):
2034 provider_id = providers[selection]
2035 else:
2036 console.print(f"\n[{self.COLORS['error']}]Invalid provider selection[/{self.COLORS['error']}]\n")
2037 return True
2039 if provider_id not in KNOWN_PROVIDERS:
2040 console.print(f"\n[{self.COLORS['error']}]Unknown provider: {provider_id}[/{self.COLORS['error']}]\n")
2041 return True
2043 settings.provider = provider_id
2044 settings.provider_base_url = None
2046 models = KNOWN_PROVIDERS[provider_id].get("models", [])
2047 if models and settings.default_model not in models:
2048 settings.default_model = models[0]
2050 settings.save()
2052 try:
2053 get_async_client().reinitialize()
2054 except Exception:
2055 pass
2056 try:
2057 from .client import get_client
2058 get_client().reinitialize()
2059 except Exception:
2060 pass
2062 console.print(f"\n[{self.COLORS['success']}]✓ Provider set to {provider_id}[/{self.COLORS['success']}]")
2063 if models:
2064 console.print(f"[{self.COLORS['muted']}]Default model: {settings.default_model}[/{self.COLORS['muted']}]")
2066 api_key_name = KNOWN_PROVIDERS[provider_id].get("api_key_name")
2067 if api_key_name and not getattr(settings, api_key_name, None):
2068 env_var = api_key_name.upper()
2069 console.print(f"[{self.COLORS['warning']}]API key not set: {env_var}[/{self.COLORS['warning']}]")
2070 set_key = Prompt.ask("Set API key now? (y/N)", default="n").strip().lower()
2071 if set_key.startswith("y"):
2072 api_key = Prompt.ask("Enter API key", password=True).strip()
2073 if api_key:
2074 setattr(settings, api_key_name, SecretStr(api_key))
2075 settings.save()
2076 try:
2077 get_async_client().reinitialize()
2078 except Exception:
2079 pass
2080 console.print(f"[{self.COLORS['success']}]✓ API key saved[/{self.COLORS['success']}]")
2081 console.print()
2082 return True
2084 elif cmd == "/model":
2085 from .config import KNOWN_PROVIDERS
2086 from .client import fetch_models_dynamic
2088 console.print(f"\n[{self.COLORS['primary']}]Current model:[/{self.COLORS['primary']}] {settings.default_model}")
2089 models = []
2090 provider_info = KNOWN_PROVIDERS.get(settings.provider, {})
2091 api_key_name = provider_info.get("api_key_name")
2092 api_key = None
2093 if api_key_name:
2094 key_value = getattr(settings, api_key_name, None)
2095 if key_value and hasattr(key_value, "get_secret_value"):
2096 api_key = key_value.get_secret_value()
2097 else:
2098 api_key = key_value
2100 try:
2101 console.print(f"[{self.COLORS['muted']}]Fetching available models...[/{self.COLORS['muted']}]")
2102 models = await fetch_models_dynamic(settings.provider, api_key=api_key)
2103 except Exception:
2104 models = provider_info.get("models", [])
2106 if models:
2107 console.print(f"[{self.COLORS['muted']}]Available models:[/{self.COLORS['muted']}]")
2108 for idx, model_name in enumerate(models, 1):
2109 console.print(f" {idx}. {model_name}")
2110 choice = args.strip() or Prompt.ask("\nChoose model (name or number)", default=settings.default_model)
2111 else:
2112 choice = args.strip() or Prompt.ask("\nEnter model name", default=settings.default_model)
2114 if not choice:
2115 console.print()
2116 return True
2118 model_name = choice.strip()
2119 if model_name.isdigit() and models:
2120 selection = int(model_name) - 1
2121 if 0 <= selection < len(models):
2122 model_name = models[selection]
2123 else:
2124 console.print(f"\n[{self.COLORS['error']}]Invalid model selection[/{self.COLORS['error']}]\n")
2125 return True
2126 elif models and model_name not in models:
2127 proceed = Prompt.ask("Model not in list. Use anyway? (y/N)", default="n").strip().lower()
2128 if not proceed.startswith("y"):
2129 console.print()
2130 return True
2132 settings.default_model = model_name
2133 settings.save()
2135 console.print(f"\n[{self.COLORS['success']}]✓ Model set to {model_name}[/{self.COLORS['success']}]\n")
2136 return True
2138 elif cmd == "/setup":
2139 from .setup import show_setup_menu
2140 await show_setup_menu()
2141 return True
2143 elif cmd == "/status":
2144 from .config import KNOWN_PROVIDERS
2146 provider = settings.provider
2147 api_key_name = KNOWN_PROVIDERS.get(provider, {}).get("api_key_name")
2148 api_key_set = bool(api_key_name and getattr(settings, api_key_name, None))
2150 console.print(f"\n[{self.COLORS['primary']}]Provider:[/{self.COLORS['primary']}] {provider}")
2151 console.print(f"[{self.COLORS['primary']}]Model:[/{self.COLORS['primary']}] {settings.default_model}")
2152 if api_key_name:
2153 env_var = api_key_name.upper()
2154 status = "set" if api_key_set else "missing"
2155 console.print(f"[{self.COLORS['primary']}]API key:[/{self.COLORS['primary']}] {env_var} ({status})")
2156 console.print()
2157 return True
2159 elif cmd == "/history":
2160 conversation = self.chat_engine.get_conversation()
2161 if not conversation:
2162 console.print(f"[{self.COLORS['warning']}]No conversation history yet[/{self.COLORS['warning']}]\n")
2163 else:
2164 console.print(f"\n[bold {self.COLORS['accent']}]Conversation History ({len(conversation)} messages)[/bold {self.COLORS['accent']}]\n")
2165 for i, msg in enumerate(conversation[-10:], 1):
2166 role = msg.get("role", "unknown")
2167 content = str(msg.get("content", ""))[:80]
2168 console.print(f" {i}. [{role}] {content}...")
2169 console.print()
2170 return True
2172 elif cmd == "/context":
2173 osiris_md = Path.cwd() / "OSIRIS.md"
2174 if osiris_md.exists():
2175 console.print(f"\n[{self.COLORS['success']}]✓ OSIRIS.md found[/{self.COLORS['success']}]")
2176 console.print(f"[{self.COLORS['muted']}] {osiris_md}[/{self.COLORS['muted']}]\n")
2177 else:
2178 console.print(f"\n[{self.COLORS['warning']}]No OSIRIS.md found in current directory[/{self.COLORS['warning']}]")
2179 console.print(f"[{self.COLORS['muted']}]Tip: Create OSIRIS.md to provide project context[/{self.COLORS['muted']}]\n")
2180 self._print_section(
2181 title="Context",
2182 body=self._build_context_budget(""),
2183 style=self.COLORS["muted"],
2184 )
2185 return True
2187 elif cmd == "/remember":
2188 if not args:
2189 console.print(f"\n[{self.COLORS['warning']}]Usage: /remember key:value[/{self.COLORS['warning']}]")
2190 console.print(f"[{self.COLORS['muted']}]Example: /remember project_type:CLI application[/{self.COLORS['muted']}]\n")
2191 else:
2192 if ":" in args:
2193 key, value = args.split(":", 1)
2194 key = key.strip()
2195 value = value.strip()
2197 if "core_memory" not in self.memory:
2198 self.memory["core_memory"] = {}
2200 self.memory["core_memory"][key] = value
2202 # Save to file
2203 memory_file = Path.home() / ".osiris" / "agents" / f"{self.agent_name}.json"
2204 memory_file.parent.mkdir(parents=True, exist_ok=True)
2205 import json
2206 with open(memory_file, 'w') as f:
2207 json.dump(self.memory, f, indent=2)
2209 console.print(f"\n[{self.COLORS['success']}]✓ Remembered: {key} = {value}[/{self.COLORS['success']}]\n")
2210 else:
2211 console.print(f"\n[{self.COLORS['error']}]Invalid format. Use key:value[/{self.COLORS['error']}]\n")
2212 return True
2214 elif cmd == "/forget":
2215 if not args:
2216 console.print(f"\n[{self.COLORS['warning']}]Usage: /forget key[/{self.COLORS['warning']}]")
2217 console.print(f"[{self.COLORS['muted']}]Example: /forget project_type[/{self.COLORS['muted']}]\n")
2218 else:
2219 key = args.strip()
2220 if "core_memory" in self.memory and key in self.memory["core_memory"]:
2221 del self.memory["core_memory"][key]
2223 # Save to file
2224 memory_file = Path.home() / ".osiris" / "agents" / f"{self.agent_name}.json"
2225 import json
2226 with open(memory_file, 'w') as f:
2227 json.dump(self.memory, f, indent=2)
2229 console.print(f"\n[{self.COLORS['success']}]✓ Forgot: {key}[/{self.COLORS['success']}]\n")
2230 else:
2231 console.print(f"\n[{self.COLORS['warning']}]Key '{key}' not found in memory[/{self.COLORS['warning']}]\n")
2232 return True
2234 elif cmd == "/skill":
2235 if not args:
2236 console.print(f"\n[{self.COLORS['warning']}]Usage: /skill skill_name[/{self.COLORS['warning']}]")
2237 console.print(f"[{self.COLORS['muted']}]Example: /skill python debugging[/{self.COLORS['muted']}]\n")
2238 else:
2239 skill = args.strip()
2241 if "skills" not in self.memory:
2242 self.memory["skills"] = []
2244 if skill not in self.memory["skills"]:
2245 self.memory["skills"].append(skill)
2247 # Save to file
2248 memory_file = Path.home() / ".osiris" / "agents" / f"{self.agent_name}.json"
2249 memory_file.parent.mkdir(parents=True, exist_ok=True)
2250 import json
2251 with open(memory_file, 'w') as f:
2252 json.dump(self.memory, f, indent=2)
2254 console.print(f"\n[{self.COLORS['success']}]✓ Learned skill: {skill}[/{self.COLORS['success']}]\n")
2255 else:
2256 console.print(f"\n[{self.COLORS['warning']}]Already know: {skill}[/{self.COLORS['warning']}]\n")
2257 return True
2259 elif cmd == "/session":
2260 console.print(f"\n[bold {self.COLORS['accent']}]Session Management[/bold {self.COLORS['accent']}]\n")
2261 console.print(f" • [bold {self.COLORS['primary']}]/summary[/bold {self.COLORS['primary']}] - Show current session summary")
2262 console.print(f" • [bold {self.COLORS['primary']}]/reset[/bold {self.COLORS['primary']}] - Reset conversation (keep memory)")
2263 console.print(f" • [bold {self.COLORS['primary']}]/clear[/bold {self.COLORS['primary']}] - Clear screen")
2264 console.print(f" • [bold {self.COLORS['primary']}]/exit[/bold {self.COLORS['primary']}] - Exit and save session\n")
2265 return True
2267 elif cmd == "/profile":
2268 parts = args.split()
2269 if not parts:
2270 profiles = list_profiles()
2271 console.print()
2272 if not profiles:
2273 console.print(f"[{self.COLORS['warning']}]No profiles saved yet[/{self.COLORS['warning']}]\n")
2274 else:
2275 console.print(f"[bold {self.COLORS['accent']}]Profiles[/bold {self.COLORS['accent']}]")
2276 for name in profiles:
2277 console.print(f" • [{self.COLORS['primary']}]{name}[/{self.COLORS['primary']}]")
2278 console.print()
2279 console.print(f"[{self.COLORS['muted']}]Usage: /profile save <name> | /profile load <name> | /profile delete <name>[/{self.COLORS['muted']}]\n")
2280 return True
2281 action = parts[0].lower()
2282 name = " ".join(parts[1:]).strip()
2283 if action == "save" and name:
2284 save_profile(name, settings)
2285 console.print(f"\n[{self.COLORS['success']}]✓ Saved profile '{name}'[/{self.COLORS['success']}]\n")
2286 elif action == "load" and name:
2287 if load_profile(name, settings):
2288 get_async_client().reinitialize()
2289 console.print(f"\n[{self.COLORS['success']}]✓ Loaded profile '{name}'[/{self.COLORS['success']}]\n")
2290 else:
2291 console.print(f"\n[{self.COLORS['error']}]Profile '{name}' not found[/{self.COLORS['error']}]\n")
2292 elif action == "delete" and name:
2293 if delete_profile(name):
2294 console.print(f"\n[{self.COLORS['success']}]✓ Deleted profile '{name}'[/{self.COLORS['success']}]\n")
2295 else:
2296 console.print(f"\n[{self.COLORS['error']}]Profile '{name}' not found[/{self.COLORS['error']}]\n")
2297 else:
2298 console.print(f"\n[{self.COLORS['warning']}]Usage: /profile save <name> | /profile load <name> | /profile delete <name>[/{self.COLORS['warning']}]\n")
2299 return True
2301 elif cmd == "/export":
2302 self.export_session(args.strip())
2303 return True
2305 elif cmd == "/usage":
2306 self.show_usage()
2307 return True
2309 elif cmd == "/tracking":
2310 if not args:
2311 self.show_tracking_settings()
2312 console.print(f"[{self.COLORS['muted']}]Usage: /tracking <minimal|standard|verbose> [min_tools] [min_steps][/{self.COLORS['muted']}]\n")
2313 return True
2314 parts = args.split()
2315 mode = parts[0].lower()
2316 if mode not in {"minimal", "standard", "verbose"}:
2317 console.print(f"\n[{self.COLORS['warning']}]Invalid mode. Use minimal|standard|verbose.[/{self.COLORS['warning']}]\n")
2318 return True
2319 settings.tracking_mode = mode
2320 if len(parts) > 1:
2321 try:
2322 settings.tracking_min_tools = max(1, int(parts[1]))
2323 except ValueError:
2324 console.print(f"\n[{self.COLORS['warning']}]Invalid min_tools value.[/{self.COLORS['warning']}]\n")
2325 if len(parts) > 2:
2326 try:
2327 settings.tracking_min_steps = max(1, int(parts[2]))
2328 except ValueError:
2329 console.print(f"\n[{self.COLORS['warning']}]Invalid min_steps value.[/{self.COLORS['warning']}]\n")
2330 settings.save()
2331 self.show_tracking_settings()
2332 return True
2334 elif cmd == "/report":
2335 self.generate_site_report(args.strip())
2336 return True
2338 elif cmd == "/activity":
2339 activity_monitor.show_summary()
2340 return True
2342 elif cmd == "/details":
2343 self.show_details = not self.show_details
2344 state = "visible" if self.show_details else "hidden"
2345 console.print(f"\n[{self.COLORS['muted']}]Details panel is now {state}.[/{self.COLORS['muted']}]\n")
2346 if self.show_details:
2347 self._render_details_panel()
2348 return True
2350 elif cmd == "/demo":
2351 self._run_demo_sequence()
2352 return True
2354 elif cmd == "/plan":
2355 parts = args.split(maxsplit=1)
2356 action = parts[0].lower() if parts else "status"
2357 payload = parts[1].strip() if len(parts) > 1 else ""
2358 if action in {"status", ""}:
2359 plan_mode.display_plan()
2360 self._render_plan_summary()
2361 return True
2362 if action == "cancel":
2363 if plan_mode.cancel_plan():
2364 console.print(f"[{self.COLORS['success']}]✓ Plan cancelled[/{self.COLORS['success']}]\n")
2365 else:
2366 console.print(f"[{self.COLORS['muted']}]No active plan to cancel.[/{self.COLORS['muted']}]\n")
2367 return True
2368 if action == "list":
2369 plans = plan_mode.list_plans()
2370 if not plans:
2371 console.print(f"[{self.COLORS['muted']}]No saved plans yet.[/{self.COLORS['muted']}]\n")
2372 return True
2373 console.print(f"[{self.COLORS['primary']}]Saved Plans[/]")
2374 for plan in plans[:5]:
2375 console.print(f" • {plan['id']} ({plan['status']}) - {plan['title']} [{plan['steps']} steps]")
2376 console.print()
2377 return True
2378 if action == "create":
2379 if not payload or "|" not in payload:
2380 console.print(f"\n[{self.COLORS['warning']}]Usage: /plan create Title | Description | Step1 | Step2 ...[/{self.COLORS['warning']}]\n")
2381 return True
2382 sections = [s.strip() for s in payload.split("|")]
2383 title = sections[0]
2384 description = sections[1] if len(sections) > 1 else "Plan Mode"
2385 steps_raw = sections[2:] if len(sections) > 2 else []
2386 steps = []
2387 for chunk in steps_raw:
2388 steps.extend([part.strip() for part in re.split(r"[;\n]", chunk) if part.strip()])
2389 if not steps:
2390 console.print(f"[{self.COLORS['warning']}]Please provide at least one step.[/{self.COLORS['warning']}]\n")
2391 return True
2392 plan = plan_mode.create_plan(title, description, steps)
2393 plan_mode.display_plan(plan)
2394 self._render_plan_summary()
2395 next_step = plan_mode.get_next_step()
2396 if next_step:
2397 plan_mode.start_step(next_step.id)
2398 self._attempt_plan_tool_action(next_step.description)
2399 return True
2400 if action in {"next", "continue"}:
2401 step = plan_mode.get_next_step()
2402 if not step:
2403 console.print(f"\n[{self.COLORS['muted']}]No pending steps to start.[/{self.COLORS['muted']}]\n")
2404 return True
2405 plan_mode.start_step(step.id)
2406 console.print(f"[{self.COLORS['success']}]Started step {step.id}: {step.description}[/{self.COLORS['success']}]\n")
2407 self._attempt_plan_tool_action(step.description)
2408 return True
2409 if action == "complete":
2410 if not plan_mode.current_plan:
2411 console.print(f"[{self.COLORS['warning']}]No plan loaded.[/{self.COLORS['warning']}]\n")
2412 return True
2413 step_num = int(payload) if payload.isdigit() else plan_mode.current_plan.current_step + 1
2414 plan_mode.complete_step(step_num)
2415 plan_mode.display_plan()
2416 self._render_plan_summary()
2417 return True
2418 if action == "load":
2419 if not payload:
2420 console.print(f"[{self.COLORS['warning']}]Usage: /plan load <plan_id>[/{self.COLORS['warning']}]\n")
2421 return True
2422 plan = plan_mode.load_plan(payload)
2423 if plan:
2424 plan_mode.display_plan(plan)
2425 self._render_plan_summary()
2426 else:
2427 console.print(f"[{self.COLORS['error']}]Plan not found: {payload}[/{self.COLORS['error']}]\n")
2428 return True
2429 console.print(f"[{self.COLORS['warning']}]Unknown plan action '{action}'. Use list/create/start/complete/load/status.[/{self.COLORS['warning']}]\n")
2430 return True
2432 return False
2434 def show_help(self):
2435 """Beautiful help display with categorized commands"""
2436 console.print()
2438 # Group commands by category
2439 categories = {}
2440 for cmd_info in SLASH_COMMANDS:
2441 cat = cmd_info["category"]
2442 if cat not in categories:
2443 categories[cat] = []
2444 categories[cat].append(cmd_info)
2446 # Display each category
2447 for category, commands in categories.items():
2448 console.print(f"[bold {self.COLORS['accent']}]{category}[/bold {self.COLORS['accent']}]")
2450 table = Table(show_header=False, box=None, padding=(0, 2))
2451 table.add_column(style=f"bold {self.COLORS['primary']}", width=15)
2452 table.add_column(style=self.COLORS["muted"])
2454 for cmd in commands:
2455 table.add_row(cmd["cmd"], cmd["desc"])
2457 console.print(table)
2458 console.print()
2460 def _run_demo_sequence(self):
2461 """Run a safe, read-only demo sequence to showcase capabilities."""
2462 console.print(f"\n[bold {self.COLORS['accent']}]Running Safe Demo[/bold {self.COLORS['accent']}]")
2463 steps = [
2464 ("list_directory", {"path": "."}),
2465 ("get_file_tree", {"path": ".", "max_depth": 1}),
2466 ]
2468 for name, args in steps:
2469 console.print(f"\n[{self.COLORS['muted']}]⧗ {name}[/] with {json.dumps(args)}")
2470 try:
2471 result = tools_registry.execute(name, args)
2472 except Exception as exc:
2473 console.print(Panel(f"[bold red]Error:[/bold red] {exc}", title=f"Demo • {name}", style="red"))
2474 continue
2476 snippet = str(result)
2477 if len(snippet) > 700:
2478 snippet = snippet[:700] + "\n...truncated..."
2480 console.print(Panel(snippet, title=f"Demo • {name}", style=self.COLORS["primary"]))
2482 def _render_plan_summary(self):
2483 plan = plan_mode.current_plan
2484 if not plan or self.print_mode:
2485 return
2487 total = len(plan.steps)
2488 completed = sum(1 for s in plan.steps if s.status == StepStatus.COMPLETED)
2489 percent = (completed / total * 100) if total else 0
2490 next_step = None
2491 if 0 <= plan.current_step < total:
2492 next_step = plan.steps[plan.current_step]
2494 lines = [
2495 f"[bold]{plan.title}[/bold] • Status: {plan.status.upper()}",
2496 f"Progress: {completed}/{total} steps ({percent:.1f}%)",
2497 ]
2498 if next_step:
2499 lines.append(f"Next: {next_step.description}")
2501 console.print(Panel("\n".join(lines), title="Plan Summary", border_style=self.COLORS["accent"]))
2503 def _render_details_panel(self):
2504 console.print("\n[bold]Session Details[/bold]")
2505 if session.activity_log:
2506 table = Table(show_header=True, header_style="bold", box=None)
2507 table.add_column("Time", style="dim", width=20)
2508 table.add_column("Activity")
2509 for entry in session.activity_log[-5:]:
2510 table.add_row(entry["timestamp"].split("T")[1][:8], entry["description"])
2511 console.print(Panel(table, title="Activity Log", border_style=self.COLORS["muted"]))
2512 else:
2513 console.print(Panel("No activity logged yet.", border_style=self.COLORS["muted"]))
2515 if session.plan_history:
2516 lines = []
2517 for entry in session.plan_history[-3:]:
2518 plan_text = entry["plan"].strip().splitlines()[0]
2519 lines.append(f"{entry['timestamp'].split('T')[0]} • {plan_text}")
2520 console.print(Panel("\n".join(lines), title="Plan History", border_style=self.COLORS["accent"]))
2521 else:
2522 console.print(Panel("No plan history recorded.", border_style=self.COLORS["accent"]))
2524 def _render_working_panel(self, status: str):
2525 if self.print_mode:
2526 return
2528 panel_lines = [status, ""]
2529 plan = plan_mode.current_plan
2530 if plan:
2531 panel_lines.append(f"Active Plan: {plan.title} ({plan.status})")
2532 recent = session.activity_log[-3:]
2533 if recent:
2534 panel_lines.append("Recent Activity:")
2535 panel_lines.extend(f"• {entry['description']}" for entry in recent[-3:])
2537 console.print(Panel("\n".join(panel_lines), title="Working / Activity", border_style=self.COLORS["muted"]))
2539 def _render_plan_summary(self):
2540 plan = plan_mode.current_plan
2541 if not plan or self.print_mode:
2542 return
2544 total = len(plan.steps)
2545 completed = sum(1 for s in plan.steps if s.status == StepStatus.COMPLETED)
2546 percent = (completed / total * 100) if total else 0
2547 next_step = None
2548 if 0 <= plan.current_step < total:
2549 next_step = plan.steps[plan.current_step]
2551 lines = [
2552 f"[bold]{plan.title}[/bold] • Status: {plan.status.upper()}",
2553 f"Progress: {completed}/{total} steps ({percent:.1f}%)",
2554 ]
2555 if next_step:
2556 lines.append(f"Next: {next_step.description}")
2558 console.print(Panel("\n".join(lines), title="Plan Summary", border_style=self.COLORS["accent"]))
2560 def _attempt_plan_tool_action(self, description: str):
2561 suggestion = plan_executor.interpret_step(description)
2562 if not suggestion:
2563 return False
2565 tool_name = suggestion["tool"]
2566 args = suggestion["args"]
2567 console.print(Panel(f"[bold]{tool_name}[/bold] • `{json.dumps(args)}`", title="Plan Action", style=self.COLORS["warning"]))
2568 console.print(f"[{self.COLORS['muted']}]Working on plan step → {tool_name}[/{self.COLORS['muted']}]")
2569 session.add_activity(f"Plan queued tool: {tool_name}")
2570 self._render_working_panel(f"Executing plan step: {tool_name}")
2572 if not session.is_tool_allowed(tool_name):
2573 session.add_activity(f"Plan tool blocked by policy: {tool_name}")
2574 console.print(Panel(f"[bold red]Blocked[/bold red] • {tool_name} is forbidden by session policy.", style="red"))
2575 return False
2577 try:
2578 activity_monitor.start_tool(tool_name, args)
2579 result = tools_registry.execute(tool_name, args)
2580 activity_monitor.end_tool(str(result), success=True)
2581 snippet = str(result)
2582 if len(snippet) > 400:
2583 snippet = snippet[:400] + "\n...truncated..."
2584 console.print(Panel(snippet, title=f"Plan auto • {tool_name}", style=self.COLORS["primary"]))
2585 session.add_activity(f"Plan auto-run: {tool_name}")
2586 session.add_tool_output(f"plan-{tool_name}", tool_name, snippet)
2587 self._render_plan_summary()
2588 return True
2589 except Exception as exc:
2590 activity_monitor.end_tool(str(exc), success=False)
2591 console.print(Panel(f"[red]Plan tool failed:[/red] {exc}", style="red"))
2592 session.add_activity(f"Plan tool failed: {tool_name}")
2593 return False
2595 def show_memory(self):
2596 """Display agent memory"""
2597 if not self.memory:
2598 console.print(f"[{self.COLORS['warning']}]No memory stored yet[/{self.COLORS['warning']}]\n")
2599 return
2601 console.print()
2603 memory_table = Table.grid(padding=(0, 2))
2604 memory_table.add_column(style=self.COLORS["muted"])
2605 memory_table.add_column()
2607 if "persona" in self.memory:
2608 memory_table.add_row("Persona", Text(self.memory["persona"][:60], style=self.COLORS["primary"]))
2610 if "skills" in self.memory:
2611 skills = ", ".join(self.memory["skills"][:5])
2612 memory_table.add_row("Skills", Text(skills, style=self.COLORS["accent"]))
2614 if "core_memory" in self.memory:
2615 memory_table.add_row("Facts", Text(f"{len(self.memory['core_memory'])} stored", style=self.COLORS["success"]))
2616 preview_items = list(self.memory["core_memory"].items())[:3]
2617 for key, value in preview_items:
2618 memory_table.add_row("", Text(f"• {key}: {value}", style=self.COLORS["muted"]))
2620 console.print(Panel(
2621 memory_table,
2622 title="[bold]Agent Memory[/bold]",
2623 border_style=self.COLORS["primary"],
2624 padding=(1, 2)
2625 ))
2626 console.print()
2628 def edit_memory_cards(self):
2629 """Edit or add core memory facts"""
2630 if "core_memory" not in self.memory:
2631 self.memory["core_memory"] = {}
2632 core = self.memory["core_memory"]
2633 console.print()
2634 console.print(f"[bold {self.COLORS['accent']}]Memory Cards[/bold {self.COLORS['accent']}]")
2635 if core:
2636 for key, value in list(core.items())[:8]:
2637 console.print(f" • [{self.COLORS['primary']}]{key}[/{self.COLORS['primary']}] = {value}")
2638 else:
2639 console.print(f"[{self.COLORS['muted']}]No memory cards yet.[/{self.COLORS['muted']}]")
2640 console.print()
2641 key = Prompt.ask("Key to edit/add (blank to cancel)", default="").strip()
2642 if not key:
2643 console.print()
2644 return
2645 value = Prompt.ask("Value", default=core.get(key, "")).strip()
2646 core[key] = value
2647 memory_file = Path.home() / ".osiris" / "agents" / f"{self.agent_name}.json"
2648 memory_file.parent.mkdir(parents=True, exist_ok=True)
2649 with open(memory_file, 'w') as f:
2650 json.dump(self.memory, f, indent=2)
2651 console.print(f"\n[{self.COLORS['success']}]✓ Saved memory card '{key}'[/{self.COLORS['success']}]\n")
2653 def show_usage(self):
2654 """Show session cost and token usage"""
2655 stats = cost_tracker.get_session_costs(self.session_id) if self.session_id else {}
2656 total_tokens = stats.get("total_tokens", 0)
2657 total_cost = stats.get("total_cost", 0.0)
2658 providers = ", ".join(stats.get("providers", [])) if stats else "n/a"
2659 models = ", ".join(stats.get("models", [])) if stats else "n/a"
2661 table = Table.grid(padding=(0, 2))
2662 table.add_column(style=self.COLORS["muted"], width=16)
2663 table.add_column()
2664 table.add_row("Session", Text(self.session_id, style=self.COLORS["primary"]))
2665 table.add_row("Tokens", Text(f"{total_tokens:,}", style=self.COLORS["accent"]))
2666 table.add_row("Cost", Text(f"${total_cost:.4f}", style=self.COLORS["success"]))
2667 table.add_row("Providers", Text(providers, style=self.COLORS["muted"]))
2668 table.add_row("Models", Text(models, style=self.COLORS["muted"]))
2670 console.print()
2671 console.print(Panel(
2672 table,
2673 title="[bold]Usage[/bold]",
2674 border_style=self.COLORS["primary"],
2675 padding=(1, 2)
2676 ))
2677 console.print()
2679 def show_tracking_settings(self):
2680 mode = getattr(settings, "tracking_mode", "standard")
2681 min_tools = getattr(settings, "tracking_min_tools", 2)
2682 min_steps = getattr(settings, "tracking_min_steps", 2)
2683 table = Table.grid(padding=(0, 2))
2684 table.add_column(style=self.COLORS["muted"], width=16)
2685 table.add_column()
2686 table.add_row("Mode", Text(mode, style=self.COLORS["primary"]))
2687 table.add_row("Min tools", Text(str(min_tools), style=self.COLORS["accent"]))
2688 table.add_row("Min steps", Text(str(min_steps), style=self.COLORS["accent"]))
2689 console.print()
2690 console.print(Panel(
2691 table,
2692 title="[bold]Tracking[/bold]",
2693 border_style=self.COLORS["primary"],
2694 padding=(1, 2)
2695 ))
2696 console.print()
2698 def export_session(self, args: str):
2699 """Export session to markdown or JSON"""
2700 fmt = "md"
2701 if args:
2702 fmt = args.split()[0].lower()
2703 if fmt not in {"md", "json"}:
2704 console.print(f"\n[{self.COLORS['warning']}]Usage: /export [md|json][/{self.COLORS['warning']}]\n")
2705 return
2707 export_dir = Path.cwd() / "exports"
2708 export_dir.mkdir(parents=True, exist_ok=True)
2709 filename = export_dir / f"osiris-session-{self.session_id}.{fmt}"
2711 conversation = self.chat_engine.get_conversation()
2712 duration_seconds = int(time.time() - self.session_start)
2713 duration_str = f"{duration_seconds // 60}m {duration_seconds % 60}s"
2714 stats = cost_tracker.get_session_costs(self.session_id) if self.session_id else {}
2715 export_data = {
2716 "session_id": self.session_id,
2717 "started_at": datetime.fromtimestamp(self.session_start).isoformat(),
2718 "duration": duration_str,
2719 "provider": settings.provider,
2720 "model": settings.default_model,
2721 "system_prompt": settings.system_prompt,
2722 "conversation": conversation,
2723 "tool_trace": self.tool_trace,
2724 "files_modified": self.files_modified,
2725 "usage": stats,
2726 }
2728 if fmt == "json":
2729 filename.write_text(json.dumps(export_data, indent=2), encoding="utf-8")
2730 else:
2731 lines = [
2732 f"# Osiris Session Export",
2733 f"- Session: `{self.session_id}`",
2734 f"- Provider: `{settings.provider}`",
2735 f"- Model: `{settings.default_model}`",
2736 f"- Duration: `{duration_str}`",
2737 f"- Tokens: `{stats.get('total_tokens', 0)}`",
2738 f"- Cost: `${stats.get('total_cost', 0.0):.4f}`",
2739 "",
2740 "## Conversation",
2741 ]
2742 for msg in conversation:
2743 role = msg.get("role", "unknown").upper()
2744 content = str(msg.get("content", ""))
2745 lines.append(f"### {role}")
2746 lines.append(content)
2747 lines.append("")
2748 if self.tool_trace:
2749 lines.append("## Tool Trace")
2750 for entry in self.tool_trace:
2751 lines.append(f"- {entry['tool']} [{entry['status']}] ({entry['duration']})")
2752 if self.files_modified:
2753 lines.append("")
2754 lines.append("## Files Modified")
2755 for path in self.files_modified:
2756 lines.append(f"- `{path}`")
2757 filename.write_text("\n".join(lines), encoding="utf-8")
2759 console.print(f"\n[{self.COLORS['success']}]✓ Exported session to {filename}[/{self.COLORS['success']}]\n")
2761 def generate_site_report(self, args: str):
2762 """Generate a site report using the site_report tool"""
2763 parts = args.split()
2764 if not parts:
2765 console.print(f"\n[{self.COLORS['warning']}]Usage: /report <url> [max_pages] [max_depth][/{self.COLORS['warning']}]\n")
2766 return
2767 url = parts[0]
2768 max_pages = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 6
2769 max_depth = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 2
2770 report = execute_tool("site_report", {
2771 "url": url,
2772 "max_pages": max_pages,
2773 "max_depth": max_depth,
2774 })
2775 if report.startswith("Error") or report.startswith("No pages"):
2776 console.print(f"\n[{self.COLORS['error']}]✗ {report}[/{self.COLORS['error']}]\n")
2777 return
2778 parsed = urlparse(url if url.startswith(("http://", "https://")) else f"https://{url}")
2779 safe_name = (parsed.netloc or "site_report").replace(".", "_")
2780 report_dir = Path.cwd() / "reports"
2781 report_dir.mkdir(parents=True, exist_ok=True)
2782 filename = report_dir / f"{safe_name}_report.md"
2783 filename.write_text(report, encoding="utf-8")
2784 console.print(f"\n[{self.COLORS['success']}]✓ Report saved to {filename}[/{self.COLORS['success']}]\n")
2786 def show_tools(self):
2787 """Display available tools"""
2788 tools = tools_registry.list_tools()
2790 console.print()
2791 console.print(f"[bold {self.COLORS['accent']}]Available Tools ({len(tools)})[/bold {self.COLORS['accent']}]\n")
2793 for tool in tools[:15]: # Show first 15
2794 tool_text = Text()
2795 tool_text.append(" • ", style=self.COLORS["muted"])
2796 tool_text.append(tool.name, style=f"bold {self.COLORS['primary']}")
2797 tool_text.append(f" - {tool.description[:60]}", style=self.COLORS["muted"])
2798 console.print(tool_text)
2800 if len(tools) > 15:
2801 console.print(f"\n[{self.COLORS['muted']}] ... and {len(tools) - 15} more[/{self.COLORS['muted']}]")
2803 console.print()
2805 def show_session_summary(self):
2806 """Beautiful session summary"""
2807 duration_seconds = int(time.time() - self.session_start)
2808 duration_str = f"{duration_seconds // 60}m {duration_seconds % 60}s"
2809 stats = cost_tracker.get_session_costs(self.session_id) if self.session_id else {}
2811 console.print()
2813 summary_table = Table.grid(padding=(0, 2))
2814 summary_table.add_column(style=self.COLORS["muted"], width=15)
2815 summary_table.add_column()
2817 summary_table.add_row("Session", Text(self.session_id, style=f"bold {self.COLORS['primary']}"))
2818 summary_table.add_row("Duration", Text(duration_str, style=f"bold {self.COLORS['primary']}"))
2819 summary_table.add_row("Tools used", Text(str(self.tools_used), style=self.COLORS["accent"]))
2820 summary_table.add_row("Files modified", Text(str(len(set(self.files_modified))), style=self.COLORS["success"]))
2821 summary_table.add_row("Tokens", Text(f"{stats.get('total_tokens', 0):,}", style=self.COLORS["accent"]))
2822 summary_table.add_row("Cost", Text(f"${stats.get('total_cost', 0.0):.4f}", style=self.COLORS["success"]))
2824 if self.key_actions:
2825 summary_table.add_row("", "")
2826 summary_table.add_row("Key actions", "")
2827 for action in self.key_actions[-5:]:
2828 summary_table.add_row("", Text(f"• {action}", style="dim"))
2830 console.print(Panel(
2831 summary_table,
2832 title="[bold]Session Summary[/bold]",
2833 border_style=self.COLORS["primary"],
2834 padding=(1, 2)
2835 ))
2836 console.print()
2838 def show_shortcuts(self):
2839 """Display keyboard shortcuts"""
2840 console.print()
2842 shortcuts_table = Table(show_header=False, box=None, padding=(0, 2))
2843 shortcuts_table.add_column(style=f"bold {self.COLORS['primary']}", width=12)
2844 shortcuts_table.add_column(style=self.COLORS["muted"])
2846 for shortcut in KEYBOARD_SHORTCUTS:
2847 shortcuts_table.add_row(shortcut["key"], shortcut["action"])
2849 console.print(Panel(
2850 shortcuts_table,
2851 title="[bold]Keyboard Shortcuts[/bold]",
2852 border_style=self.COLORS["primary"],
2853 padding=(1, 2)
2854 ))
2855 console.print()
2857 def show_system_prompts(self):
2858 """Display available system prompts"""
2859 from .system_prompts import list_system_prompts
2861 prompts = list_system_prompts()
2863 console.print()
2864 console.print(f"[bold {self.COLORS['accent']}]Available System Prompts[/bold {self.COLORS['accent']}]\n")
2866 for prompt in prompts:
2867 prompt_text = Text()
2868 prompt_text.append(" • ", style=self.COLORS["muted"])
2869 prompt_text.append(prompt["id"], style=f"bold {self.COLORS['primary']}")
2870 prompt_text.append(f" - {prompt['name']}", style=self.COLORS["accent"])
2871 prompt_text.append(f"\n {prompt['description']}", style="dim")
2872 console.print(prompt_text)
2874 console.print()
2875 console.print(f"[{self.COLORS['muted']}]Usage: /system <prompt_id>[/{self.COLORS['muted']}]")
2876 console.print(f"[{self.COLORS['muted']}]Example: /system codex[/{self.COLORS['muted']}]\n")
2878 def _build_heatmap_panel(self) -> Optional[RenderableType]:
2879 """High-End Brutalist Workspace Heatmap."""
2880 if not self.file_heat: return None
2882 lines = []
2883 top_files = self.file_heat.most_common(5)
2884 for path, heat in top_files:
2885 color = "white"
2886 if heat >= 3: color = "bright_red"
2887 elif heat >= 2: color = "bright_yellow"
2888 line = Text()
2889 bar = "\u2588" * min(heat, 5)
2890 line.append(f" {bar} ", style=color)
2891 line.append(f"{Path(path).name} ", style="bold white")
2892 line.append(f"({heat}x mods)", style="dim")
2893 lines.append(line)
2895 return Panel(
2896 Group(*lines),
2897 title="[b] WORKSPACE HEATMAP [/b]",
2898 border_style="bright_blue",
2899 box=box.HEAVY,
2900 padding=(0, 1)
2901 )
2903 def _build_live_activity_panel(self, tool_events: List[Tuple[Any, float]]) -> Optional[RenderableType]:
2904 """High-End Brutalist Activity Panel with grouped events"""
2905 if not tool_events: return None
2907 # Grouping logic
2908 grouped = []
2909 if not tool_events: return None
2910 curr = [tool_events[0]]
2911 for i in range(1, len(tool_events)):
2912 t, d = tool_events[i]
2913 if t.name == curr[-1][0].name and t.arguments == curr[-1][0].arguments:
2914 curr.append((t, d))
2915 else:
2916 grouped.append(curr)
2917 curr = [(t, d)]
2918 grouped.append(curr)
2920 lines = []
2921 for group in grouped[-6:]:
2922 tool, _ = group[0]
2923 count = len(group)
2925 if hasattr(tool, "status") and tool.status == ToolCallStatus.EXECUTING:
2926 status_icon = "\u280b"
2927 status_color = self.COLORS["accent"]
2928 else:
2929 status_icon = "\u2714" if not hasattr(tool, "status") or tool.status == ToolCallStatus.SUCCESS else "\u2718"
2930 status_color = self.COLORS["success"] if status_icon == "\u2714" else self.COLORS["error"]
2932 line = Text()
2933 line.append(f" {status_icon} ", style=f"bold {status_color}")
2934 line.append(f"{tool.name.upper()}", style=f"bold {self.COLORS['primary']}")
2935 if count > 1:
2936 line.append(f" x{count}", style=f"bold {self.COLORS['warning']}")
2938 # Error feedback
2939 error_msg = getattr(tool, "error", "")
2940 if status_icon == "\u2718" and error_msg:
2941 clean_err = str(error_msg).replace("\n", " ")
2942 if len(clean_err) > 60: clean_err = clean_err[:57] + "..."
2943 line.append(f" \u2298 {clean_err}", style=f"bold {self.COLORS['error']}")
2944 else:
2945 # Extract main argument
2946 arg = ""
2947 if "command" in tool.arguments: arg = tool.arguments["command"]
2948 elif "path" in tool.arguments: arg = tool.arguments["path"]
2949 elif "url" in tool.arguments: arg = tool.arguments["url"]
2950 elif tool.arguments: arg = next(iter(tool.arguments.values()))
2952 if arg:
2953 clean_arg = str(arg).replace("\n", " ")
2954 if len(clean_arg) > 60: clean_arg = clean_arg[:57] + "..."
2955 line.append(f" › {clean_arg}", style=self.COLORS["muted"])
2957 # SENSORY PREVIEW (2025 Hardening)
2958 if status_icon == "\u2714" and any(x in tool.name.lower() for x in ["read", "list", "tree", "insights", "map"]):
2959 res = getattr(tool, "result", "")
2960 if res:
2961 clean_res = str(res).strip().splitlines()[0]
2962 if len(clean_res) > 80: clean_res = clean_res[:77] + "..."
2963 line.append(f"\n \u21b3 {clean_res}", style="dim")
2965 lines.append(line)
2967 return Panel(
2968 Group(*lines),
2969 title="[b] SYSTEM ACTIVITY [/b]",
2970 title_align="left",
2971 border_style=self.COLORS["primary"],
2972 box=box.HEAVY,
2973 padding=(0, 1)
2974 )
2976 def _print_turn_summary(self, start_time: float):
2977 """Final Brutalist Summary at end of turn"""
2978 duration = time.time() - start_time
2979 unique = sorted(set(self.tool_names))
2981 table = Table.grid(padding=(0, 2))
2982 table.add_row(
2983 Text("STATUS", style="bold green"),
2984 Text("COMPLETED", style="bold white")
2985 )
2986 table.add_row(
2987 Text("RUNTIME", style="bold cyan"),
2988 Text(f"{duration:.2f}s", style="white")
2989 )
2990 if unique:
2991 table.add_row(
2992 Text("TOOLS", style="bold magenta"),
2993 Text(", ".join(unique), style="white")
2994 )
2996 console.print(Panel(
2997 table,
2998 title="[b] TURN SUMMARY [/b]",
2999 title_align="left",
3000 border_style="white",
3001 box=box.HEAVY,
3002 expand=False
3003 ))
3005 async def _run_prefetch_logic(self, user_input: str) -> Optional[str]:
3006 """Perform autonomous orientation scans."""
3007 prefetch_parts = []
3009 # 0. System Grounding
3010 if user_input == "AUTO_START":
3011 try:
3012 grounding = await execute_tool("collect_env_grounding", {})
3013 prefetch_parts.append(grounding)
3014 except: pass
3016 # 1. Orientation Scan (Turn 0 or explicit request)
3017 if user_input == "AUTO_START" or self.turn_count == 0 or self._is_env_intent(user_input):
3018 if settings.yolo_mode:
3019 try:
3020 # Run a quick scan to orient the agent
3021 structure = await execute_tool("summarize_repo_structure", {"path": ".", "max_depth": 2})
3022 insights = await execute_tool("collect_project_insights", {"path": "."})
3023 prefetch_parts.append(f"# Project Orientation (Auto-Scan)\n## Structure\n{structure}\n\n## Insights\n{insights}")
3025 # Deep Dive into project configs
3026 for config_file in ["pyproject.toml", "package.json", "requirements.txt"]:
3027 p = Path.cwd() / config_file
3028 if p.exists():
3029 content = await execute_tool("read_file", {"path": config_file})
3030 prefetch_parts.append(f"## Config: {config_file}\n{content}")
3032 # Track orientation as events
3033 tool = SimpleNamespace(name="auto_orientation_scan", arguments={}, status=ToolCallStatus.SUCCESS)
3034 self.tool_events.append((tool, 0.0))
3035 except:
3036 pass
3038 # 1. Research Request (site_report)
3039 if self.chat_engine.is_research_request(user_input):
3040 url = self._extract_first_url(user_input)
3041 if url:
3042 try:
3043 start = time.time()
3044 report = await execute_tool("site_report", {"url": url, "max_pages": 8, "max_depth": 2})
3045 duration = time.time() - start
3047 # Track event for activity display
3048 from types import SimpleNamespace
3049 tool = SimpleNamespace(name="site_report", arguments={"url": url}, status=ToolCallStatus.SUCCESS)
3050 self.tool_events.append((tool, duration))
3051 self.tool_names.append("site_report")
3053 prefetch_parts.append(f"# Prefetched Research: {url}\n{report}")
3054 except Exception as e:
3055 self.tool_trace.append({"tool": "site_report", "status": "error", "error": str(e)})
3057 # 2. Environment Intent (pwd, git status, etc)
3058 elif self._is_env_intent(user_input):
3059 if settings.yolo_mode:
3060 env_notes = []
3061 commands = [("pwd", "pwd"), ("user", "whoami"), ("git", "git status -sb")]
3062 for label, cmd in env_commands:
3063 try:
3064 start = time.time()
3065 res = await execute_tool("run_shell_command", {"command": cmd})
3066 duration = time.time() - start
3068 tool = SimpleNamespace(name="run_shell_command", arguments={"command": cmd}, status=ToolCallStatus.SUCCESS)
3069 self.tool_events.append((tool, duration))
3071 env_notes.append(f"{label}: {str(res).strip()}")
3072 except: continue
3073 if env_notes:
3074 prefetch_parts.append("# Environment Probe\n" + "\n".join(env_notes))
3076 return "\n\n".join(prefetch_parts) if prefetch_parts else None
3078 async def _async_handle_path_mentions(self, user_input: str) -> Optional[str]:
3079 """Restored path-mention logic (@file) - Asynchronous version"""
3080 path_mentions = self._extract_path_mentions(user_input)
3081 if not path_mentions:
3082 return None
3084 file_blocks = []
3085 for path in path_mentions:
3086 try:
3087 start = time.time()
3088 result = await execute_tool("read_file", {"path": path, "start_line": 1, "end_line": 200})
3089 duration = time.time() - start
3091 tool = SimpleNamespace(name="read_file", arguments={"path": path}, status=ToolCallStatus.SUCCESS)
3092 self.tool_events.append((tool, duration))
3093 self.tool_names.append("read_file")
3095 text = str(result)
3096 if len(text) > 4000: text = text[:4000] + "\n... (truncated)"
3097 file_blocks.append(f"# Attached File: {path}\n{text}")
3098 except: continue
3100 return "\n\n".join(file_blocks) if file_blocks else None
3102 def _get_active_subsystems(self) -> set:
3103 """Map tool names to HUD subsystems"""
3104 active = set()
3105 for name in self.tool_names:
3106 n = name.lower()
3107 if any(x in n for x in ["file", "dir", "tree", "cpp", "read", "list", "insights", "map"]): active.add("FILES")
3108 if any(x in n for x in ["url", "search", "site"]): active.add("NET")
3109 if any(x in n for x in ["shell", "terminal", "mkdir", "write", "replace", "build"]): active.add("SHELL")
3110 if any(x in n for x in ["memory", "recall"]): active.add("MEM")
3111 if any(x in n for x in ["test", "diagnostics"]): active.add("TEST")
3112 if any(x in n for x in ["write", "replace", "artifact", "mkdir"]): active.add("FLOW")
3113 return active
3115 async def run(self):
3116 """Main CLI loop with interactive prompt"""
3117 await self.show_welcome()
3119 if not sys.stdin.isatty():
3120 console.print("[red]Error: This CLI requires an interactive terminal session.")
3121 console.print("Run in a real terminal: ./bin/osiris")
3122 return
3124 while self.running:
3125 try:
3126 # Beautiful prompt with auto-completion
3127 prompt_text = HTML('<ansicyan><b>You</b></ansicyan> <style fg="#6B7280">›</style> ')
3129 user_input = await self.prompt_session.prompt_async(
3130 prompt_text
3131 )
3133 if not user_input.strip():
3134 continue
3136 if user_input.startswith("!"):
3137 command = user_input[1:].strip()
3138 if not command:
3139 continue
3140 console.print()
3141 result = execute_tool("run_shell_command", {"command": command})
3142 output_text = str(result).strip()
3143 if len(output_text) > 4000:
3144 output_text = output_text[:4000] + "\n... (truncated)"
3145 self.pending_shell_context.append(f"$ {command}\n{output_text}")
3146 shell_body = Syntax(output_text or "(no output)", "bash", theme="ansi_dark", word_wrap=True)
3147 self._print_section(
3148 title=f"Shell • {command}",
3149 body=shell_body,
3150 style=self.COLORS["primary"],
3151 )
3152 console.print()
3153 continue
3155 if user_input.startswith("#"):
3156 payload = user_input[1:].strip()
3157 if not payload:
3158 continue
3159 if ":" in payload:
3160 await self.handle_slash_command(f"/remember {payload}")
3161 else:
3162 note_key = f"note_{datetime.now().strftime('%H%M%S')}"
3163 await self.handle_slash_command(f"/remember {note_key}:{payload}")
3164 continue
3166 if user_input.startswith("/"):
3167 if await self.handle_slash_command(user_input):
3168 continue
3170 await self.handle_message(user_input)
3172 except KeyboardInterrupt:
3173 console.print(f"\n[{self.COLORS['warning']}]Use /exit to quit[/{self.COLORS['warning']}]\n")
3174 continue
3175 except EOFError:
3176 self.show_session_summary()
3177 break
3179 # Goodbye message
3180 goodbye = Text()
3181 goodbye.append("\n✨ ", style="bold")
3182 goodbye.append("Goodbye!", style=f"bold {self.COLORS['primary']}")
3183 goodbye.append("\n", style="")
3184 console.print(goodbye)
3187async def main(agent_name: str = "main", prompt: Optional[str] = None, print_mode: bool = False, context_file: Optional[str] = None):
3188 """Entry point"""
3189 cli = BeautifulCLI(agent_name=agent_name, context_file=context_file)
3190 cli.print_mode = print_mode
3191 if prompt:
3192 cli.prompt_mode = True # Disable follow-up in prompt mode
3194 if prompt:
3195 # 2025 Sovereign UX: Show welcome even in prompt mode unless print_mode is on
3196 if not print_mode:
3197 await cli.show_welcome()
3198 await cli.handle_message(prompt)
3199 if not print_mode:
3200 cli.show_session_summary()
3201 else:
3202 await cli.run()