Compare commits

...
2 Commits
Author SHA1 Message Date
Mervin Praison fd23d62841 Release v3.7.7 2026-01-11 21:30:04 +00:00
Mervin Praison 5f3bcb78ac Release v3.7.6 2026-01-11 08:23:08 +00:00
12 changed files with 2382 additions and 20 deletions

No files matched your search

+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.5" \
"praisonai>=3.7.7" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.5" \
"praisonai>=3.7.7" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.5" \
"praisonai>=3.7.7" \
"praisonai[ui]" \
"praisonai[crewai]"
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.5.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.5.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.7.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.7.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+4 -4
View File
@@ -143,15 +143,15 @@ def main_callback(
# If no command provided, start interactive mode
if ctx.invoked_subcommand is None:
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
tui_config = TUIConfig(
tui_config = AsyncTUIConfig(
model="gpt-4o-mini",
show_logo=True,
compact_mode=state.output_format == OutputFormat.json,
show_status_bar=state.output_format != OutputFormat.json,
)
tui = PraisonTUI(config=tui_config)
tui = AsyncTUI(config=tui_config)
tui.run()
+8 -9
View File
@@ -171,30 +171,29 @@ def chat_main(
typer.echo(" Use: praisonai chat \"your prompt\" --profile", err=True)
# Parse memory flag: --no-memory takes precedence, then --memory value
memory_value = _parse_memory_flag(memory, no_memory)
# TODO: Pass memory_value to TUI when memory support is added
_parse_memory_flag(memory, no_memory)
# Use the new TUI application (Aider/Claude Code style)
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
# Use the async TUI (non-blocking, scrollable output)
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
tui_config = TUIConfig(
tui_config = AsyncTUIConfig(
model=model or "gpt-4o-mini",
show_logo=not compact,
show_tips=not compact,
show_status_bar=not compact,
compact_mode=compact,
session_id=session_id,
workspace=workspace,
)
tui = PraisonTUI(config=tui_config)
tui = AsyncTUI(config=tui_config)
if prompt:
# Single prompt mode
# Single prompt mode - direct response, no streaming
response = tui.run_single(prompt)
if response:
print(response)
else:
# Interactive TUI mode
# Interactive split-pane TUI mode
tui.run()
@@ -0,0 +1,991 @@
"""
PraisonAI Async TUI Application.
Full-screen terminal UI with:
- Non-blocking LLM execution (UI stays responsive)
- Scrollable output area with keyboard navigation
- Fixed bottom input pane (always visible)
- Activity indicators during processing
Key fixes:
- Uses threading for LLM calls to prevent UI freeze
- Uses Buffer + BufferControl for scrollable output
- Proper app.invalidate() for UI refresh
"""
import shutil
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional
from datetime import datetime
# ============================================================================
# ASCII Art Logo - "Praison AI" branding
# ============================================================================
LOGO = r"""
██████╗ ██████╗ █████╗ ██╗███████╗ ██████╗ ███╗ ██╗ █████╗ ██╗
██╔══██╗██╔══██╗██╔══██╗██║██╔════╝██╔═══██╗████╗ ██║ ██╔══██╗██║
██████╔╝██████╔╝███████║██║███████╗██║ ██║██╔██╗ ██║ ███████║██║
██╔═══╝ ██╔══██╗██╔══██║██║╚════██║██║ ██║██║╚██╗██║ ██╔══██║██║
██║ ██║ ██║██║ ██║██║███████║╚██████╔╝██║ ╚████║ ██║ ██║██║
╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═╝ ╚═╝╚═╝
"""
LOGO_SMALL = r"""
╔═╗┬─┐┌─┐┬┌─┐┌─┐┌┐┌ ╔═╗╦
╠═╝├┬┘├─┤│└─┐│ ││││ ╠═╣║
╩ ┴└─┴ ┴┴└─┘└─┘┘└┘ ╩ ╩╩
"""
LOGO_MINIMAL = "▶ Praison AI"
def get_logo(width: int = 80) -> str:
"""Get appropriate logo based on terminal width."""
if width >= 75:
return LOGO
elif width >= 40:
return LOGO_SMALL
else:
return LOGO_MINIMAL
def get_version() -> str:
"""Get PraisonAI version."""
try:
from praisonai import __version__
return __version__
except Exception:
return "1.0.0"
# ============================================================================
# Configuration
# ============================================================================
@dataclass
class AsyncTUIConfig:
"""Configuration for the Async TUI application."""
model: str = "gpt-4o-mini"
show_logo: bool = True
show_status_bar: bool = True
session_id: Optional[str] = None
workspace: Optional[str] = None
history_file: Optional[str] = None
compact_mode: bool = False
multiline_mode: bool = False
# ============================================================================
# Message Types
# ============================================================================
@dataclass
class ChatMessage:
"""A chat message."""
role: str # "user", "assistant", "system", "status"
content: str
timestamp: datetime = field(default_factory=datetime.now)
# ============================================================================
# Async TUI Application
# ============================================================================
class AsyncTUI:
"""
Async TUI with non-blocking LLM execution and scrollable output.
Key features:
- Input box ALWAYS visible at bottom (never freezes)
- Output scrolls above independently
- LLM runs in background thread
- UI stays responsive during processing
"""
def __init__(self, config: Optional[AsyncTUIConfig] = None):
self.config = config or AsyncTUIConfig()
self.messages: List[ChatMessage] = []
self._running = False
self._agent = None
self._processing = False
self._status_text = ""
self._app = None
self._output_buffer = None
self._prompt_queue: List[str] = [] # Queue for pending prompts
self._conversation_history: List[dict] = [] # Full conversation history
self._total_tokens = 0
self._total_cost = 0.0
self._workspace_files: List[str] = [] # Files in workspace for @ completion
self._runtime = None # InteractiveRuntime for ACP/LSP
self._runtime_started = False
# Terminal size
self.term_width, self.term_height = shutil.get_terminal_size((80, 24))
# Session ID
import uuid
self.session_id = self.config.session_id or str(uuid.uuid4())[:8]
# Scan workspace for files
self._scan_workspace()
def _scan_workspace(self):
"""Scan workspace for files (for @ completion)."""
import os
workspace = self.config.workspace or os.getcwd()
try:
for root, dirs, files in os.walk(workspace):
# Skip hidden directories
dirs[:] = [d for d in dirs if not d.startswith('.')]
for f in files:
if not f.startswith('.'):
rel_path = os.path.relpath(os.path.join(root, f), workspace)
self._workspace_files.append(rel_path)
# Limit to 1000 files
if len(self._workspace_files) >= 1000:
break
except Exception:
pass
def _process_file_mentions(self, prompt: str) -> str:
"""Process @file mentions and include file contents."""
import os
import re
workspace = self.config.workspace or os.getcwd()
# Find all @file mentions
pattern = r'@([^\s]+)'
matches = re.findall(pattern, prompt)
if not matches:
return prompt
file_contents = []
for match in matches:
file_path = os.path.join(workspace, match)
if os.path.isfile(file_path):
try:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
file_contents.append(f"\n--- Content of {match} ---\n{content}\n--- End of {match} ---\n")
except Exception as e:
file_contents.append(f"\n[Error reading {match}: {e}]\n")
if file_contents:
return prompt + "\n" + "\n".join(file_contents)
return prompt
def _get_agent(self):
"""Lazy-load the agent with tools."""
if self._agent is None:
try:
from praisonaiagents import Agent
# Load interactive tools (read_file, write_file, execute_command, etc.)
tools = self._load_tools()
self._agent = Agent(
name="Praison",
role="AI Assistant",
goal="Help the user with their requests",
instructions="You are Praison AI, a helpful assistant. Be concise and helpful. You have access to tools for file operations, code execution, and web search.",
llm=self.config.model,
tools=tools if tools else None,
)
except ImportError as e:
raise RuntimeError(f"Failed to import praisonaiagents: {e}")
return self._agent
async def _start_runtime(self):
"""Start the InteractiveRuntime with ACP/LSP servers."""
if self._runtime_started:
return
try:
from praisonai.cli.features.interactive_runtime import create_runtime
self._runtime = create_runtime(
workspace=self.config.workspace or ".",
lsp=True,
acp=True,
approval="auto", # Auto-approve for interactive mode
)
# Start runtime (this starts ACP/LSP servers)
await self._runtime.start()
self._runtime_started = True
# Return status
return {
"lsp": "ready" if self._runtime.lsp_ready else "failed",
"acp": "ready" if self._runtime.acp_ready else "failed",
"read_only": self._runtime.read_only
}
except ImportError as e:
# Runtime not available, continue without it
self._runtime_started = True # Mark as attempted
return {"error": f"Runtime not available: {e}"}
except Exception as e:
self._runtime_started = True
return {"error": str(e)}
def _load_tools(self):
"""Load interactive tools for the agent."""
tools = []
# Try to load all interactive tools (basic + ACP + LSP)
try:
from praisonai.cli.features.interactive_tools import get_interactive_tools
tools = get_interactive_tools(
groups=["basic", "acp", "lsp"], # All tool groups
workspace=self.config.workspace,
)
if tools:
return tools
except ImportError:
pass
except Exception:
pass
# Fallback: try to load basic tools directly from praisonaiagents
try:
from praisonaiagents.tools import (
read_file,
write_file,
list_files,
execute_command,
)
tools = [read_file, write_file, list_files, execute_command]
except ImportError:
pass
# Try to add internet search
try:
from praisonaiagents.tools import internet_search
tools.append(internet_search)
except ImportError:
pass
return tools
def _format_output(self) -> str:
"""Format all messages for the output pane."""
lines = []
# Logo (only if no messages yet)
if self.config.show_logo and not self.messages:
logo = get_logo(self.term_width)
lines.append(logo.strip())
lines.append("")
lines.append(f" v{get_version()} · Model: {self.config.model}")
lines.append("")
lines.append(" Type your message and press Enter. Use /help for commands.")
lines.append(" Use PageUp/PageDown or Ctrl+Up/Down to scroll.")
lines.append("")
# Messages
for msg in self.messages:
if msg.role == "user":
lines.append(f"› {msg.content}")
lines.append("")
elif msg.role == "assistant":
lines.append(f"● {msg.content}")
lines.append("")
elif msg.role == "system":
lines.append(f" {msg.content}")
lines.append("")
elif msg.role == "status":
lines.append(f" ⏳ {msg.content}")
lines.append("")
# Processing indicator
if self._processing:
lines.append(f" ⏳ {self._status_text}")
lines.append("")
return "\n".join(lines)
def _update_output(self):
"""Update the output buffer and refresh UI."""
if self._output_buffer:
new_text = self._format_output()
self._output_buffer.set_document(
self._output_buffer.document.__class__(
text=new_text,
cursor_position=len(new_text) # Cursor at end for auto-scroll
),
bypass_readonly=True
)
if self._app:
self._app.invalidate()
def _handle_command(self, command: str) -> bool:
"""Handle slash commands. Returns True if handled."""
parts = command.split(maxsplit=1)
cmd = parts[0].lower().lstrip("/")
args = parts[1] if len(parts) > 1 else ""
if cmd in ("exit", "quit", "q"):
self._running = False
if self._app:
self._app.exit()
return True
elif cmd == "help":
help_text = """Available Commands:
/help Show this help
/exit, /quit Exit Praison AI
/clear Clear conversation
/new Start new conversation
/model [name] Show or change model
/session Show current session info
/sessions List all saved sessions
/continue Continue most recent session
/history Show conversation history
/export [file] Export conversation to file
/import <file> Import conversation from file
/cost Show token usage and cost
/status Show ACP/LSP runtime status
/compact Toggle compact output mode
/multiline Toggle multiline input mode
/files List workspace files for @ mentions
/queue Show pending prompts in queue
Keyboard Shortcuts:
PageUp/Down Scroll output
Ctrl+Up/Down Scroll output
Ctrl+C Clear input
Ctrl+D Exit (if input empty)
Ctrl+Q Exit
Tips:
Use @filename to include file contents in your prompt
Type multiple prompts while AI is thinking (queued)"""
self.messages.append(ChatMessage(role="system", content=help_text))
return True
elif cmd == "clear":
self.messages.clear()
self._conversation_history.clear()
self.messages.append(ChatMessage(role="system", content="Conversation cleared."))
return True
elif cmd == "new":
self.messages.clear()
self._conversation_history.clear()
import uuid
self.session_id = str(uuid.uuid4())[:8]
self.messages.append(ChatMessage(role="system", content=f"New session: {self.session_id}"))
return True
elif cmd == "model":
if args:
self.config.model = args
self._agent = None
self.messages.append(ChatMessage(role="system", content=f"Model changed to: {args}"))
else:
self.messages.append(ChatMessage(role="system", content=f"Current model: {self.config.model}"))
return True
elif cmd == "session":
info = f"Session: {self.session_id}\nModel: {self.config.model}\nMessages: {len(self._conversation_history)}"
self.messages.append(ChatMessage(role="system", content=info))
return True
elif cmd == "history":
if not self._conversation_history:
self.messages.append(ChatMessage(role="system", content="No conversation history."))
else:
history_lines = []
for i, msg in enumerate(self._conversation_history[-10:], 1):
role = msg.get("role", "unknown")
content = msg.get("content", "")[:80]
history_lines.append(f"{i}. [{role}] {content}...")
self.messages.append(ChatMessage(role="system", content="\n".join(history_lines)))
return True
elif cmd == "export":
filename = args or f"conversation_{self.session_id}.txt"
try:
with open(filename, "w") as f:
f.write("# Praison AI Conversation\n")
f.write(f"# Session: {self.session_id}\n")
f.write(f"# Model: {self.config.model}\n\n")
for msg in self._conversation_history:
f.write(f"[{msg.get('role', 'unknown')}]\n")
f.write(f"{msg.get('content', '')}\n\n")
self.messages.append(ChatMessage(role="system", content=f"Exported to {filename}"))
except Exception as e:
self.messages.append(ChatMessage(role="system", content=f"Export failed: {e}"))
return True
elif cmd == "cost":
cost_info = f"Total tokens: {self._total_tokens}\nEstimated cost: ${self._total_cost:.4f}"
self.messages.append(ChatMessage(role="system", content=cost_info))
return True
elif cmd == "compact":
self.config.compact_mode = not self.config.compact_mode
mode = "enabled" if self.config.compact_mode else "disabled"
self.messages.append(ChatMessage(role="system", content=f"Compact mode {mode}"))
return True
elif cmd == "multiline":
self.config.multiline_mode = not self.config.multiline_mode
mode = "enabled" if self.config.multiline_mode else "disabled"
self.messages.append(ChatMessage(role="system", content=f"Multiline mode {mode}. Use Ctrl+D to submit."))
return True
elif cmd == "files":
if not self._workspace_files:
self.messages.append(ChatMessage(role="system", content="No files found in workspace."))
else:
files_list = self._workspace_files[:20]
more = f"\n... and {len(self._workspace_files) - 20} more" if len(self._workspace_files) > 20 else ""
self.messages.append(ChatMessage(role="system", content="Workspace files:\n" + "\n".join(f" @{f}" for f in files_list) + more))
return True
elif cmd == "queue":
if not self._prompt_queue:
self.messages.append(ChatMessage(role="system", content="No prompts in queue."))
else:
queue_list = [f"{i+1}. {p[:50]}..." for i, p in enumerate(self._prompt_queue)]
self.messages.append(ChatMessage(role="system", content="Queued prompts:\n" + "\n".join(queue_list)))
return True
elif cmd == "sessions":
# List available sessions
try:
from praisonai.cli.session import get_session_store
store = get_session_store()
sessions = store.list_sessions()
if not sessions:
self.messages.append(ChatMessage(role="system", content="No saved sessions found."))
else:
lines = ["Available sessions:"]
for s in sessions[:10]:
sid = s.get("session_id", "unknown")
title = s.get("title", "Untitled")
lines.append(f" {sid}: {title}")
if len(sessions) > 10:
lines.append(f" ... and {len(sessions) - 10} more")
self.messages.append(ChatMessage(role="system", content="\n".join(lines)))
except Exception as e:
self.messages.append(ChatMessage(role="system", content=f"Could not list sessions: {e}"))
return True
elif cmd == "continue":
# Continue most recent session
try:
from praisonai.cli.session import get_session_store
store = get_session_store()
sessions = store.list_sessions()
if not sessions:
self.messages.append(ChatMessage(role="system", content="No sessions to continue."))
else:
# Get most recent
sorted_sessions = sorted(
sessions,
key=lambda s: s.get("updated_at", s.get("created_at", 0)),
reverse=True
)
if sorted_sessions:
session = store.load(sorted_sessions[0].get("session_id"))
if session:
self.session_id = session.session_id
# Load history
history = session.get_chat_history()
self._conversation_history = history
self.messages.append(ChatMessage(
role="system",
content=f"Continued session: {self.session_id} ({len(history)} messages)"
))
else:
self.messages.append(ChatMessage(role="system", content="Could not load session."))
except Exception as e:
self.messages.append(ChatMessage(role="system", content=f"Could not continue session: {e}"))
return True
elif cmd == "import":
# Import session from file
if not args:
self.messages.append(ChatMessage(role="system", content="Usage: /import <filename>"))
else:
try:
import json
with open(args) as f:
data = json.load(f)
# Load messages from file
messages = data.get("messages", [])
self._conversation_history = messages
self.session_id = data.get("session_id", self.session_id)
self.messages.append(ChatMessage(
role="system",
content=f"Imported {len(messages)} messages from {args}"
))
except FileNotFoundError:
self.messages.append(ChatMessage(role="system", content=f"File not found: {args}"))
except Exception as e:
self.messages.append(ChatMessage(role="system", content=f"Import failed: {e}"))
return True
elif cmd == "status":
# Show runtime status (ACP/LSP)
if self._runtime:
status = self._runtime.get_status()
lsp = status.get("lsp", {})
acp = status.get("acp", {})
info = f"Runtime Status:\n LSP: {lsp.get('status', 'unknown')}\n ACP: {acp.get('status', 'unknown')}\n Read-only: {status.get('read_only', False)}"
self.messages.append(ChatMessage(role="system", content=info))
else:
self.messages.append(ChatMessage(role="system", content="Runtime not initialized."))
return True
else:
self.messages.append(ChatMessage(role="system", content=f"Unknown command: /{cmd}. Use /help for available commands."))
return True
def _execute_prompt(self, prompt: str) -> Optional[str]:
"""Execute a prompt and return the response (suppresses agent output)."""
import sys
import io
import asyncio
try:
agent = self._get_agent()
# Suppress agent's Rich output
old_stdout = sys.stdout
old_stderr = sys.stderr
sys.stdout = io.StringIO()
sys.stderr = io.StringIO()
try:
# Try async execution first for better non-blocking behavior
if hasattr(agent, 'astart'):
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
response = loop.run_until_complete(agent.astart(prompt))
loop.close()
except Exception:
# Fallback to sync
response = agent.start(prompt)
else:
response = agent.start(prompt)
finally:
sys.stdout = old_stdout
sys.stderr = old_stderr
return str(response) if response else None
except Exception as e:
return f"Error: {e}"
def _execute_in_background(self, prompt: str):
"""Execute prompt in background thread (non-blocking)."""
def run():
self._processing = True
self._status_text = "Praison AI is thinking..."
self._update_output()
# Process @file mentions
processed_prompt = self._process_file_mentions(prompt)
# Add to conversation history
self._conversation_history.append({"role": "user", "content": prompt})
# Animate spinner while processing
spinner = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
start_time = time.time()
# Start LLM execution in another thread
result = [None]
error = [None]
done = [False]
def execute_llm():
try:
result[0] = self._execute_prompt(processed_prompt)
except Exception as e:
error[0] = str(e)
finally:
done[0] = True
llm_thread = threading.Thread(target=execute_llm)
llm_thread.start()
# Update spinner while waiting
idx = 0
while not done[0]:
elapsed = time.time() - start_time
queue_info = f" | Queue: {len(self._prompt_queue)}" if self._prompt_queue else ""
self._status_text = f"{spinner[idx]} Praison AI is thinking... ({elapsed:.1f}s){queue_info}"
self._update_output()
idx = (idx + 1) % len(spinner)
time.sleep(0.1)
llm_thread.join()
# Done processing current prompt
self._processing = False
self._status_text = ""
if error[0]:
self.messages.append(ChatMessage(role="assistant", content=f"Error: {error[0]}"))
self._conversation_history.append({"role": "assistant", "content": f"Error: {error[0]}"})
elif result[0]:
self.messages.append(ChatMessage(role="assistant", content=result[0]))
self._conversation_history.append({"role": "assistant", "content": result[0]})
self._update_output()
# Process next item in queue if any
if self._prompt_queue:
next_prompt = self._prompt_queue.pop(0)
self.messages.append(ChatMessage(role="user", content=next_prompt))
self._update_output()
self._execute_in_background(next_prompt)
# Run in background thread
thread = threading.Thread(target=run, daemon=True)
thread.start()
def _queue_or_execute(self, prompt: str):
"""Queue prompt if processing, otherwise execute immediately."""
if self._processing:
# Add to queue
self._prompt_queue.append(prompt)
self.messages.append(ChatMessage(role="system", content=f"Queued: {prompt[:50]}..."))
self._update_output()
else:
# Execute immediately
self.messages.append(ChatMessage(role="user", content=prompt))
self._update_output()
self._execute_in_background(prompt)
def run(self) -> None:
"""Run the async TUI."""
import asyncio
self._running = True
# Start runtime (ACP/LSP servers) before TUI
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
runtime_status = loop.run_until_complete(self._start_runtime())
loop.close()
# Show runtime status
if runtime_status and "error" not in runtime_status:
lsp = runtime_status.get("lsp", "unknown")
acp = runtime_status.get("acp", "unknown")
if lsp == "ready" or acp == "ready":
self.messages.append(ChatMessage(
role="system",
content=f"Runtime started: LSP={lsp}, ACP={acp}"
))
except Exception as e:
# Continue without runtime
self.messages.append(ChatMessage(
role="system",
content=f"Runtime unavailable: {e}"
))
try:
self._run_with_prompt_toolkit()
except ImportError:
self._run_simple()
def _run_simple(self) -> None:
"""Simple fallback mode without prompt_toolkit."""
print(self._format_output())
while self._running:
try:
if self.config.show_status_bar:
status = f" ? for help | {self.config.model} | Session: {self.session_id} "
print(f"\033[90m{status}\033[0m")
user_input = input("› ").strip()
if not user_input:
continue
if user_input.startswith("/"):
self._handle_command(user_input)
for msg in self.messages:
if msg.role == "system":
print(f" {msg.content}")
self.messages = [m for m in self.messages if m.role != "system"]
continue
self.messages.append(ChatMessage(role="user", content=user_input))
print(" ⏳ Praison AI is thinking...")
response = self._execute_prompt(user_input)
if response:
self.messages.append(ChatMessage(role="assistant", content=response))
print(f"\n● {response}\n")
except KeyboardInterrupt:
print("\n Use /exit to quit")
except EOFError:
self._running = False
print("\n Goodbye from Praison AI!")
def _run_with_prompt_toolkit(self) -> None:
"""Run with prompt_toolkit for non-blocking split-pane layout."""
from prompt_toolkit import Application
from prompt_toolkit.buffer import Buffer
from prompt_toolkit.document import Document
from prompt_toolkit.layout.containers import HSplit, VSplit, Window, Float, FloatContainer
from prompt_toolkit.layout.controls import BufferControl, FormattedTextControl
from prompt_toolkit.layout.layout import Layout
from prompt_toolkit.layout.margins import ScrollbarMargin
from prompt_toolkit.layout.menus import CompletionsMenu
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.styles import Style
from prompt_toolkit.history import FileHistory
from prompt_toolkit.completion import Completer, Completion
# Create completer for commands and files
commands = ["help", "exit", "quit", "clear", "new", "model", "session", "sessions",
"continue", "history", "export", "import", "cost", "status", "compact",
"multiline", "files", "queue"]
workspace_files = self._workspace_files
class PraisonCompleter(Completer):
def get_completions(self, document, complete_event):
text = document.text_before_cursor
# Get the current word being typed
# Handle case where text might be empty or have no words
if not text:
return
# Find the current word (last word or partial word)
words = text.split()
if not words:
current_word = text
elif text.endswith(' '):
current_word = ""
else:
current_word = words[-1]
# Show all slash commands when "/" is typed
if current_word == "/":
for cmd in commands:
yield Completion(f"/{cmd}", start_position=-1, display=f"/{cmd}", display_meta="command")
# Filter slash commands
elif current_word.startswith("/"):
cmd_part = current_word[1:].lower()
for cmd in commands:
if cmd.lower().startswith(cmd_part):
yield Completion(f"/{cmd}", start_position=-len(current_word), display=f"/{cmd}", display_meta="command")
# Show all files when "@" is typed
elif current_word == "@":
for f in workspace_files[:30]:
yield Completion(f"@{f}", start_position=-1, display=f"@{f}", display_meta="file")
# Filter @ file mentions
elif current_word.startswith("@"):
file_part = current_word[1:].lower()
for f in workspace_files[:50]:
if file_part in f.lower():
yield Completion(f"@{f}", start_position=-len(current_word), display=f"@{f}", display_meta="file")
# Setup history
history_file = self.config.history_file or str(Path.home() / ".praison" / "history")
Path(history_file).parent.mkdir(parents=True, exist_ok=True)
# Output buffer (read-only, scrollable)
initial_output = self._format_output()
self._output_buffer = Buffer(
document=Document(initial_output, cursor_position=len(initial_output)),
read_only=True,
name="output"
)
# Scroll offset for output
self._scroll_offset = 0
# Input buffer with completer and auto-complete on typing
input_buffer = Buffer(
multiline=False,
history=FileHistory(history_file),
name="input",
completer=PraisonCompleter(),
complete_while_typing=True, # Show completions as you type
)
# Key bindings
kb = KeyBindings()
@kb.add("enter")
def handle_enter(event):
user_input = input_buffer.text.strip()
if not user_input:
return
input_buffer.reset()
# Handle commands (always process immediately)
if user_input.startswith("/"):
self._handle_command(user_input)
self._update_output()
return
# Queue or execute prompt
self._queue_or_execute(user_input)
@kb.add("c-c")
def handle_ctrl_c(event):
input_buffer.reset()
@kb.add("c-d")
def handle_ctrl_d(event):
if not input_buffer.text:
self._running = False
event.app.exit()
@kb.add("c-q")
def handle_ctrl_q(event):
self._running = False
event.app.exit()
# Scroll bindings for output
@kb.add("pageup")
def handle_pageup(event):
self._output_buffer.cursor_up(count=10)
@kb.add("pagedown")
def handle_pagedown(event):
self._output_buffer.cursor_down(count=10)
@kb.add("c-up")
def handle_ctrl_up(event):
self._output_buffer.cursor_up(count=3)
@kb.add("c-down")
def handle_ctrl_down(event):
self._output_buffer.cursor_down(count=3)
@kb.add("home")
def handle_home(event):
self._output_buffer.cursor_position = 0
@kb.add("end")
def handle_end(event):
self._output_buffer.cursor_position = len(self._output_buffer.text)
# Style
style = Style.from_dict({
"output-area": "bg:#0d0d0d",
"input-area": "bg:#1a1a1a",
"status-bar": "bg:#333333 #888888",
"prompt": "#00aaff bold",
"scrollbar.background": "bg:#333333",
"scrollbar.button": "bg:#666666",
})
# Status bar
def get_status_bar():
if self._processing:
return f" {self._status_text} "
left = "? for help | PageUp/Down to scroll"
center = self.config.model
right = f"Session: {self.session_id}"
return f" {left} | {center} | {right} "
# Layout: Output (top, scrollable) + Status bar + Input (bottom, fixed)
# Use FloatContainer to show completion menu as floating overlay
main_layout = HSplit([
# Output area (scrollable with scrollbar)
Window(
content=BufferControl(
buffer=self._output_buffer,
focusable=False,
),
wrap_lines=True,
style="class:output-area",
right_margins=[ScrollbarMargin(display_arrows=True)],
),
# Status bar (1 line)
Window(
content=FormattedTextControl(get_status_bar),
height=1,
style="class:status-bar",
),
# Input area (fixed at bottom)
HSplit([
Window(height=1), # Spacer
VSplit([
Window(
content=FormattedTextControl([("class:prompt", "› ")]),
width=2,
),
Window(
content=BufferControl(buffer=input_buffer),
wrap_lines=True,
),
]),
Window(height=1), # Spacer
], height=3, style="class:input-area"),
])
# Wrap in FloatContainer to show completions menu
layout = Layout(
FloatContainer(
content=main_layout,
floats=[
Float(
xcursor=True,
ycursor=True,
content=CompletionsMenu(max_height=10, scroll_offset=1),
),
],
)
)
# Create application
self._app = Application(
layout=layout,
key_bindings=kb,
style=style,
full_screen=True,
mouse_support=True,
refresh_interval=0.1, # Refresh every 100ms for spinner
)
# Run
self._app.run()
print("\n Goodbye from Praison AI!")
def run_single(self, prompt: str) -> Optional[str]:
"""Run a single prompt (non-interactive)."""
return self._execute_prompt(prompt)
def start_async_tui(
model: str = "gpt-4o-mini",
show_logo: bool = True,
**kwargs
) -> None:
"""Start the async TUI application."""
config = AsyncTUIConfig(
model=model,
show_logo=show_logo,
)
tui = AsyncTUI(config=config)
tui.run()
@@ -0,0 +1,490 @@
"""
PraisonAI Split-Pane TUI Application.
Full-screen terminal UI with:
- Fixed bottom input pane (always visible)
- Scrollable top output pane
- Direct response (no streaming - faster)
- Activity indicators showing agent status
Uses prompt_toolkit Application for true split-pane layout.
"""
import shutil
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional
from datetime import datetime
# ============================================================================
# ASCII Art Logo - "Praison AI" branding
# ============================================================================
LOGO = r"""
██████╗ ██████╗ █████╗ ██╗███████╗ ██████╗ ███╗ ██╗ █████╗ ██╗
██╔══██╗██╔══██╗██╔══██╗██║██╔════╝██╔═══██╗████╗ ██║ ██╔══██╗██║
██████╔╝██████╔╝███████║██║███████╗██║ ██║██╔██╗ ██║ ███████║██║
██╔═══╝ ██╔══██╗██╔══██║██║╚════██║██║ ██║██║╚██╗██║ ██╔══██║██║
██║ ██║ ██║██║ ██║██║███████║╚██████╔╝██║ ╚████║ ██║ ██║██║
╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═╝ ╚═╝╚═╝
"""
LOGO_SMALL = r"""
╔═╗┬─┐┌─┐┬┌─┐┌─┐┌┐┌ ╔═╗╦
╠═╝├┬┘├─┤│└─┐│ ││││ ╠═╣║
╩ ┴└─┴ ┴┴└─┘└─┘┘└┘ ╩ ╩╩
"""
LOGO_MINIMAL = "▶ Praison AI"
def get_logo(width: int = 80) -> str:
"""Get appropriate logo based on terminal width."""
if width >= 75:
return LOGO
elif width >= 40:
return LOGO_SMALL
else:
return LOGO_MINIMAL
def get_version() -> str:
"""Get PraisonAI version."""
try:
from praisonai import __version__
return __version__
except Exception:
return "1.0.0"
# ============================================================================
# Configuration
# ============================================================================
@dataclass
class SplitTUIConfig:
"""Configuration for the Split TUI application."""
model: str = "gpt-4o-mini"
show_logo: bool = True
show_status_bar: bool = True
session_id: Optional[str] = None
workspace: Optional[str] = None
history_file: Optional[str] = None
# ============================================================================
# Message Types
# ============================================================================
@dataclass
class ChatMessage:
"""A chat message."""
role: str # "user", "assistant", "system", "status"
content: str
timestamp: datetime = field(default_factory=datetime.now)
# ============================================================================
# Split TUI Application
# ============================================================================
class SplitTUI:
"""
Split-pane TUI with fixed bottom input and scrollable top output.
Key features:
- Input box ALWAYS visible at bottom
- Output scrolls above independently
- Direct response (no streaming) for speed
- Activity indicators during processing
"""
def __init__(self, config: Optional[SplitTUIConfig] = None):
self.config = config or SplitTUIConfig()
self.messages: List[ChatMessage] = []
self._running = False
self._agent = None
self._processing = False
self._status_text = ""
# Terminal size
self.term_width, self.term_height = shutil.get_terminal_size((80, 24))
# Session ID
import uuid
self.session_id = self.config.session_id or str(uuid.uuid4())[:8]
def _get_agent(self):
"""Lazy-load the agent with verbose=False for no streaming."""
if self._agent is None:
try:
from praisonaiagents import Agent
self._agent = Agent(
name="Praison",
role="AI Assistant",
goal="Help the user with their requests",
instructions="You are Praison AI, a helpful assistant. Be concise and helpful.",
llm=self.config.model,
)
except ImportError as e:
raise RuntimeError(f"Failed to import praisonaiagents: {e}")
return self._agent
def _format_output(self) -> str:
"""Format all messages for the output pane."""
lines = []
# Logo
if self.config.show_logo and not self.messages:
logo = get_logo(self.term_width)
lines.append(logo.strip())
lines.append("")
lines.append(f" v{get_version()} · Model: {self.config.model}")
lines.append("")
lines.append(" Type your message and press Enter. Use /help for commands.")
lines.append("")
# Messages
for msg in self.messages:
if msg.role == "user":
lines.append(f"› {msg.content}")
lines.append("")
elif msg.role == "assistant":
lines.append(f"● {msg.content}")
lines.append("")
elif msg.role == "system":
lines.append(f" {msg.content}")
lines.append("")
elif msg.role == "status":
lines.append(f" ⏳ {msg.content}")
# Processing indicator
if self._processing:
lines.append(f" ⏳ {self._status_text}")
return "\n".join(lines)
def _handle_command(self, command: str) -> bool:
"""Handle slash commands. Returns True if handled."""
parts = command.split(maxsplit=1)
cmd = parts[0].lower().lstrip("/")
args = parts[1] if len(parts) > 1 else ""
if cmd in ("exit", "quit", "q"):
self._running = False
return True
elif cmd == "help":
help_text = """Available Commands:
/help Show this help
/exit, /quit Exit Praison AI
/clear Clear conversation
/model [name] Show or change model
/new Start new conversation"""
self.messages.append(ChatMessage(role="system", content=help_text))
return True
elif cmd == "clear":
self.messages.clear()
return True
elif cmd == "new":
self.messages.clear()
import uuid
self.session_id = str(uuid.uuid4())[:8]
self.messages.append(ChatMessage(role="system", content=f"New session: {self.session_id}"))
return True
elif cmd == "model":
if args:
self.config.model = args
self._agent = None
self.messages.append(ChatMessage(role="system", content=f"Model changed to: {args}"))
else:
self.messages.append(ChatMessage(role="system", content=f"Current model: {self.config.model}"))
return True
else:
self.messages.append(ChatMessage(role="system", content=f"Unknown command: /{cmd}. Use /help for available commands."))
return True
def _execute_prompt(self, prompt: str) -> Optional[str]:
"""Execute a prompt and return the response (direct, no streaming)."""
import sys
import io
try:
agent = self._get_agent()
# Suppress agent's Rich output by redirecting stdout/stderr
old_stdout = sys.stdout
old_stderr = sys.stderr
sys.stdout = io.StringIO()
sys.stderr = io.StringIO()
try:
response = agent.start(prompt)
finally:
# Restore stdout/stderr
sys.stdout = old_stdout
sys.stderr = old_stderr
return str(response) if response else None
except Exception as e:
return f"Error: {e}"
def _run_with_status(self, prompt: str) -> Optional[str]:
"""Run prompt with status updates."""
self._processing = True
self._status_text = "Praison AI is thinking..."
# Execute in background
result = [None]
error = [None]
def execute():
try:
result[0] = self._execute_prompt(prompt)
except Exception as e:
error[0] = str(e)
thread = threading.Thread(target=execute)
thread.start()
# Update status while waiting
start_time = time.time()
spinner = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
idx = 0
while thread.is_alive():
elapsed = time.time() - start_time
self._status_text = f"{spinner[idx]} Praison AI is thinking... ({elapsed:.1f}s)"
idx = (idx + 1) % len(spinner)
time.sleep(0.1)
thread.join()
self._processing = False
self._status_text = ""
if error[0]:
return f"Error: {error[0]}"
return result[0]
def run(self) -> None:
"""Run the split-pane TUI."""
self._running = True
# Try to use prompt_toolkit for better experience
try:
self._run_with_prompt_toolkit()
except ImportError:
# Fallback to simple mode
self._run_simple()
def _run_simple(self) -> None:
"""Simple fallback mode without prompt_toolkit."""
# Print welcome
print(self._format_output())
while self._running:
try:
# Status bar
if self.config.show_status_bar:
status = f" ? for help | {self.config.model} | Session: {self.session_id} "
print(f"\033[90m{status}\033[0m")
# Get input
user_input = input("› ").strip()
if not user_input:
continue
# Handle commands
if user_input.startswith("/"):
self._handle_command(user_input)
# Print system messages
for msg in self.messages:
if msg.role == "system":
print(f" {msg.content}")
self.messages = [m for m in self.messages if m.role != "system"]
continue
# Add user message
self.messages.append(ChatMessage(role="user", content=user_input))
# Execute with status
print(" ⏳ Praison AI is thinking...")
response = self._execute_prompt(user_input)
if response:
self.messages.append(ChatMessage(role="assistant", content=response))
print(f"\n● {response}\n")
except KeyboardInterrupt:
print("\n Use /exit to quit")
except EOFError:
self._running = False
print("\n Goodbye from Praison AI!")
def _run_with_prompt_toolkit(self) -> None:
"""Run with prompt_toolkit for split-pane layout."""
from prompt_toolkit import Application
from prompt_toolkit.buffer import Buffer
from prompt_toolkit.layout.containers import HSplit, VSplit, Window
from prompt_toolkit.layout.controls import BufferControl, FormattedTextControl
from prompt_toolkit.layout.layout import Layout
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.styles import Style
from prompt_toolkit.history import FileHistory
# Setup history
history_file = self.config.history_file or str(Path.home() / ".praison" / "history")
Path(history_file).parent.mkdir(parents=True, exist_ok=True)
# Output buffer (read-only, for displaying conversation)
output_text = [self._format_output()]
def get_output_text():
return output_text[0]
# Input buffer
input_buffer = Buffer(
multiline=False,
history=FileHistory(history_file),
)
# Key bindings
kb = KeyBindings()
@kb.add("enter")
def handle_enter(event):
user_input = input_buffer.text.strip()
if not user_input:
return
input_buffer.reset()
# Handle commands
if user_input.startswith("/"):
self._handle_command(user_input)
output_text[0] = self._format_output()
return
# Add user message
self.messages.append(ChatMessage(role="user", content=user_input))
# Show processing status
self.messages.append(ChatMessage(role="status", content="Praison AI is thinking..."))
output_text[0] = self._format_output()
# Execute (this blocks but that's OK for direct response)
response = self._execute_prompt(user_input)
# Remove status message
self.messages = [m for m in self.messages if m.role != "status"]
if response:
self.messages.append(ChatMessage(role="assistant", content=response))
output_text[0] = self._format_output()
@kb.add("c-c")
def handle_ctrl_c(event):
input_buffer.reset()
@kb.add("c-d")
def handle_ctrl_d(event):
if not input_buffer.text:
self._running = False
event.app.exit()
@kb.add("c-q")
def handle_ctrl_q(event):
self._running = False
event.app.exit()
# Style
style = Style.from_dict({
"output-area": "bg:#0d0d0d",
"input-area": "bg:#1a1a1a",
"status-bar": "bg:#333333 #888888",
"prompt": "#00aaff bold",
})
# Status bar text
def get_status_bar():
left = "? for help"
center = self.config.model
right = f"Session: {self.session_id}"
return f" {left} | {center} | {right} "
# Layout: Output (top, scrollable) + Status bar + Input (bottom, fixed)
layout = Layout(
HSplit([
# Output area (takes remaining space)
Window(
content=FormattedTextControl(get_output_text),
wrap_lines=True,
style="class:output-area",
),
# Status bar (1 line)
Window(
content=FormattedTextControl(get_status_bar),
height=1,
style="class:status-bar",
),
# Input area (fixed at bottom, 3 lines for visibility)
HSplit([
Window(height=1), # Spacer
VSplit([
Window(
content=FormattedTextControl([("class:prompt", "› ")]),
width=2,
),
Window(
content=BufferControl(buffer=input_buffer),
wrap_lines=True,
),
]),
Window(height=1), # Spacer
], height=3, style="class:input-area"),
])
)
# Create application
app = Application(
layout=layout,
key_bindings=kb,
style=style,
full_screen=True,
mouse_support=True,
)
# Run
app.run()
print("\n Goodbye from Praison AI!")
def run_single(self, prompt: str) -> Optional[str]:
"""Run a single prompt (non-interactive)."""
return self._execute_prompt(prompt)
def start_split_tui(
model: str = "gpt-4o-mini",
show_logo: bool = True,
**kwargs
) -> None:
"""Start the split-pane TUI application."""
config = SplitTUIConfig(
model=model,
show_logo=show_logo,
)
tui = SplitTUI(config=config)
tui.run()
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==3.7.5 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==3.7.7 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "3.7.5"
__version__ = "3.7.7"
@@ -0,0 +1,600 @@
"""
Tests for PraisonAI Async TUI Application.
"""
class TestAsyncTUIConfig:
"""Tests for AsyncTUIConfig dataclass."""
def test_default_values(self):
"""Test default configuration values."""
from praisonai.cli.interactive.async_tui import AsyncTUIConfig
config = AsyncTUIConfig()
assert config.model == "gpt-4o-mini"
assert config.show_logo is True
assert config.show_status_bar is True
assert config.session_id is None
assert config.workspace is None
def test_custom_values(self):
"""Test custom configuration values."""
from praisonai.cli.interactive.async_tui import AsyncTUIConfig
config = AsyncTUIConfig(
model="gpt-4o",
show_logo=False,
session_id="test123",
)
assert config.model == "gpt-4o"
assert config.show_logo is False
assert config.session_id == "test123"
class TestChatMessage:
"""Tests for ChatMessage dataclass."""
def test_message_creation(self):
"""Test message creation."""
from praisonai.cli.interactive.async_tui import ChatMessage
msg = ChatMessage(role="user", content="Hello")
assert msg.role == "user"
assert msg.content == "Hello"
assert msg.timestamp is not None
def test_message_roles(self):
"""Test different message roles."""
from praisonai.cli.interactive.async_tui import ChatMessage
roles = ["user", "assistant", "system", "status"]
for role in roles:
msg = ChatMessage(role=role, content="test")
assert msg.role == role
class TestGetLogo:
"""Tests for get_logo function."""
def test_logo_wide_terminal(self):
"""Test logo for wide terminal."""
from praisonai.cli.interactive.async_tui import get_logo, LOGO
logo = get_logo(100)
assert logo == LOGO
def test_logo_medium_terminal(self):
"""Test logo for medium terminal."""
from praisonai.cli.interactive.async_tui import get_logo, LOGO_SMALL
logo = get_logo(50)
assert logo == LOGO_SMALL
def test_logo_narrow_terminal(self):
"""Test logo for narrow terminal."""
from praisonai.cli.interactive.async_tui import get_logo, LOGO_MINIMAL
logo = get_logo(30)
assert logo == LOGO_MINIMAL
def test_logo_contains_praison_ai(self):
"""Test that logo contains 'Praison AI' branding."""
from praisonai.cli.interactive.async_tui import LOGO_MINIMAL
assert "Praison AI" in LOGO_MINIMAL
class TestAsyncTUI:
"""Tests for AsyncTUI class."""
def test_init_default(self):
"""Test default initialization."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert tui.config is not None
assert tui.messages == []
assert tui._running is False
assert tui._processing is False
assert tui.session_id is not None
def test_init_with_config(self):
"""Test initialization with config."""
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
config = AsyncTUIConfig(model="gpt-4o", show_logo=False)
tui = AsyncTUI(config=config)
assert tui.config.model == "gpt-4o"
assert tui.config.show_logo is False
def test_format_output_empty(self):
"""Test format_output with no messages."""
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
config = AsyncTUIConfig(show_logo=True)
tui = AsyncTUI(config=config)
output = tui._format_output()
assert "Type your message" in output or "gpt-4o-mini" in output
def test_format_output_with_messages(self):
"""Test format_output with messages."""
from praisonai.cli.interactive.async_tui import AsyncTUI, ChatMessage
tui = AsyncTUI()
tui.messages = [
ChatMessage(role="user", content="Hello"),
ChatMessage(role="assistant", content="Hi there"),
]
output = tui._format_output()
assert "Hello" in output
assert "Hi there" in output
def test_format_output_with_processing(self):
"""Test format_output shows processing indicator."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._processing = True
tui._status_text = "Thinking..."
output = tui._format_output()
assert "Thinking..." in output
def test_handle_command_exit(self):
"""Test handling exit command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._running = True
result = tui._handle_command("/exit")
assert result is True
assert tui._running is False
def test_handle_command_quit(self):
"""Test handling quit command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._running = True
result = tui._handle_command("/quit")
assert result is True
assert tui._running is False
def test_handle_command_help(self):
"""Test handling help command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/help")
assert result is True
assert len(tui.messages) == 1
assert tui.messages[0].role == "system"
assert "PageUp" in tui.messages[0].content # Scroll instructions
def test_handle_command_clear(self):
"""Test handling clear command."""
from praisonai.cli.interactive.async_tui import AsyncTUI, ChatMessage
tui = AsyncTUI()
tui.messages = [ChatMessage(role="user", content="test")]
result = tui._handle_command("/clear")
assert result is True
# Clear now adds a confirmation message
assert len(tui.messages) == 1
assert "cleared" in tui.messages[0].content.lower()
def test_handle_command_new(self):
"""Test handling new command."""
from praisonai.cli.interactive.async_tui import AsyncTUI, ChatMessage
tui = AsyncTUI()
old_session = tui.session_id
tui.messages = [ChatMessage(role="user", content="test")]
result = tui._handle_command("/new")
assert result is True
assert len(tui.messages) == 1
assert tui.session_id != old_session
def test_handle_command_model_show(self):
"""Test handling model command (show)."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/model")
assert result is True
assert len(tui.messages) == 1
assert "gpt-4o-mini" in tui.messages[0].content
def test_handle_command_model_change(self):
"""Test handling model command (change)."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/model gpt-4o")
assert result is True
assert tui.config.model == "gpt-4o"
assert tui._agent is None
def test_handle_command_unknown(self):
"""Test handling unknown command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/unknown")
assert result is True
assert len(tui.messages) == 1
assert "Unknown" in tui.messages[0].content
class TestStartAsyncTUI:
"""Tests for start_async_tui function."""
def test_function_exists(self):
"""Test that start_async_tui function exists."""
from praisonai.cli.interactive.async_tui import start_async_tui
assert callable(start_async_tui)
class TestASCIILogos:
"""Tests for ASCII logo constants."""
def test_logo_exists(self):
"""Test LOGO constant exists and has content."""
from praisonai.cli.interactive.async_tui import LOGO
assert len(LOGO) > 0
assert "██" in LOGO
def test_logo_small_exists(self):
"""Test LOGO_SMALL constant exists."""
from praisonai.cli.interactive.async_tui import LOGO_SMALL
assert len(LOGO_SMALL) > 0
def test_logo_minimal_branding(self):
"""Test LOGO_MINIMAL has correct branding."""
from praisonai.cli.interactive.async_tui import LOGO_MINIMAL
assert "Praison AI" in LOGO_MINIMAL
class TestMessageFormatting:
"""Tests for message formatting."""
def test_user_message_prefix(self):
"""Test user messages have correct prefix."""
from praisonai.cli.interactive.async_tui import AsyncTUI, ChatMessage
tui = AsyncTUI()
tui.messages = [ChatMessage(role="user", content="test")]
output = tui._format_output()
assert "› test" in output
def test_assistant_message_prefix(self):
"""Test assistant messages have correct prefix."""
from praisonai.cli.interactive.async_tui import AsyncTUI, ChatMessage
tui = AsyncTUI()
tui.messages = [ChatMessage(role="assistant", content="response")]
output = tui._format_output()
assert "● response" in output
def test_status_message_prefix(self):
"""Test status messages have correct prefix."""
from praisonai.cli.interactive.async_tui import AsyncTUI, ChatMessage
tui = AsyncTUI()
tui.messages = [ChatMessage(role="status", content="thinking")]
output = tui._format_output()
assert "⏳ thinking" in output
class TestNonBlockingExecution:
"""Tests for non-blocking execution features."""
def test_processing_flag(self):
"""Test processing flag is used correctly."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert tui._processing is False
tui._processing = True
tui._status_text = "Working..."
output = tui._format_output()
assert "Working..." in output
def test_status_text_updates(self):
"""Test status text can be updated."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._processing = True
tui._status_text = "Step 1"
output1 = tui._format_output()
assert "Step 1" in output1
tui._status_text = "Step 2"
output2 = tui._format_output()
assert "Step 2" in output2
class TestNewSlashCommands:
"""Tests for newly added slash commands."""
def test_session_command(self):
"""Test /session command shows session info."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/session")
assert result is True
assert len(tui.messages) == 1
assert "Session:" in tui.messages[0].content
assert tui.session_id in tui.messages[0].content
def test_history_command_empty(self):
"""Test /history command with no history."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/history")
assert result is True
assert "No conversation history" in tui.messages[0].content
def test_history_command_with_history(self):
"""Test /history command with conversation history."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._conversation_history = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there"},
]
result = tui._handle_command("/history")
assert result is True
assert "[user]" in tui.messages[0].content
def test_cost_command(self):
"""Test /cost command shows token usage."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._total_tokens = 100
tui._total_cost = 0.001
result = tui._handle_command("/cost")
assert result is True
assert "100" in tui.messages[0].content
def test_compact_command(self):
"""Test /compact command toggles compact mode."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert tui.config.compact_mode is False
result = tui._handle_command("/compact")
assert result is True
assert tui.config.compact_mode is True
result = tui._handle_command("/compact")
assert tui.config.compact_mode is False
def test_multiline_command(self):
"""Test /multiline command toggles multiline mode."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert tui.config.multiline_mode is False
result = tui._handle_command("/multiline")
assert result is True
assert tui.config.multiline_mode is True
def test_files_command(self):
"""Test /files command lists workspace files."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._workspace_files = ["test.py", "main.py"]
result = tui._handle_command("/files")
assert result is True
assert "@test.py" in tui.messages[0].content
def test_queue_command_empty(self):
"""Test /queue command with empty queue."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/queue")
assert result is True
assert "No prompts in queue" in tui.messages[0].content
def test_queue_command_with_items(self):
"""Test /queue command with queued prompts."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._prompt_queue = ["First prompt", "Second prompt"]
result = tui._handle_command("/queue")
assert result is True
assert "First prompt" in tui.messages[0].content
class TestQueueSystem:
"""Tests for prompt queue system."""
def test_queue_initialization(self):
"""Test queue is initialized empty."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert tui._prompt_queue == []
def test_queue_or_execute_when_not_processing(self):
"""Test _queue_or_execute executes immediately when not processing."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tui._processing = False
# This would normally start background execution
# Just test that it adds user message
tui._queue_or_execute("test prompt")
assert len(tui.messages) == 1
assert tui.messages[0].role == "user"
assert tui.messages[0].content == "test prompt"
class TestFileProcessing:
"""Tests for @file mention processing."""
def test_process_file_mentions_no_mentions(self):
"""Test processing prompt with no @file mentions."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._process_file_mentions("Hello world")
assert result == "Hello world"
def test_process_file_mentions_with_mention(self):
"""Test processing prompt with @file mention."""
from praisonai.cli.interactive.async_tui import AsyncTUI
import tempfile
import os
tui = AsyncTUI()
# Create a temp file
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
f.write("test content")
temp_path = f.name
try:
tui.config.workspace = os.path.dirname(temp_path)
filename = os.path.basename(temp_path)
result = tui._process_file_mentions(f"Check @{filename}")
assert "test content" in result
finally:
os.unlink(temp_path)
class TestWorkspaceScanning:
"""Tests for workspace file scanning."""
def test_workspace_files_initialized(self):
"""Test workspace files list is initialized."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert isinstance(tui._workspace_files, list)
def test_scan_workspace_limits_files(self):
"""Test workspace scanning limits files (may exceed slightly due to batch processing)."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
# Verify the limit is approximately enforced (may exceed slightly due to batch)
assert len(tui._workspace_files) <= 1100 # Allow some margin
class TestToolLoading:
"""Tests for tool loading functionality."""
def test_load_tools_returns_list(self):
"""Test _load_tools returns a list."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
tools = tui._load_tools()
assert isinstance(tools, list)
def test_load_tools_method_exists(self):
"""Test _load_tools method exists."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert hasattr(tui, '_load_tools')
assert callable(tui._load_tools)
class TestNewSessionCommands:
"""Tests for session-related commands."""
def test_sessions_command(self):
"""Test /sessions command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/sessions")
assert result is True
# Should have a message about sessions
assert len(tui.messages) >= 1
def test_continue_command(self):
"""Test /continue command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/continue")
assert result is True
assert len(tui.messages) >= 1
def test_import_command_no_args(self):
"""Test /import command without arguments."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/import")
assert result is True
assert "Usage" in tui.messages[0].content
def test_import_command_file_not_found(self):
"""Test /import command with non-existent file."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/import nonexistent_file.json")
assert result is True
assert "not found" in tui.messages[0].content.lower() or "failed" in tui.messages[0].content.lower()
def test_status_command(self):
"""Test /status command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/status")
assert result is True
# Should show runtime status or "not initialized"
assert len(tui.messages) >= 1
class TestRuntimeIntegration:
"""Tests for runtime integration."""
def test_runtime_fields_initialized(self):
"""Test runtime fields are initialized."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert tui._runtime is None
assert tui._runtime_started is False
def test_start_runtime_method_exists(self):
"""Test _start_runtime method exists."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert hasattr(tui, '_start_runtime')
# It's an async method
import asyncio
assert asyncio.iscoroutinefunction(tui._start_runtime)
@@ -0,0 +1,282 @@
"""
Tests for PraisonAI Split-Pane TUI Application.
"""
class TestSplitTUIConfig:
"""Tests for SplitTUIConfig dataclass."""
def test_default_values(self):
"""Test default configuration values."""
from praisonai.cli.interactive.split_tui import SplitTUIConfig
config = SplitTUIConfig()
assert config.model == "gpt-4o-mini"
assert config.show_logo is True
assert config.show_status_bar is True
assert config.session_id is None
assert config.workspace is None
def test_custom_values(self):
"""Test custom configuration values."""
from praisonai.cli.interactive.split_tui import SplitTUIConfig
config = SplitTUIConfig(
model="gpt-4o",
show_logo=False,
session_id="test123",
)
assert config.model == "gpt-4o"
assert config.show_logo is False
assert config.session_id == "test123"
class TestChatMessage:
"""Tests for ChatMessage dataclass."""
def test_message_creation(self):
"""Test message creation."""
from praisonai.cli.interactive.split_tui import ChatMessage
msg = ChatMessage(role="user", content="Hello")
assert msg.role == "user"
assert msg.content == "Hello"
assert msg.timestamp is not None
def test_message_roles(self):
"""Test different message roles."""
from praisonai.cli.interactive.split_tui import ChatMessage
roles = ["user", "assistant", "system", "status"]
for role in roles:
msg = ChatMessage(role=role, content="test")
assert msg.role == role
class TestGetLogo:
"""Tests for get_logo function."""
def test_logo_wide_terminal(self):
"""Test logo for wide terminal."""
from praisonai.cli.interactive.split_tui import get_logo, LOGO
logo = get_logo(100)
assert logo == LOGO
def test_logo_medium_terminal(self):
"""Test logo for medium terminal."""
from praisonai.cli.interactive.split_tui import get_logo, LOGO_SMALL
logo = get_logo(50)
assert logo == LOGO_SMALL
def test_logo_narrow_terminal(self):
"""Test logo for narrow terminal."""
from praisonai.cli.interactive.split_tui import get_logo, LOGO_MINIMAL
logo = get_logo(30)
assert logo == LOGO_MINIMAL
def test_logo_contains_praison_ai(self):
"""Test that logo contains 'Praison AI' branding."""
from praisonai.cli.interactive.split_tui import LOGO_MINIMAL
assert "Praison AI" in LOGO_MINIMAL
class TestSplitTUI:
"""Tests for SplitTUI class."""
def test_init_default(self):
"""Test default initialization."""
from praisonai.cli.interactive.split_tui import SplitTUI
tui = SplitTUI()
assert tui.config is not None
assert tui.messages == []
assert tui._running is False
assert tui.session_id is not None
def test_init_with_config(self):
"""Test initialization with config."""
from praisonai.cli.interactive.split_tui import SplitTUI, SplitTUIConfig
config = SplitTUIConfig(model="gpt-4o", show_logo=False)
tui = SplitTUI(config=config)
assert tui.config.model == "gpt-4o"
assert tui.config.show_logo is False
def test_format_output_empty(self):
"""Test format_output with no messages."""
from praisonai.cli.interactive.split_tui import SplitTUI, SplitTUIConfig
config = SplitTUIConfig(show_logo=True)
tui = SplitTUI(config=config)
output = tui._format_output()
# Should contain welcome message
assert "Type your message" in output or "gpt-4o-mini" in output
def test_format_output_with_messages(self):
"""Test format_output with messages."""
from praisonai.cli.interactive.split_tui import SplitTUI, ChatMessage
tui = SplitTUI()
tui.messages = [
ChatMessage(role="user", content="Hello"),
ChatMessage(role="assistant", content="Hi there"),
]
output = tui._format_output()
assert "Hello" in output
assert "Hi there" in output
def test_handle_command_exit(self):
"""Test handling exit command."""
from praisonai.cli.interactive.split_tui import SplitTUI
tui = SplitTUI()
tui._running = True
result = tui._handle_command("/exit")
assert result is True
assert tui._running is False
def test_handle_command_quit(self):
"""Test handling quit command."""
from praisonai.cli.interactive.split_tui import SplitTUI
tui = SplitTUI()
tui._running = True
result = tui._handle_command("/quit")
assert result is True
assert tui._running is False
def test_handle_command_help(self):
"""Test handling help command."""
from praisonai.cli.interactive.split_tui import SplitTUI
tui = SplitTUI()
result = tui._handle_command("/help")
assert result is True
assert len(tui.messages) == 1
assert tui.messages[0].role == "system"
assert "/help" in tui.messages[0].content
def test_handle_command_clear(self):
"""Test handling clear command."""
from praisonai.cli.interactive.split_tui import SplitTUI, ChatMessage
tui = SplitTUI()
tui.messages = [ChatMessage(role="user", content="test")]
result = tui._handle_command("/clear")
assert result is True
assert len(tui.messages) == 0
def test_handle_command_new(self):
"""Test handling new command."""
from praisonai.cli.interactive.split_tui import SplitTUI, ChatMessage
tui = SplitTUI()
old_session = tui.session_id
tui.messages = [ChatMessage(role="user", content="test")]
result = tui._handle_command("/new")
assert result is True
assert len(tui.messages) == 1 # New session message
assert tui.session_id != old_session
def test_handle_command_model_show(self):
"""Test handling model command (show)."""
from praisonai.cli.interactive.split_tui import SplitTUI
tui = SplitTUI()
result = tui._handle_command("/model")
assert result is True
assert len(tui.messages) == 1
assert "gpt-4o-mini" in tui.messages[0].content
def test_handle_command_model_change(self):
"""Test handling model command (change)."""
from praisonai.cli.interactive.split_tui import SplitTUI
tui = SplitTUI()
result = tui._handle_command("/model gpt-4o")
assert result is True
assert tui.config.model == "gpt-4o"
assert tui._agent is None # Agent should be reset
def test_handle_command_unknown(self):
"""Test handling unknown command."""
from praisonai.cli.interactive.split_tui import SplitTUI
tui = SplitTUI()
result = tui._handle_command("/unknown")
assert result is True
assert len(tui.messages) == 1
assert "Unknown" in tui.messages[0].content
class TestStartSplitTUI:
"""Tests for start_split_tui function."""
def test_function_exists(self):
"""Test that start_split_tui function exists."""
from praisonai.cli.interactive.split_tui import start_split_tui
assert callable(start_split_tui)
class TestASCIILogos:
"""Tests for ASCII logo constants."""
def test_logo_exists(self):
"""Test LOGO constant exists and has content."""
from praisonai.cli.interactive.split_tui import LOGO
assert len(LOGO) > 0
assert "██" in LOGO # Contains block characters
def test_logo_small_exists(self):
"""Test LOGO_SMALL constant exists."""
from praisonai.cli.interactive.split_tui import LOGO_SMALL
assert len(LOGO_SMALL) > 0
def test_logo_minimal_branding(self):
"""Test LOGO_MINIMAL has correct branding."""
from praisonai.cli.interactive.split_tui import LOGO_MINIMAL
assert "Praison AI" in LOGO_MINIMAL
class TestMessageFormatting:
"""Tests for message formatting."""
def test_user_message_prefix(self):
"""Test user messages have correct prefix."""
from praisonai.cli.interactive.split_tui import SplitTUI, ChatMessage
tui = SplitTUI()
tui.messages = [ChatMessage(role="user", content="test")]
output = tui._format_output()
assert "› test" in output
def test_assistant_message_prefix(self):
"""Test assistant messages have correct prefix."""
from praisonai.cli.interactive.split_tui import SplitTUI, ChatMessage
tui = SplitTUI()
tui.messages = [ChatMessage(role="assistant", content="response")]
output = tui._format_output()
assert "● response" in output
def test_status_message_prefix(self):
"""Test status messages have correct prefix."""
from praisonai.cli.interactive.split_tui import SplitTUI, ChatMessage
tui = SplitTUI()
tui.messages = [ChatMessage(role="status", content="thinking")]
output = tui._format_output()
assert "⏳ thinking" in output