| """Rich TUI for the DRIFT companion bot. |
| |
| Run with: python tui.py |
| """ |
|
|
| import queue |
| import sys |
| import threading |
| from typing import Any, List |
|
|
| from rich.align import Align |
| from rich.console import Console |
| from rich.layout import Layout |
| from rich.live import Live |
| from rich.panel import Panel |
| from rich.rule import Rule |
| from rich.text import Text |
|
|
| sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent)) |
|
|
| from infj_bot.core.brain import DriftBrain |
| from infj_bot.core.commands import BotState, handle_command, is_command, parse_command |
| from infj_bot.core.config import DEFAULT_AUTHORIZED_TARGETS |
| from infj_bot.core.plugins.documents import DocumentStore |
| from infj_bot.core.plugins.goals import GoalsDB |
| from infj_bot.core.plugins.growth import growth_profile |
| from infj_bot.core.history import ChatHistory |
| from infj_bot.core.memory import DriftMemory |
| from infj_bot.core.prompt_builder import build_chat_prompt |
| from infj_bot.core.plugins.proactive import ProactiveState |
|
|
| console = Console() |
|
|
|
|
| class DriftTUI: |
| def __init__(self): |
| self.brain = DriftBrain() |
| self.memory = DriftMemory() |
| self.history = ChatHistory() |
| self.state = BotState(authorized_targets=set(DEFAULT_AUTHORIZED_TARGETS)) |
| self.goals_db = GoalsDB() |
| self.proactive_state = ProactiveState() |
| self.doc_store = DocumentStore() |
|
|
| self.messages = [] |
| self.input_queue = queue.Queue() |
| self.running = True |
| self._live = None |
|
|
| def _make_header(self) -> Panel: |
| text = Text() |
| text.append("DRIFT Companion Bot ", style="bold cyan") |
| text.append("v1.3 ", style="dim") |
| text.append(f"Mode: {self.state.mode} ", style="green") |
| text.append(f"Turns: {self.state.turns}", style="dim") |
| return Panel(Align.left(text), height=3, style="on black") |
|
|
| def _make_sidebar(self) -> Panel: |
| lines: List[Text] = [] |
| |
| health = self.brain.health_check() |
| lines.append(Text("Health", style="bold yellow")) |
| g = "[green]●[/]" if health["gemini"]["ok"] else "[red]●[/]" |
| local_dot = "[green]●[/]" if health["local"]["ok"] else "[red]●[/]" |
| lines.append(Text.from_markup(f" Gemini {g} Ollama {local_dot}")) |
| lines.append(Text("")) |
|
|
| |
| gp = growth_profile(self.memory, self.state.turns) |
| lines.append(Text("Growth", style="bold yellow")) |
| lines.append(Text.from_markup(f" {gp['stage']} ({gp['points']} pts)")) |
| lines.append(Text("")) |
|
|
| |
| lines.append(Text("Memory", style="bold yellow")) |
| lines.append(Text(f" {self.memory.count()} entries")) |
| lines.append(Text("")) |
|
|
| |
| lines.append(Text("Reminders", style="bold yellow")) |
| tasks = self.state.scheduler.list_pending(limit=3) |
| if tasks: |
| for t in tasks: |
| due = t.run_at.strftime("%H:%M") |
| lines.append(Text(f" • {due} {t.title[:24]}", style="dim")) |
| else: |
| lines.append(Text(" None", style="dim")) |
| lines.append(Text("")) |
|
|
| |
| if self.state.authorized_targets: |
| lines.append(Text("Authorized", style="bold yellow")) |
| for d in sorted(self.state.authorized_targets)[:5]: |
| lines.append(Text(f" • {d}", style="dim")) |
|
|
| content = Text("\n").join(lines) |
| return Panel(content, title="Status", border_style="blue", width=28) |
|
|
| def _make_chat(self) -> Panel: |
| content: Any |
| if not self.messages: |
| content = Align.center( |
| Text("Start a conversation...", style="dim italic"), vertical="middle" |
| ) |
| else: |
| parts: List[Text] = [] |
| for role, text in self.messages[-50:]: |
| if role == "user": |
| parts.append( |
| Text.from_markup( |
| f"[bold blue]user[/] {text[:300]}{'…' if len(text) > 300 else ''}" |
| ) |
| ) |
| elif role == "bot": |
| parts.append(Text(text)) |
| elif role == "system": |
| parts.append(Text.from_markup(f"[dim italic]{text}[/]")) |
| parts.append(Text("")) |
| content = Text("\n").join(parts) |
| return Panel(content, title="Conversation", border_style="green") |
|
|
| def _make_footer(self) -> Panel: |
| return Panel( |
| Text("Type /help for commands • exit to quit", style="dim"), |
| height=3, |
| style="on black", |
| ) |
|
|
| def render(self) -> Layout: |
| layout = Layout() |
| layout.split_column( |
| Layout(self._make_header(), size=3), |
| Layout(name="main"), |
| Layout(self._make_footer(), size=3), |
| ) |
| layout["main"].split_row( |
| Layout(self._make_chat(), name="chat"), |
| Layout(self._make_sidebar(), name="sidebar", size=30), |
| ) |
| return layout |
|
|
| def add_message(self, role: str, text: str): |
| self.messages.append((role, text)) |
|
|
| def _input_thread(self): |
| """Read user input in a background thread.""" |
| while self.running: |
| try: |
| user_input = console.input("[bold blue]JUDE> [/]") |
| self.input_queue.put(user_input) |
| except (EOFError, KeyboardInterrupt): |
| self.input_queue.put("exit") |
| break |
|
|
| def _process_input(self, user_input: str): |
| if user_input.lower() in ("exit", "quit"): |
| self.running = False |
| return |
|
|
| self.add_message("user", user_input) |
|
|
| if is_command(user_input): |
| command, args = parse_command(user_input) |
| output = handle_command( |
| command, |
| args, |
| self.state, |
| self.brain, |
| self.memory, |
| self.history, |
| self.goals_db, |
| self.doc_store, |
| ) |
| self.add_message("system", output) |
| return |
|
|
| prompt, emotion, dissonance = build_chat_prompt( |
| user_input, |
| self.state, |
| self.memory, |
| goals_db=self.goals_db, |
| doc_store=self.doc_store, |
| prefs=self.state.prefs, |
| ) |
| output = self.brain.agent_turn(prompt, tools_enabled=True) |
| try: |
| self.brain.evaluate_last(prompt, output) |
| except Exception: |
| pass |
|
|
| importance = min( |
| 0.95, 0.45 + emotion["intensity"] * 0.3 + dissonance["score"] * 0.15 |
| ) |
| self.memory.save_interaction( |
| user_input, |
| output, |
| mode=self.state.mode, |
| emotion=emotion, |
| importance=importance, |
| dissonance=dissonance, |
| ) |
| self.history.append(user_input, output, self.state.mode, emotion, dissonance) |
| self.state.turns += 1 |
| self.proactive_state.record_interaction(user_input, emotion, dissonance) |
|
|
| self.add_message("bot", output) |
|
|
| def run(self): |
| console.clear() |
| console.print(Rule("[bold cyan]DRIFT Companion Bot v1.3[/]", style="cyan")) |
| console.print("[dim]A mind that listens, remembers, and wonders.[/]\n") |
| console.print("[dim]Type /help for commands, exit to quit.[/]\n") |
|
|
| input_thread = threading.Thread(target=self._input_thread, daemon=True) |
| input_thread.start() |
|
|
| with Live( |
| self.render(), console=console, refresh_per_second=4, screen=False |
| ) as live: |
| self._live = live |
| while self.running: |
| try: |
| user_input = self.input_queue.get(timeout=0.5) |
| except queue.Empty: |
| live.update(self.render()) |
| continue |
|
|
| if user_input.lower() in ("exit", "quit"): |
| self.running = False |
| break |
|
|
| self._process_input(user_input) |
| live.update(self.render()) |
|
|
| console.print( |
| "\n[dim]I'll be here in the quiet if you need me again. Goodbye, user.[/]" |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| try: |
| DriftTUI().run() |
| except KeyboardInterrupt: |
| console.print("\n[dim]Manual override. Powering down.[/]") |
|
|
