Compare commits

...
7 Commits
33 changed files with 4880 additions and 101 deletions

No files matched your search

+57
View File
@@ -0,0 +1,57 @@
name: Update API Reference
on:
push:
branches: [ "main" ]
paths:
- 'src/praisonai-agents/praisonaiagents/**/*.py'
- 'src/praisonai/praisonai/**/*.py'
- 'src/praisonai-ts/src/**/*.ts'
- '.github/workflows/update-api-md.yml'
permissions:
contents: write
jobs:
update-api-md:
runs-on: ubuntu-latest
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
ref: main
persist-credentials: true
fetch-depth: 0
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Generate api.md
run: |
cd $GITHUB_WORKSPACE
python3 src/praisonai/scripts/generate_api_md.py
- name: Check for changes
id: check_changes
run: |
if git diff --quiet api.md; then
echo "changes=false" >> $GITHUB_OUTPUT
echo "No changes to api.md"
else
echo "changes=true" >> $GITHUB_OUTPUT
echo "api.md has been updated"
git diff api.md | head -50
fi
- name: Commit and push changes
if: steps.check_changes.outputs.changes == 'true'
uses: stefanzweifel/git-auto-commit-action@v5
with:
commit_message: "docs: auto-update api.md [skip ci]"
file_pattern: 'api.md'
commit_user_name: "github-actions[bot]"
commit_user_email: "41898282+github-actions[bot]@users.noreply.github.com"
commit_author: "github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>"
+6 -1
View File
@@ -552,7 +552,7 @@ Methods:
Types:
```python
from praisonaiagents import AutoRagConfig, Citation, CodeExecutionStep, DeepResearchResponse, ExpandResult, ExpandStrategy, FileSearchCall, MCPCall, Provider, RagRetrievalPolicy, ReasoningStep, RewriteResult, RewriteStrategy, Task, WebSearchCall
from praisonaiagents import AutoRagConfig, Citation, CodeExecutionStep, DeepResearchResponse, ExpandResult, ExpandStrategy, FileSearchCall, MCPCall, Provider, RagRetrievalPolicy, ReasoningStep, RewriteResult, RewriteStrategy, Task, WebSearchCall, warmup
```
Methods:
@@ -569,6 +569,7 @@ Methods:
* <code title="class Task">Task.<a href="./src/praisonai-agents/praisonaiagents/task/task.py">execute_callback_sync</a>(task_output: TaskOutput) -> None</code>
* <code title="class Task">Task.<a href="./src/praisonai-agents/praisonaiagents/task/task.py">initialize_memory</a>()</code>
* <code title="class Task">Task.<a href="./src/praisonai-agents/praisonaiagents/task/task.py">store_in_memory</a>(content: str, agent_name: str = None, task_id: str = None)</code>
* <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/__init__.py">warmup</a>(include_litellm: bool = False, include_openai: bool = True) -> dict</code>
# Wrapper (praisonai)
@@ -731,6 +732,10 @@ Methods:
* <code title="cli">praisonai skills install <a href="./src/praisonai/praisonai/cli/commands/skills.py">--help</a></code>
* <code title="cli">praisonai skills list <a href="./src/praisonai/praisonai/cli/commands/skills.py">--help</a></code>
* <code title="cli">praisonai skills validate <a href="./src/praisonai/praisonai/cli/commands/skills.py">--help</a></code>
* <code title="cli">praisonai standardise check <a href="./src/praisonai/praisonai/cli/commands/standardise.py">--help</a></code>
* <code title="cli">praisonai standardise fix <a href="./src/praisonai/praisonai/cli/commands/standardise.py">--help</a></code>
* <code title="cli">praisonai standardise init <a href="./src/praisonai/praisonai/cli/commands/standardise.py">--help</a></code>
* <code title="cli">praisonai standardise report <a href="./src/praisonai/praisonai/cli/commands/standardise.py">--help</a></code>
* <code title="cli">praisonai templates create <a href="./src/praisonai/praisonai/cli/commands/templates.py">--help</a></code>
* <code title="cli">praisonai templates list <a href="./src/praisonai/praisonai/cli/commands/templates.py">--help</a></code>
* <code title="cli">praisonai test info <a href="./src/praisonai/praisonai/cli/commands/test.py">--help</a></code>
+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.3" \
"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.3" \
"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.3" \
"praisonai>=3.7.7" \
"praisonai[ui]" \
"praisonai[crewai]"
Whitespace-only changes.
@@ -649,6 +649,7 @@ class OpenAIClient:
chunks = []
first_token_emitted = False
last_content_time = None
ttft_logged = False
# If display function provided, use Live display
if display_fn:
@@ -669,6 +670,12 @@ class OpenAIClient:
full_response_text += content
last_content_time = time.perf_counter()
# Log TTFT on first token
if not ttft_logged:
ttft_ms = (last_content_time - request_start_perf) * 1000
logging.debug(f"TTFT (Time To First Token): {ttft_ms:.0f}ms")
ttft_logged = True
# Emit FIRST_TOKEN on first content
if _emit and not first_token_emitted:
stream_callback(StreamEvent(
@@ -715,11 +722,17 @@ class OpenAIClient:
for chunk in response_stream:
chunks.append(chunk)
if _emit and chunk.choices and chunk.choices[0].delta.content:
if chunk.choices and chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
last_content_time = time.perf_counter()
if not first_token_emitted:
# Log TTFT on first token
if not ttft_logged:
ttft_ms = (last_content_time - request_start_perf) * 1000
logging.debug(f"TTFT (Time To First Token): {ttft_ms:.0f}ms")
ttft_logged = True
if _emit and not first_token_emitted:
stream_callback(StreamEvent(
type=StreamEventType.FIRST_TOKEN,
timestamp=last_content_time,
@@ -1130,7 +1143,9 @@ class OpenAIClient:
if display_fn and console:
# When verbose (display_fn provided), use streaming for better UX
try:
with Live(display_fn("", start_time), console=console, refresh_per_second=4, transient=True) as live:
request_start_perf = time.perf_counter()
ttft_logged = False
with Live(display_fn("", start_time), console=console, refresh_per_second=10, transient=True) as live:
# Use streaming when display_fn is provided for progressive display
response_stream = self.create_completion(
messages=messages,
@@ -1148,6 +1163,11 @@ class OpenAIClient:
for chunk in response_stream:
chunks.append(chunk)
if chunk.choices[0].delta.content:
# Log TTFT on first token
if not ttft_logged:
ttft_ms = (time.perf_counter() - request_start_perf) * 1000
logging.debug(f"TTFT (Time To First Token): {ttft_ms:.0f}ms")
ttft_logged = True
full_response_text += chunk.choices[0].delta.content
live.update(display_fn(full_response_text, start_time))
@@ -0,0 +1,232 @@
#!/usr/bin/env python3
"""
Detailed profiler to identify the exact source of streaming delay.
Traces every step from agent.start() to HTTP request.
"""
import os
import sys
import time
import functools
os.environ['LOGLEVEL'] = 'DEBUG'
# Microsecond precision timestamps
def ts():
return time.perf_counter()
start_time = ts()
def log(msg):
elapsed = (ts() - start_time) * 1000
print(f"[{elapsed:8.2f}ms] {msg}")
log("Script starting...")
# ============================================================
# PATCH 1: OpenAI client initialization
# ============================================================
import openai
original_openai_init = openai.OpenAI.__init__
@functools.wraps(original_openai_init)
def traced_openai_init(self, *args, **kwargs):
t0 = ts()
result = original_openai_init(self, *args, **kwargs)
log(f"OpenAI.__init__() took {(ts()-t0)*1000:.2f}ms")
return result
openai.OpenAI.__init__ = traced_openai_init
# ============================================================
# PATCH 2: OpenAI completions.create
# ============================================================
original_create = openai.resources.chat.completions.Completions.create
@functools.wraps(original_create)
def traced_create(self, *args, **kwargs):
log(f">>> OpenAI.create() CALLED - stream={kwargs.get('stream', False)}")
t0 = ts()
result = original_create(self, *args, **kwargs)
log(f"<<< OpenAI.create() RETURNED in {(ts()-t0)*1000:.2f}ms")
return result
openai.resources.chat.completions.Completions.create = traced_create
# ============================================================
# PATCH 3: httpx.Client.send (actual HTTP request)
# ============================================================
import httpx
original_send = httpx.Client.send
@functools.wraps(original_send)
def traced_send(self, request, *args, **kwargs):
log(f">>> HTTPX.send() {request.method} {request.url}")
t0 = ts()
result = original_send(self, request, *args, **kwargs)
log(f"<<< HTTPX.send() returned in {(ts()-t0)*1000:.2f}ms - status={result.status_code}")
return result
httpx.Client.send = traced_send
# ============================================================
# PATCH 4: Rich Live display
# ============================================================
from rich.live import Live
original_live_init = Live.__init__
original_live_enter = Live.__enter__
original_live_exit = Live.__exit__
@functools.wraps(original_live_init)
def traced_live_init(self, *args, **kwargs):
log(f">>> Live.__init__() called")
t0 = ts()
result = original_live_init(self, *args, **kwargs)
log(f"<<< Live.__init__() took {(ts()-t0)*1000:.2f}ms")
return result
@functools.wraps(original_live_enter)
def traced_live_enter(self):
log(f">>> Live.__enter__() called")
t0 = ts()
result = original_live_enter(self)
log(f"<<< Live.__enter__() took {(ts()-t0)*1000:.2f}ms")
return result
@functools.wraps(original_live_exit)
def traced_live_exit(self, *args):
log(f">>> Live.__exit__() called")
t0 = ts()
result = original_live_exit(self, *args)
log(f"<<< Live.__exit__() took {(ts()-t0)*1000:.2f}ms")
return result
Live.__init__ = traced_live_init
Live.__enter__ = traced_live_enter
Live.__exit__ = traced_live_exit
log("Patches applied, importing praisonaiagents...")
# ============================================================
# Import and trace Agent
# ============================================================
t0 = ts()
from praisonaiagents import Agent
log(f"Agent import took {(ts()-t0)*1000:.2f}ms")
# ============================================================
# PATCH 5: Agent methods
# ============================================================
from praisonaiagents.agent.agent import Agent as AgentClass
original_chat_completion = AgentClass._chat_completion
original_format_tools = AgentClass._format_tools_for_completion
@functools.wraps(original_chat_completion)
def traced_chat_completion(self, messages, temperature=1.0, tools=None, stream=True,
reasoning_steps=False, task_name=None, task_description=None, task_id=None):
log(f">>> Agent._chat_completion() stream={stream}, _using_custom_llm={self._using_custom_llm}")
t0 = ts()
result = original_chat_completion(self, messages, temperature, tools, stream,
reasoning_steps, task_name, task_description, task_id)
log(f"<<< Agent._chat_completion() took {(ts()-t0)*1000:.2f}ms")
return result
@functools.wraps(original_format_tools)
def traced_format_tools(self, tools):
log(f">>> Agent._format_tools_for_completion() tools={tools}")
t0 = ts()
result = original_format_tools(self, tools)
log(f"<<< Agent._format_tools_for_completion() took {(ts()-t0)*1000:.2f}ms")
return result
AgentClass._chat_completion = traced_chat_completion
AgentClass._format_tools_for_completion = traced_format_tools
# ============================================================
# PATCH 6: OpenAIClient methods
# ============================================================
from praisonaiagents.llm.openai_client import OpenAIClient
original_chat_with_tools = OpenAIClient.chat_completion_with_tools
original_process_stream = OpenAIClient.process_stream_response
original_create_completion = OpenAIClient.create_completion
original_format_tools_client = OpenAIClient.format_tools
@functools.wraps(original_chat_with_tools)
def traced_chat_with_tools(self, *args, **kwargs):
log(f">>> OpenAIClient.chat_completion_with_tools() stream={kwargs.get('stream', True)}")
t0 = ts()
result = original_chat_with_tools(self, *args, **kwargs)
log(f"<<< OpenAIClient.chat_completion_with_tools() took {(ts()-t0)*1000:.2f}ms")
return result
@functools.wraps(original_process_stream)
def traced_process_stream(self, *args, **kwargs):
log(f">>> OpenAIClient.process_stream_response()")
t0 = ts()
result = original_process_stream(self, *args, **kwargs)
log(f"<<< OpenAIClient.process_stream_response() took {(ts()-t0)*1000:.2f}ms")
return result
@functools.wraps(original_create_completion)
def traced_create_completion(self, *args, **kwargs):
log(f">>> OpenAIClient.create_completion() stream={kwargs.get('stream', False)}")
t0 = ts()
result = original_create_completion(self, *args, **kwargs)
log(f"<<< OpenAIClient.create_completion() took {(ts()-t0)*1000:.2f}ms")
return result
@functools.wraps(original_format_tools_client)
def traced_format_tools_client(self, tools):
log(f">>> OpenAIClient.format_tools() tools count={len(tools) if tools else 0}")
t0 = ts()
result = original_format_tools_client(self, tools)
log(f"<<< OpenAIClient.format_tools() took {(ts()-t0)*1000:.2f}ms")
return result
OpenAIClient.chat_completion_with_tools = traced_chat_with_tools
OpenAIClient.process_stream_response = traced_process_stream
OpenAIClient.create_completion = traced_create_completion
OpenAIClient.format_tools = traced_format_tools_client
# ============================================================
# PATCH 7: sync_client property (lazy init)
# ============================================================
original_sync_client_getter = OpenAIClient.sync_client.fget
def traced_sync_client_getter(self):
log(f">>> OpenAIClient.sync_client property accessed")
t0 = ts()
result = original_sync_client_getter(self)
log(f"<<< OpenAIClient.sync_client took {(ts()-t0)*1000:.2f}ms")
return result
OpenAIClient.sync_client = property(traced_sync_client_getter)
log("All patches applied. Creating agent...")
# ============================================================
# RUN THE TEST
# ============================================================
t0 = ts()
agent = Agent(
instructions="You are a helpful assistant"
)
log(f"Agent created in {(ts()-t0)*1000:.2f}ms")
log(f"Agent.stream={agent.stream}, Agent.verbose={agent.verbose}")
log("=" * 60)
log("Starting agent.start()...")
log("=" * 60)
t0 = ts()
result = agent.start("Say hello in one word")
log(f"agent.start() completed in {(ts()-t0)*1000:.2f}ms")
log("=" * 60)
log("DONE")
log("=" * 60)
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.11.2"
version = "0.11.3"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
Whitespace-only changes.
Whitespace-only changes.
View File
Whitespace-only changes.
Whitespace-only changes.
Whitespace-only changes.
Whitespace-only changes.
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.11.2"
version = "0.11.3"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+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.3.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.3.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]"
+11 -4
View File
@@ -141,11 +141,18 @@ def main_callback(
)
set_output_controller(state.output_controller)
# If no command provided, show help or handle legacy behavior
# If no command provided, start interactive mode
if ctx.invoked_subcommand is None:
# Check if there are remaining args (legacy behavior)
# This is handled by the legacy adapter in main.py
pass
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
tui_config = AsyncTUIConfig(
model="gpt-4o-mini",
show_logo=True,
show_status_bar=state.output_format != OutputFormat.json,
)
tui = AsyncTUI(config=tui_config)
tui.run()
def get_output_controller() -> OutputController:
+23 -66
View File
@@ -149,25 +149,12 @@ def chat_main(
praisonai chat "Summarize this" --file README.md
praisonai chat "What is 2+2?" --profile
"""
import asyncio
import os
# Set workspace if provided
if workspace:
os.environ["PRAISONAI_WORKSPACE"] = workspace
# Configure UI backend
from praisonai.cli.ui import select_backend, UIConfig
ui_config = UIConfig(
ui_backend=ui_backend,
json_output=json_output,
no_color=no_color,
theme=theme,
compact=compact,
verbose=verbose,
)
backend = select_backend(ui_config)
# Handle profiling for single prompt mode
if prompt and (profile or profile_deep):
_run_profiled_chat(
@@ -184,60 +171,30 @@ 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)
# Try InteractiveCore first (preferred terminal-native implementation)
try:
from praisonai.cli.interactive import InteractiveCore, InteractiveConfig
from praisonai.cli.interactive.frontends import RichFrontend
config = InteractiveConfig(
model=model,
session_id=session_id,
continue_session=continue_session,
workspace=workspace or None,
verbose=verbose,
memory=memory_value,
files=list(file) if file else [],
autonomy=autonomy,
)
core = InteractiveCore(config=config)
if prompt:
# Single prompt mode - use backend for output
from praisonai.cli.ui.events import UIEventType
async def run_prompt():
if continue_session:
core.continue_session()
backend.emit(UIEventType.MESSAGE_START, {})
response = await core.prompt(prompt)
backend.emit(UIEventType.MESSAGE_CHUNK, {'content': response})
backend.emit(UIEventType.MESSAGE_END, {})
asyncio.run(run_prompt())
else:
# Interactive REPL mode - pass ui_config to frontend
frontend = RichFrontend(core=core, config=config, ui_config=ui_config)
asyncio.run(frontend.run())
except ImportError:
# Fallback to legacy terminal-native interactive mode
_run_legacy_terminal_chat(
prompt=prompt,
model=model,
verbose=verbose,
memory=memory,
tools=tools,
user_id=user_id,
session_id=session_id,
continue_session=continue_session,
workspace=workspace,
no_acp=no_acp,
no_lsp=no_lsp,
)
# Use the async TUI (non-blocking, scrollable output)
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
tui_config = AsyncTUIConfig(
model=model or "gpt-4o-mini",
show_logo=not compact,
show_status_bar=not compact,
session_id=session_id,
workspace=workspace,
)
tui = AsyncTUI(config=tui_config)
if prompt:
# Single prompt mode - direct response, no streaming
response = tui.run_single(prompt)
if response:
print(response)
else:
# Interactive split-pane TUI mode
tui.run()
def _run_profiled_chat(
@@ -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,502 @@
"""
PraisonIO - Interactive CLI I/O handler inspired by Aider.
Key design principles:
1. prompt_toolkit used SYNCHRONOUSLY (never inside asyncio.run)
2. Rich for output rendering
3. MarkdownStream for flicker-free streaming
4. Clean separation of input/output concerns
"""
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional, Set
# Lazy imports for performance
_PROMPT_TOOLKIT_AVAILABLE = None
_RICH_AVAILABLE = None
def _check_prompt_toolkit():
global _PROMPT_TOOLKIT_AVAILABLE
if _PROMPT_TOOLKIT_AVAILABLE is None:
try:
import prompt_toolkit # noqa: F401
_PROMPT_TOOLKIT_AVAILABLE = True
except ImportError:
_PROMPT_TOOLKIT_AVAILABLE = False
return _PROMPT_TOOLKIT_AVAILABLE
def _check_rich():
global _RICH_AVAILABLE
if _RICH_AVAILABLE is None:
try:
import rich # noqa: F401
_RICH_AVAILABLE = True
except ImportError:
_RICH_AVAILABLE = False
return _RICH_AVAILABLE
@dataclass
class IOConfig:
"""Configuration for PraisonIO."""
# Appearance
pretty: bool = True
user_input_color: str = "#00cc00"
tool_output_color: str = "#0088ff"
tool_error_color: str = "#ff0000"
tool_warning_color: str = "#ffaa00"
assistant_color: str = "#00aaff"
# Input settings
multiline_mode: bool = False
vi_mode: bool = False
# History
input_history_file: Optional[str] = None
# Completion
enable_completions: bool = True
# Streaming
stream_delay: float = 0.05 # 20fps default
class PraisonCompleter:
"""Auto-completer for commands, files, and symbols."""
def __init__(self, commands: Optional[Dict[str, str]] = None):
self.commands = commands or {}
self.files: Set[str] = set()
self.symbols: Set[str] = set()
def add_files(self, files: List[str]) -> None:
"""Add files to completion."""
self.files.update(files)
def add_symbols(self, symbols: List[str]) -> None:
"""Add symbols to completion."""
self.symbols.update(symbols)
def get_completions(self, text: str, cursor_pos: int) -> List[tuple]:
"""Get completions for current input."""
before_cursor = text[:cursor_pos]
words = before_cursor.split()
current_word = words[-1] if words else ""
completions = []
# Slash commands
if current_word.startswith("/"):
cmd_part = current_word[1:].lower()
for cmd, desc in self.commands.items():
if cmd.lower().startswith(cmd_part):
completions.append((f"/{cmd}", desc))
# @ file mentions
elif current_word.startswith("@"):
file_part = current_word[1:].lower()
for f in self.files:
if file_part in f.lower():
completions.append((f"@{f}", "file"))
# General word completion
elif current_word:
word_lower = current_word.lower()
for symbol in self.symbols:
if symbol.lower().startswith(word_lower):
completions.append((symbol, "symbol"))
return completions[:20]
def create_prompt_toolkit_completer(self):
"""Create prompt_toolkit Completer."""
if not _check_prompt_toolkit():
return None
from prompt_toolkit.completion import Completer, Completion
outer = self
class PTCompleter(Completer):
def get_completions(self, document, complete_event):
text = document.text
cursor_pos = document.cursor_position
for completion, meta in outer.get_completions(text, cursor_pos):
words = text[:cursor_pos].split()
current_word = words[-1] if words else ""
start_pos = -len(current_word)
yield Completion(completion, start_position=start_pos, display_meta=meta)
return PTCompleter()
class MarkdownStream:
"""Flicker-free streaming markdown renderer.
Uses Rich Live with a sliding window approach:
- Stable lines print above the live area
- Unstable lines (last N) render in the live window
"""
def __init__(self, live_window: int = 6):
self.live_window = live_window
self.printed: List[str] = []
self.live = None
self._live_started = False
self.min_delay = 1.0 / 20 # 20fps
self.when = 0
def _render_to_lines(self, text: str) -> List[str]:
"""Render markdown to lines."""
if not _check_rich():
return text.splitlines(keepends=True)
import io
from rich.console import Console
from rich.markdown import Markdown
string_io = io.StringIO()
console = Console(file=string_io, force_terminal=True, width=100)
console.print(Markdown(text))
return string_io.getvalue().splitlines(keepends=True)
def update(self, text: str, final: bool = False) -> None:
"""Update the streaming display."""
if not _check_rich():
# Fallback: just print
print(text, end="" if not final else "\n", flush=True)
return
from rich.live import Live
from rich.text import Text
# Start live on first update
if not self._live_started:
self.live = Live(Text(""), refresh_per_second=20)
self.live.start()
self._live_started = True
# Throttle updates
now = time.time()
if not final and now - self.when < self.min_delay:
return
self.when = now
lines = self._render_to_lines(text)
num_lines = len(lines)
# Calculate stable lines
if not final:
num_lines -= self.live_window
if final or num_lines > 0:
num_printed = len(self.printed)
show_count = num_lines - num_printed
if show_count > 0:
show = lines[num_printed:num_lines]
show_text = "".join(show)
self.live.console.print(Text.from_ansi(show_text), end="")
self.printed = lines[:num_lines]
if final:
self.live.update(Text(""))
self.live.stop()
self.live = None
return
# Update live window
rest = lines[num_lines:]
rest_text = "".join(rest)
self.live.update(Text.from_ansi(rest_text))
def __del__(self):
if self.live:
try:
self.live.stop()
except Exception:
pass
class PraisonIO:
"""
Main I/O handler for PraisonAI interactive CLI.
Inspired by Aider's io.py - handles all terminal I/O with:
- Rich formatting for output
- prompt_toolkit for input (with completions, history)
- Streaming markdown rendering
"""
def __init__(self, config: Optional[IOConfig] = None):
self.config = config or IOConfig()
self.completer = PraisonCompleter()
self._console = None
self._prompt_session = None
self._history_file = None
# Setup history file
if self.config.input_history_file:
self._history_file = Path(self.config.input_history_file)
else:
self._history_file = Path.home() / ".praison" / "input_history"
self._history_file.parent.mkdir(parents=True, exist_ok=True)
@property
def console(self):
"""Lazy-load Rich console."""
if self._console is None and _check_rich():
from rich.console import Console
self._console = Console()
return self._console
def _create_prompt_session(self):
"""Create prompt_toolkit session."""
if not _check_prompt_toolkit():
return None
from prompt_toolkit import PromptSession
from prompt_toolkit.history import FileHistory
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.key_binding import KeyBindings
kb = KeyBindings()
# Ctrl+C to cancel current input
@kb.add("c-c")
def _(event):
event.current_buffer.reset()
# Alt+Enter for newline in multiline mode
@kb.add("escape", "enter")
def _(event):
event.current_buffer.insert_text("\n")
# Enter to submit (unless in multiline mode)
@kb.add("enter")
def _(event):
if self.config.multiline_mode:
event.current_buffer.insert_text("\n")
else:
event.current_buffer.validate_and_handle()
# Ctrl+D to submit in multiline mode
@kb.add("c-d")
def _(event):
event.current_buffer.validate_and_handle()
session_kwargs = {
"history": FileHistory(str(self._history_file)),
"auto_suggest": AutoSuggestFromHistory(),
"key_bindings": kb,
"multiline": self.config.multiline_mode,
}
if self.config.enable_completions:
completer = self.completer.create_prompt_toolkit_completer()
if completer:
from prompt_toolkit.completion import ThreadedCompleter
session_kwargs["completer"] = ThreadedCompleter(completer)
if self.config.vi_mode:
from prompt_toolkit.enums import EditingMode
session_kwargs["editing_mode"] = EditingMode.VI
return PromptSession(**session_kwargs)
def _get_style(self):
"""Get prompt_toolkit style."""
if not _check_prompt_toolkit() or not self.config.pretty:
return None
from prompt_toolkit.styles import Style
style_dict = {}
if self.config.user_input_color:
style_dict[""] = self.config.user_input_color
return Style.from_dict(style_dict) if style_dict else None
def get_input(self, prompt: str = "You: ") -> str:
"""
Get input from user.
This is SYNCHRONOUS - never call from inside asyncio.run().
"""
if self._prompt_session is None:
self._prompt_session = self._create_prompt_session()
if self._prompt_session:
try:
style = self._get_style()
return self._prompt_session.prompt(prompt, style=style)
except KeyboardInterrupt:
return ""
except EOFError:
return "/exit"
else:
# Fallback to basic input
try:
return input(prompt)
except KeyboardInterrupt:
return ""
except EOFError:
return "/exit"
def rule(self) -> None:
"""Print a horizontal rule."""
if self.console and self.config.pretty:
self.console.rule(style=self.config.user_input_color)
else:
print()
def print_welcome(self, model: str = "gpt-4o-mini", tools_count: int = 0, session_id: str = "") -> None:
"""Print welcome message."""
if self.console and self.config.pretty:
from rich.panel import Panel
from rich.text import Text
welcome = Text()
welcome.append("PraisonAI Interactive Mode\n", style="bold cyan")
welcome.append("Type your prompt, use ", style="dim")
welcome.append("/help", style="bold green")
welcome.append(" for commands, ", style="dim")
welcome.append("/exit", style="bold yellow")
welcome.append(" to quit\n", style="dim")
welcome.append(f"Model: {model}", style="dim cyan")
if tools_count:
welcome.append(f" | Tools: {tools_count}", style="dim")
if session_id:
welcome.append(f" | Session: {session_id[:8]}", style="dim")
self.console.print(Panel(welcome, border_style="blue", padding=(0, 1)))
else:
print("PraisonAI Interactive Mode")
print("Type your prompt, use /help for commands, /exit to quit")
print(f"Model: {model} | Tools: {tools_count}")
def print_user_input(self, text: str) -> None:
"""Echo user input with styling."""
if self.console and self.config.pretty:
from rich.text import Text
self.console.print(Text(text, style=self.config.user_input_color))
def print_assistant_start(self) -> None:
"""Print assistant response start marker."""
if self.console and self.config.pretty:
self.console.print("\n[bold cyan]Assistant:[/bold cyan]")
else:
print("\nAssistant:")
def print_assistant_response(self, response: str) -> None:
"""Print assistant response with markdown formatting."""
if self.console and self.config.pretty:
from rich.markdown import Markdown
from rich.panel import Panel
try:
md = Markdown(response)
self.console.print(Panel(md, border_style="cyan", padding=(0, 1)))
except Exception:
self.console.print(response)
else:
print(response)
def stream_response(self, text: str, final: bool = False) -> MarkdownStream:
"""Get or update streaming response."""
if not hasattr(self, '_current_stream') or self._current_stream is None:
self._current_stream = MarkdownStream()
self._current_stream.update(text, final=final)
if final:
self._current_stream = None
return self._current_stream
def tool_output(self, message: str) -> None:
"""Print tool output."""
if self.console and self.config.pretty:
self.console.print(f"[{self.config.tool_output_color}]⚙ {message}[/]")
else:
print(f"⚙ {message}")
def tool_error(self, message: str) -> None:
"""Print tool error."""
if self.console and self.config.pretty:
self.console.print(f"[{self.config.tool_error_color}]✗ {message}[/]")
else:
print(f"✗ {message}", file=sys.stderr)
def tool_warning(self, message: str) -> None:
"""Print tool warning."""
if self.console and self.config.pretty:
self.console.print(f"[{self.config.tool_warning_color}]⚠ {message}[/]")
else:
print(f"⚠ {message}", file=sys.stderr)
def info(self, message: str) -> None:
"""Print info message."""
if self.console and self.config.pretty:
self.console.print(f"[dim]ℹ {message}[/dim]")
else:
print(f"ℹ {message}")
def success(self, message: str) -> None:
"""Print success message."""
if self.console and self.config.pretty:
self.console.print(f"[green]✓ {message}[/green]")
else:
print(f"✓ {message}")
def print_help(self, commands: Dict[str, str]) -> None:
"""Print help for available commands."""
if self.console and self.config.pretty:
from rich.table import Table
table = Table(title="Available Commands", border_style="blue")
table.add_column("Command", style="green")
table.add_column("Description", style="dim")
for cmd, desc in sorted(commands.items()):
table.add_row(f"/{cmd}", desc)
self.console.print(table)
else:
print("\nAvailable Commands:")
for cmd, desc in sorted(commands.items()):
print(f" /{cmd:<12} - {desc}")
def confirm(self, message: str, default: bool = False) -> bool:
"""Ask for confirmation."""
suffix = " [Y/n]: " if default else " [y/N]: "
if self.console and self.config.pretty:
self.console.print(f"[yellow]? {message}[/yellow]", end="")
try:
response = input(suffix).strip().lower()
if not response:
return default
return response in ("y", "yes")
except (EOFError, KeyboardInterrupt):
return False
def add_commands(self, commands: Dict[str, str]) -> None:
"""Add commands for completion."""
self.completer.commands.update(commands)
def add_files(self, files: List[str]) -> None:
"""Add files for completion."""
self.completer.add_files(files)
def add_symbols(self, symbols: List[str]) -> None:
"""Add symbols for completion."""
self.completer.add_symbols(symbols)
@@ -0,0 +1,297 @@
"""
Interactive REPL for PraisonAI.
This is the main entry point for interactive chat mode.
Uses PraisonIO for input/output (prompt_toolkit is used SYNCHRONOUSLY).
"""
import logging
from dataclasses import dataclass
from typing import Dict, List, Optional
from .praison_io import PraisonIO, IOConfig
logger = logging.getLogger(__name__)
@dataclass
class REPLConfig:
"""Configuration for the interactive REPL."""
model: str = "gpt-4o-mini"
verbose: bool = False
memory: bool = False
tools: Optional[List] = None
session_id: Optional[str] = None
continue_session: bool = False
workspace: Optional[str] = None
autonomy: bool = True
# Default slash commands
DEFAULT_COMMANDS: Dict[str, str] = {
"help": "Show available commands",
"exit": "Exit interactive mode",
"quit": "Exit interactive mode",
"clear": "Clear conversation history",
"session": "Show current session info",
"model": "Show or change model",
"cost": "Show token usage and cost",
"history": "Show conversation history",
"export": "Export conversation to file",
"compact": "Toggle compact output mode",
"multiline": "Toggle multiline input mode",
}
class InteractiveREPL:
"""
Interactive REPL for PraisonAI chat.
Key design:
- Uses PraisonIO for all I/O (prompt_toolkit used synchronously)
- Agent execution is async but input is sync
- Clean separation between I/O and agent logic
"""
def __init__(
self,
config: Optional[REPLConfig] = None,
io_config: Optional[IOConfig] = None,
):
self.config = config or REPLConfig()
self.io = PraisonIO(io_config)
self._agent = None
self._running = False
self._conversation_history: List[Dict] = []
self._session_id: Optional[str] = None
self._total_tokens = 0
self._total_cost = 0.0
# Setup commands
self.io.add_commands(DEFAULT_COMMANDS)
def _get_agent(self):
"""Lazy-load the agent."""
if self._agent is None:
try:
from praisonaiagents import Agent
agent_kwargs = {
"name": "InteractiveAgent",
"role": "Assistant",
"goal": "Help the user with their requests",
"instructions": "You are a helpful AI assistant. Be concise and helpful.",
}
if self.config.model:
agent_kwargs["llm"] = self.config.model
if self.config.memory:
agent_kwargs["memory"] = True
if self.config.tools:
agent_kwargs["tools"] = self.config.tools
self._agent = Agent(**agent_kwargs)
except ImportError as e:
self.io.tool_error(f"Failed to import praisonaiagents: {e}")
raise
return self._agent
def _handle_command(self, command: str) -> bool:
"""
Handle slash commands.
Returns True if command was handled, False otherwise.
"""
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
self.io.info("Goodbye!")
return True
elif cmd == "help":
self.io.print_help(DEFAULT_COMMANDS)
return True
elif cmd == "clear":
self._conversation_history.clear()
self.io.success("Conversation cleared.")
return True
elif cmd == "session":
if self._session_id:
self.io.info(f"Session: {self._session_id}")
else:
self.io.info("No active session")
return True
elif cmd == "model":
if args:
self.config.model = args
self._agent = None # Reset agent to use new model
self.io.success(f"Model changed to: {args}")
else:
self.io.info(f"Current model: {self.config.model}")
return True
elif cmd == "cost":
self.io.info(f"Total tokens: {self._total_tokens}")
self.io.info(f"Estimated cost: ${self._total_cost:.4f}")
return True
elif cmd == "history":
if not self._conversation_history:
self.io.info("No conversation history")
else:
for i, msg in enumerate(self._conversation_history[-10:], 1):
role = msg.get("role", "unknown")
content = msg.get("content", "")[:100]
self.io.info(f"{i}. [{role}] {content}...")
return True
elif cmd == "export":
filename = args or "conversation.txt"
try:
with open(filename, "w") as f:
for msg in self._conversation_history:
f.write(f"[{msg.get('role', 'unknown')}]\n")
f.write(f"{msg.get('content', '')}\n\n")
self.io.success(f"Exported to {filename}")
except Exception as e:
self.io.tool_error(f"Export failed: {e}")
return True
elif cmd == "compact":
self.io.config.pretty = not self.io.config.pretty
mode = "disabled" if self.io.config.pretty else "enabled"
self.io.info(f"Compact mode {mode}")
return True
elif cmd == "multiline":
self.io.config.multiline_mode = not self.io.config.multiline_mode
mode = "enabled" if self.io.config.multiline_mode else "disabled"
self.io.info(f"Multiline mode {mode}")
return True
else:
self.io.tool_warning(f"Unknown command: /{cmd}")
self.io.info("Type /help for available commands")
return True
def _execute_prompt(self, prompt: str) -> Optional[str]:
"""Execute a prompt and return the response."""
try:
agent = self._get_agent()
# Add to history
self._conversation_history.append({
"role": "user",
"content": prompt
})
# Execute
response = agent.start(prompt)
# Add response to history
if response:
self._conversation_history.append({
"role": "assistant",
"content": str(response)
})
return str(response) if response else None
except Exception as e:
logger.exception("Error executing prompt")
self.io.tool_error(f"Error: {e}")
return None
def run(self) -> None:
"""
Run the interactive REPL.
This is the main entry point. It runs synchronously,
using prompt_toolkit for input (which must be sync).
"""
self._running = True
# Get tools count for display
tools_count = len(self.config.tools) if self.config.tools else 0
# Generate session ID
import uuid
self._session_id = self.config.session_id or str(uuid.uuid4())
# Print welcome
self.io.print_welcome(
model=self.config.model,
tools_count=tools_count,
session_id=self._session_id
)
while self._running:
try:
# Get input (SYNC - this is the key!)
user_input = self.io.get_input("You: ").strip()
if not user_input:
continue
# Handle slash commands
if user_input.startswith("/"):
self._handle_command(user_input)
continue
# Show thinking indicator
self.io.print_assistant_start()
# Execute prompt
response = self._execute_prompt(user_input)
# Display response
if response:
self.io.print_assistant_response(response)
except KeyboardInterrupt:
self.io.info("\nUse /exit to quit")
except EOFError:
self._running = False
self.io.info("Goodbye!")
except Exception as e:
logger.exception("Error in REPL loop")
self.io.tool_error(f"Error: {e}")
def run_single(self, prompt: str) -> Optional[str]:
"""Run a single prompt and return the response."""
return self._execute_prompt(prompt)
def start_interactive(
model: str = "gpt-4o-mini",
verbose: bool = False,
memory: bool = False,
tools: Optional[List] = None,
session_id: Optional[str] = None,
**kwargs
) -> None:
"""
Start the interactive REPL.
This is the main entry point for `praisonai chat` and `praisonai` (no args).
"""
config = REPLConfig(
model=model,
verbose=verbose,
memory=memory,
tools=tools,
session_id=session_id,
)
repl = InteractiveREPL(config=config)
repl.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()
@@ -0,0 +1,510 @@
"""
PraisonAI Interactive TUI Application.
Full-screen terminal UI with:
- Sticky bottom input box (like Aider/Codex/Claude Code)
- Scrollable output area above
- ASCII art logo banner
- Status bar with model/context info
- Clean, spacious design inspired by Claude Code
Uses prompt_toolkit for the full-screen application.
"""
import shutil
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime
# Lazy imports for performance
_PT_AVAILABLE = None
_RICH_AVAILABLE = None
def _check_prompt_toolkit():
global _PT_AVAILABLE
if _PT_AVAILABLE is None:
try:
import prompt_toolkit # noqa: F401
_PT_AVAILABLE = True
except ImportError:
_PT_AVAILABLE = False
return _PT_AVAILABLE
def _check_rich():
global _RICH_AVAILABLE
if _RICH_AVAILABLE is None:
try:
import rich # noqa: F401
_RICH_AVAILABLE = True
except ImportError:
_RICH_AVAILABLE = False
return _RICH_AVAILABLE
# ============================================================================
# ASCII Art Logo
# ============================================================================
# PraisonAI ASCII art logo - unique gradient style
LOGO_LARGE = r"""
██████╗ ██████╗ █████╗ ██╗███████╗ ██████╗ ███╗ ██╗ █████╗ ██╗
██╔══██╗██╔══██╗██╔══██╗██║██╔════╝██╔═══██╗████╗ ██║ ██╔══██╗██║
██████╔╝██████╔╝███████║██║███████╗██║ ██║██╔██╗ ██║ ███████║██║
██╔═══╝ ██╔══██╗██╔══██║██║╚════██║██║ ██║██║╚██╗██║ ██╔══██║██║
██║ ██║ ██║██║ ██║██║███████║╚██████╔╝██║ ╚████║ ██║ ██║██║
╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═╝ ╚═╝╚═╝
"""
LOGO_MEDIUM = r"""
████████╗ ████████╗ ██████╗ ██╗███████╗ ██████╗ ███╗ ██╗
██╔═══██║ ██╔═══██║ ██╔══██╗██║██╔════╝██╔═══██╗████╗ ██║
████████║ ████████║ ███████║██║███████╗██║ ██║██╔██╗ ██║
██╔═════╝ ██╔═══██║ ██╔══██║██║╚════██║██║ ██║██║╚██╗██║
██║ ██║ ██║ ██║ ██║██║███████║╚██████╔╝██║ ╚████║
╚═╝ ╚═╝ ╚═╝ ╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝ ╚═╝ ╚═══╝
"""
LOGO_SMALL = r"""
██████╗ ██████╗ █████╗ ██╗███████╗ ██████╗ ███╗ ██╗
██╔══██╗██╔══██╗██╔══██╗██║██╔════╝██╔═══██╗████╗ ██║
██████╔╝██████╔╝███████║██║███████╗██║ ██║██╔██╗ ██║
██╔═══╝ ██╔══██╗██╔══██║██║╚════██║██║ ██║██║╚██╗██║
██║ ██║ ██║██║ ██║██║███████║╚██████╔╝██║ ╚████║
╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝ ╚═╝ ╚═══╝
"""
LOGO_TINY = r"""
╔═╗┬─┐┌─┐┬┌─┐┌─┐┌┐┌ ╔═╗╦
╠═╝├┬┘├─┤│└─┐│ ││││ ╠═╣║
╩ ┴└─┴ ┴┴└─┘└─┘┘└┘ ╩ ╩╩
"""
LOGO_MINIMAL = "▶ PraisonAI"
def get_logo(width: int = 80) -> str:
"""Get appropriate logo based on terminal width."""
if width >= 80:
return LOGO_SMALL
elif width >= 60:
return LOGO_TINY
else:
return LOGO_MINIMAL
def get_version() -> str:
"""Get PraisonAI version."""
try:
from praisonai import __version__
return __version__
except Exception:
return "1.0.0"
# ============================================================================
# Color Scheme
# ============================================================================
@dataclass
class ColorScheme:
"""Color scheme for the TUI."""
# Primary colors
primary: str = "#00aaff" # Cyan-blue (PraisonAI brand)
secondary: str = "#888888" # Gray
accent: str = "#00cc00" # Green
# Text colors
text: str = "#ffffff" # White
text_dim: str = "#666666" # Dim gray
text_muted: str = "#444444" # Very dim
# Status colors
success: str = "#00cc00" # Green
warning: str = "#ffaa00" # Orange
error: str = "#ff4444" # Red
info: str = "#00aaff" # Blue
# UI elements
border: str = "#333333" # Dark gray
input_bg: str = "#1a1a1a" # Very dark
output_bg: str = "#0d0d0d" # Almost black
DEFAULT_COLORS = ColorScheme()
# ============================================================================
# TUI Configuration
# ============================================================================
@dataclass
class TUIConfig:
"""Configuration for the TUI application."""
# Model settings
model: str = "gpt-4o-mini"
# Display settings
show_logo: bool = True
show_tips: bool = True
show_status_bar: bool = True
compact_mode: bool = False
# Colors
colors: ColorScheme = field(default_factory=ColorScheme)
# Input settings
multiline: bool = False
vi_mode: bool = False
# History
history_file: Optional[str] = None
# Session
session_id: Optional[str] = None
workspace: Optional[str] = None
# ============================================================================
# Message Types
# ============================================================================
@dataclass
class Message:
"""A chat message."""
role: str # "user", "assistant", "system", "tool"
content: str
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict = field(default_factory=dict)
# ============================================================================
# TUI Application
# ============================================================================
class PraisonTUI:
"""
Full-screen TUI application for PraisonAI.
Features:
- Sticky bottom input box
- Scrollable output area
- ASCII art logo
- Status bar
- Clean, spacious design
"""
def __init__(self, config: Optional[TUIConfig] = None):
self.config = config or TUIConfig()
self.messages: List[Message] = []
self._running = False
self._agent = None
self._total_tokens = 0
self._total_cost = 0.0
# Get terminal size
self.term_width, self.term_height = shutil.get_terminal_size((80, 24))
# Generate session ID
import uuid
self.session_id = self.config.session_id or str(uuid.uuid4())[:8]
def _get_agent(self):
"""Lazy-load the agent."""
if self._agent is None:
try:
from praisonaiagents import Agent
self._agent = Agent(
name="PraisonAgent",
role="Assistant",
goal="Help the user with their requests",
instructions="You are a helpful AI 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 _render_logo(self) -> str:
"""Render the ASCII logo with colors."""
if not self.config.show_logo:
return ""
logo = get_logo(self.term_width)
version = get_version()
info_line = f"v{version} · Model: {self.config.model} · Session: {self.session_id}"
return logo.strip() + "\n" + info_line
def _render_welcome(self) -> str:
"""Render welcome message."""
lines = []
# Logo
if self.config.show_logo:
lines.append(self._render_logo())
lines.append("")
# Tips
if self.config.show_tips:
tips = [
"Type your message and press Enter to send",
"Use /help for commands, /exit to quit",
"Press ? for keyboard shortcuts",
]
lines.append(" " + tips[0])
lines.append("")
return "\n".join(lines)
def _render_status_bar(self) -> str:
"""Render the status bar."""
if not self.config.show_status_bar:
return ""
left = f"? for shortcuts"
center = f"{self.config.model}"
right = f"Session: {self.session_id}"
# Calculate spacing
total_len = len(left) + len(center) + len(right)
available = self.term_width - total_len - 4
if available > 0:
left_space = available // 2
right_space = available - left_space
return f" {left}{' ' * left_space}{center}{' ' * right_space}{right} "
else:
return f" {center} "
def _format_message(self, msg: Message) -> str:
"""Format a message for display."""
if msg.role == "user":
prefix = "› "
return f"{prefix}{msg.content}"
elif msg.role == "assistant":
prefix = "● "
return f"{prefix}{msg.content}"
elif msg.role == "tool":
prefix = "⚙ "
return f"{prefix}{msg.content}"
else:
return msg.content
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 interactive mode
/clear Clear conversation
/model [name] Show or change model
/cost Show token usage
/compact Toggle compact mode
/multiline Toggle multiline input
"""
self.messages.append(Message(role="system", content=help_text.strip()))
return True
elif cmd == "clear":
self.messages.clear()
return True
elif cmd == "model":
if args:
self.config.model = args
self._agent = None
self.messages.append(Message(role="system", content=f"Model changed to: {args}"))
else:
self.messages.append(Message(role="system", content=f"Current model: {self.config.model}"))
return True
elif cmd == "cost":
self.messages.append(Message(
role="system",
content=f"Tokens: {self._total_tokens} | Cost: ${self._total_cost:.4f}"
))
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(Message(role="system", content=f"Compact mode {mode}"))
return True
elif cmd == "multiline":
self.config.multiline = not self.config.multiline
mode = "enabled" if self.config.multiline else "disabled"
self.messages.append(Message(role="system", content=f"Multiline mode {mode}"))
return True
else:
self.messages.append(Message(role="system", content=f"Unknown command: /{cmd}"))
return True
def _execute_prompt(self, prompt: str) -> Optional[str]:
"""Execute a prompt and return the response."""
try:
agent = self._get_agent()
response = agent.start(prompt)
return str(response) if response else None
except Exception as e:
return f"Error: {e}"
def run(self) -> None:
"""Run the TUI application."""
self._running = True
# Print welcome
print(self._render_welcome())
# Setup history
history_file = self.config.history_file or str(Path.home() / ".praison" / "history")
Path(history_file).parent.mkdir(parents=True, exist_ok=True)
# Create prompt session
if _check_prompt_toolkit():
from prompt_toolkit import PromptSession
from prompt_toolkit.history import FileHistory
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.styles import Style
# Key bindings
kb = KeyBindings()
@kb.add("c-c")
def _(event):
event.current_buffer.reset()
@kb.add("c-d")
def _(event):
if not event.current_buffer.text:
event.app.exit()
# Style
style = Style.from_dict({
"": self.config.colors.accent,
"prompt": self.config.colors.primary,
})
session = PromptSession(
history=FileHistory(history_file),
auto_suggest=AutoSuggestFromHistory(),
key_bindings=kb,
style=style,
multiline=self.config.multiline,
)
def get_input():
return session.prompt(
[("class:prompt", "› ")],
placeholder="Type your message or @path/to/file"
)
else:
def get_input():
return input("› ")
# Main loop
while self._running:
try:
# Print status bar
if self.config.show_status_bar and not self.config.compact_mode:
status = self._render_status_bar()
if _check_rich():
from rich.console import Console
Console().print(f"[dim]{status}[/dim]")
else:
print(f"\033[90m{status}\033[0m")
# Get input
user_input = get_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(self._format_message(msg))
self.messages = [m for m in self.messages if m.role != "system"]
continue
# Add user message
self.messages.append(Message(role="user", content=user_input))
# Show thinking indicator
if _check_rich():
from rich.console import Console
Console().print("[dim]Thinking...[/dim]")
# Execute
response = self._execute_prompt(user_input)
if response:
self.messages.append(Message(role="assistant", content=response))
# Display response
if _check_rich():
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
console = Console()
try:
md = Markdown(response)
console.print(Panel(md, border_style="cyan", padding=(0, 1)))
except Exception:
console.print(f"● {response}")
else:
print(f"● {response}")
print() # Spacing
except KeyboardInterrupt:
print("\nUse /exit to quit")
except EOFError:
self._running = False
print("\nGoodbye!")
except Exception as e:
print(f"Error: {e}")
print("Goodbye!")
def run_single(self, prompt: str) -> Optional[str]:
"""Run a single prompt."""
return self._execute_prompt(prompt)
def start_tui(
model: str = "gpt-4o-mini",
show_logo: bool = True,
compact: bool = False,
**kwargs
) -> None:
"""Start the TUI application."""
config = TUIConfig(
model=model,
show_logo=show_logo,
compact_mode=compact,
)
tui = PraisonTUI(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.3 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')
@@ -358,6 +358,69 @@ class AIGenerator:
pass
return self._fast_context_handler
def _detect_feature_capabilities(self, slug_str: str, source_info: Dict[str, Any]) -> Dict[str, Any]:
"""Detect feature capabilities from source analysis."""
caps = {
"supports_bool": True,
"supports_list": False,
"supports_dict": False,
"supports_presets": [],
"supports_url": False,
"usage_forms": [],
"decorator": None,
"built_in_items": [],
}
# Feature-specific detection based on slug
if slug_str == "tools":
caps["supports_list"] = True
caps["supports_bool"] = False
caps["decorator"] = "@tool"
caps["built_in_items"] = ["duckduckgo", "wikipedia_tools", "read_file", "execute_command"]
caps["usage_forms"] = [
"- List of tools: `tools=[my_func, another_func]`",
"- Using @tool decorator to create custom tools",
"- Using built-in tools: duckduckgo, wikipedia_tools, etc.",
]
elif slug_str == "memory":
caps["supports_dict"] = True
caps["supports_presets"] = ["sqlite", "redis", "postgres", "qdrant", "chroma"]
caps["supports_url"] = True
caps["usage_forms"] = [
"- Boolean: `memory=True`",
"- Preset: `memory=\"sqlite\"` or `memory=\"redis\"`",
"- URL: `memory=\"redis://localhost:6379\"`",
"- Dict: `memory={\"provider\": \"qdrant\"}`",
]
elif slug_str == "knowledge":
caps["supports_dict"] = True
caps["supports_list"] = True
caps["usage_forms"] = [
"- Boolean: `knowledge=True`",
"- List of sources: `knowledge=[\"file.pdf\", \"doc.txt\"]`",
"- Dict config: `knowledge={\"sources\": [...]}`",
]
elif slug_str == "guardrails":
caps["supports_list"] = True
caps["decorator"] = "guardrail function"
caps["usage_forms"] = [
"- Boolean: `guardrails=True`",
"- List of functions: `guardrails=[my_guardrail_func]`",
"- Custom guardrail function returning GuardrailResult",
]
elif slug_str == "streaming":
caps["supports_bool"] = True
caps["usage_forms"] = [
"- Boolean: `output=\"stream\"` or iterate over agent.start()",
"- Streaming with callbacks",
]
else:
# Generic detection from source
caps["usage_forms"] = [f"- Boolean: `{slug_str}=True`"]
logger.debug(f"Feature '{slug_str}' capabilities: {caps}")
return caps
def _get_agent(self):
"""Lazy-load the generation agent."""
if self._agent is None:
@@ -591,14 +654,19 @@ ISSUES: List any issues found (or "None")
context = {}
slug_str = slug.normalised
logger.debug(f"Gathering context for feature '{slug_str}', artifact '{artifact_type.value}'")
# Use FastContext to find relevant code in SDK
fast_context = self._get_fast_context_handler()
if fast_context and self.sdk_root:
logger.debug(f"FastContext available, searching in {self.sdk_root}")
try:
# Search for feature-related code
query = f"Find {slug_str} implementation, usage examples, and API"
matches = fast_context.search_context(query, str(self.sdk_root), use_llm_keywords=False)
logger.debug(f"FastContext found {len(matches)} matches for '{query}'")
# Read matched files
for match in matches[:3]:
file_path = self.sdk_root / match.get('file', '')
@@ -606,10 +674,11 @@ ISSUES: List any issues found (or "None")
try:
content = file_path.read_text(encoding="utf-8")[:2000]
context[f"sdk_{file_path.stem}"] = content
logger.debug(f"Added SDK context: {file_path.name} ({len(content)} chars)")
except Exception:
pass
except Exception:
pass
except Exception as e:
logger.debug(f"FastContext search failed: {e}")
# Get SDK module info (fallback)
if self.sdk_root and not context:
@@ -653,6 +722,7 @@ ISSUES: List any issues found (or "None")
if doc_path.exists():
context[f"existing_{doc_type}"] = doc_path.read_text(encoding="utf-8")[:2000]
logger.debug(f"Context gathered for '{slug_str}': {list(context.keys())}")
return context
def _read_sdk_module(self, sdk_path: Path) -> str:
@@ -866,17 +936,26 @@ print(result)
"",
]
# Add usage forms guidance (common patterns for all features)
param_name = slug_str.replace("-", "_")
prompt_parts.extend([
"## USAGE FORMS TO DEMONSTRATE (MUST cover ALL these progressively):",
f"- Boolean: `{param_name}=True` - Enable with defaults",
f"- String preset: `{param_name}=\"sqlite\"` or `{param_name}=\"redis\"` - Use predefined backend",
f"- URL: `{param_name}=\"redis://localhost:6379\"` - Backend-specific connection",
f"- Dict config: `{param_name}={{\"backend\": \"postgres\", \"user_id\": \"u1\"}}` - Custom config",
f"- Array + overrides: `{param_name}=[\"redis\", {{\"user_id\": \"u1\"}}]` - Preset + customization",
"",
])
# Get dynamic feature capabilities - NO hardcoded usage forms
capabilities = self._detect_feature_capabilities(slug_str, source_info)
# Add ONLY the usage forms this feature actually supports
prompt_parts.append("## USAGE FORMS TO DEMONSTRATE (cover these progressively):")
for usage_form in capabilities["usage_forms"]:
prompt_parts.append(usage_form)
# Add decorator info if applicable
if capabilities.get("decorator"):
prompt_parts.append(f"- MUST demonstrate the {capabilities['decorator']} pattern")
# Add built-in items if applicable
if capabilities.get("built_in_items"):
items = ", ".join(capabilities["built_in_items"][:4])
prompt_parts.append(f"- Built-in options to show: {items}")
prompt_parts.append("")
logger.debug(f"Feature '{slug_str}' usage forms: {capabilities['usage_forms']}")
# Add dynamically extracted parameters if available
main_class = source_info.get("main_class")
+1 -1
View File
@@ -1 +1 @@
__version__ = "3.7.3"
__version__ = "3.7.7"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.11.2",
"praisonaiagents>=0.11.3",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
@@ -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,406 @@
"""
Tests for PraisonIO - Interactive CLI I/O handler.
"""
from unittest.mock import patch
class TestIOConfig:
"""Tests for IOConfig dataclass."""
def test_default_values(self):
"""Test default configuration values."""
from praisonai.cli.interactive.praison_io import IOConfig
config = IOConfig()
assert config.pretty is True
assert config.user_input_color == "#00cc00"
assert config.multiline_mode is False
assert config.vi_mode is False
assert config.enable_completions is True
def test_custom_values(self):
"""Test custom configuration values."""
from praisonai.cli.interactive.praison_io import IOConfig
config = IOConfig(
pretty=False,
user_input_color="#ff0000",
multiline_mode=True,
vi_mode=True,
)
assert config.pretty is False
assert config.user_input_color == "#ff0000"
assert config.multiline_mode is True
assert config.vi_mode is True
class TestPraisonCompleter:
"""Tests for PraisonCompleter."""
def test_init_empty(self):
"""Test initialization with no commands."""
from praisonai.cli.interactive.praison_io import PraisonCompleter
completer = PraisonCompleter()
assert completer.commands == {}
assert len(completer.files) == 0
assert len(completer.symbols) == 0
def test_init_with_commands(self):
"""Test initialization with commands."""
from praisonai.cli.interactive.praison_io import PraisonCompleter
commands = {"help": "Show help", "exit": "Exit"}
completer = PraisonCompleter(commands=commands)
assert completer.commands == commands
def test_add_files(self):
"""Test adding files."""
from praisonai.cli.interactive.praison_io import PraisonCompleter
completer = PraisonCompleter()
completer.add_files(["file1.py", "file2.py"])
assert "file1.py" in completer.files
assert "file2.py" in completer.files
def test_add_symbols(self):
"""Test adding symbols."""
from praisonai.cli.interactive.praison_io import PraisonCompleter
completer = PraisonCompleter()
completer.add_symbols(["MyClass", "my_function"])
assert "MyClass" in completer.symbols
assert "my_function" in completer.symbols
def test_get_completions_slash_commands(self):
"""Test completions for slash commands."""
from praisonai.cli.interactive.praison_io import PraisonCompleter
commands = {"help": "Show help", "history": "Show history", "exit": "Exit"}
completer = PraisonCompleter(commands=commands)
completions = completer.get_completions("/h", 2)
completion_texts = [c[0] for c in completions]
assert "/help" in completion_texts
assert "/history" in completion_texts
assert "/exit" not in completion_texts
def test_get_completions_file_mentions(self):
"""Test completions for @ file mentions."""
from praisonai.cli.interactive.praison_io import PraisonCompleter
completer = PraisonCompleter()
completer.add_files(["main.py", "test.py", "README.md"])
completions = completer.get_completions("@main", 5)
completion_texts = [c[0] for c in completions]
assert "@main.py" in completion_texts
def test_get_completions_symbols(self):
"""Test completions for symbols."""
from praisonai.cli.interactive.praison_io import PraisonCompleter
completer = PraisonCompleter()
completer.add_symbols(["MyClass", "MyFunction", "other"])
completions = completer.get_completions("My", 2)
completion_texts = [c[0] for c in completions]
assert "MyClass" in completion_texts
assert "MyFunction" in completion_texts
assert "other" not in completion_texts
class TestMarkdownStream:
"""Tests for MarkdownStream."""
def test_init(self):
"""Test initialization."""
from praisonai.cli.interactive.praison_io import MarkdownStream
stream = MarkdownStream(live_window=8)
assert stream.live_window == 8
assert stream.printed == []
assert stream.live is None
def test_render_to_lines_fallback(self):
"""Test rendering without Rich."""
from praisonai.cli.interactive.praison_io import MarkdownStream
stream = MarkdownStream()
# This tests the fallback path
text = "Hello\nWorld"
lines = stream._render_to_lines(text)
assert len(lines) >= 1
class TestPraisonIO:
"""Tests for PraisonIO."""
def test_init_default(self):
"""Test default initialization."""
from praisonai.cli.interactive.praison_io import PraisonIO
io = PraisonIO()
assert io.config is not None
assert io.completer is not None
def test_init_with_config(self):
"""Test initialization with config."""
from praisonai.cli.interactive.praison_io import PraisonIO, IOConfig
config = IOConfig(pretty=False)
io = PraisonIO(config=config)
assert io.config.pretty is False
def test_add_commands(self):
"""Test adding commands."""
from praisonai.cli.interactive.praison_io import PraisonIO
io = PraisonIO()
io.add_commands({"test": "Test command"})
assert "test" in io.completer.commands
def test_add_files(self):
"""Test adding files."""
from praisonai.cli.interactive.praison_io import PraisonIO
io = PraisonIO()
io.add_files(["file.py"])
assert "file.py" in io.completer.files
def test_add_symbols(self):
"""Test adding symbols."""
from praisonai.cli.interactive.praison_io import PraisonIO
io = PraisonIO()
io.add_symbols(["MyClass"])
assert "MyClass" in io.completer.symbols
@patch('builtins.print')
def test_rule_no_rich(self, mock_print):
"""Test rule without Rich."""
from praisonai.cli.interactive.praison_io import PraisonIO, IOConfig
config = IOConfig(pretty=False)
io = PraisonIO(config=config)
io.rule()
mock_print.assert_called()
@patch('builtins.print')
def test_tool_output_no_rich(self, mock_print):
"""Test tool output without Rich."""
from praisonai.cli.interactive.praison_io import PraisonIO, IOConfig
config = IOConfig(pretty=False)
io = PraisonIO(config=config)
io.tool_output("Test message")
mock_print.assert_called_with("⚙ Test message")
@patch('builtins.print')
def test_info_no_rich(self, mock_print):
"""Test info without Rich."""
from praisonai.cli.interactive.praison_io import PraisonIO, IOConfig
config = IOConfig(pretty=False)
io = PraisonIO(config=config)
io.info("Test info")
mock_print.assert_called_with("ℹ Test info")
@patch('builtins.print')
def test_success_no_rich(self, mock_print):
"""Test success without Rich."""
from praisonai.cli.interactive.praison_io import PraisonIO, IOConfig
config = IOConfig(pretty=False)
io = PraisonIO(config=config)
io.success("Test success")
mock_print.assert_called_with("✓ Test success")
class TestREPLConfig:
"""Tests for REPLConfig."""
def test_default_values(self):
"""Test default configuration values."""
from praisonai.cli.interactive.repl import REPLConfig
config = REPLConfig()
assert config.model == "gpt-4o-mini"
assert config.verbose is False
assert config.memory is False
assert config.tools is None
assert config.autonomy is True
def test_custom_values(self):
"""Test custom configuration values."""
from praisonai.cli.interactive.repl import REPLConfig
config = REPLConfig(
model="gpt-4o",
verbose=True,
memory=True,
)
assert config.model == "gpt-4o"
assert config.verbose is True
assert config.memory is True
class TestInteractiveREPL:
"""Tests for InteractiveREPL."""
def test_init_default(self):
"""Test default initialization."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
assert repl.config is not None
assert repl.io is not None
assert repl._running is False
def test_init_with_config(self):
"""Test initialization with config."""
from praisonai.cli.interactive.repl import InteractiveREPL, REPLConfig
config = REPLConfig(model="gpt-4o")
repl = InteractiveREPL(config=config)
assert repl.config.model == "gpt-4o"
def test_handle_command_exit(self):
"""Test handling exit command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
repl._running = True
result = repl._handle_command("/exit")
assert result is True
assert repl._running is False
def test_handle_command_quit(self):
"""Test handling quit command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
repl._running = True
result = repl._handle_command("/quit")
assert result is True
assert repl._running is False
def test_handle_command_help(self):
"""Test handling help command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
result = repl._handle_command("/help")
assert result is True
def test_handle_command_clear(self):
"""Test handling clear command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
repl._conversation_history = [{"role": "user", "content": "test"}]
result = repl._handle_command("/clear")
assert result is True
assert len(repl._conversation_history) == 0
def test_handle_command_model_show(self):
"""Test handling model command (show)."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
result = repl._handle_command("/model")
assert result is True
def test_handle_command_model_change(self):
"""Test handling model command (change)."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
result = repl._handle_command("/model gpt-4o")
assert result is True
assert repl.config.model == "gpt-4o"
def test_handle_command_unknown(self):
"""Test handling unknown command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
result = repl._handle_command("/unknown")
assert result is True # Still handled (shows warning)
def test_handle_command_session(self):
"""Test handling session command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
repl._session_id = "test-session-123"
result = repl._handle_command("/session")
assert result is True
def test_handle_command_cost(self):
"""Test handling cost command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
repl._total_tokens = 100
repl._total_cost = 0.01
result = repl._handle_command("/cost")
assert result is True
def test_handle_command_history_empty(self):
"""Test handling history command with empty history."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
result = repl._handle_command("/history")
assert result is True
def test_handle_command_history_with_items(self):
"""Test handling history command with items."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
repl._conversation_history = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
]
result = repl._handle_command("/history")
assert result is True
def test_handle_command_compact(self):
"""Test handling compact command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
original = repl.io.config.pretty
result = repl._handle_command("/compact")
assert result is True
assert repl.io.config.pretty != original
def test_handle_command_multiline(self):
"""Test handling multiline command."""
from praisonai.cli.interactive.repl import InteractiveREPL
repl = InteractiveREPL()
original = repl.io.config.multiline_mode
result = repl._handle_command("/multiline")
assert result is True
assert repl.io.config.multiline_mode != original
class TestStartInteractive:
"""Tests for start_interactive function."""
def test_function_exists(self):
"""Test that start_interactive function exists."""
from praisonai.cli.interactive.repl import start_interactive
assert callable(start_interactive)
@@ -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
@@ -0,0 +1,344 @@
"""
Tests for PraisonAI TUI Application.
"""
class TestColorScheme:
"""Tests for ColorScheme dataclass."""
def test_default_values(self):
"""Test default color scheme values."""
from praisonai.cli.interactive.tui_app import ColorScheme
colors = ColorScheme()
assert colors.primary == "#00aaff"
assert colors.secondary == "#888888"
assert colors.accent == "#00cc00"
assert colors.text == "#ffffff"
assert colors.success == "#00cc00"
assert colors.error == "#ff4444"
class TestTUIConfig:
"""Tests for TUIConfig dataclass."""
def test_default_values(self):
"""Test default TUI configuration values."""
from praisonai.cli.interactive.tui_app import TUIConfig
config = TUIConfig()
assert config.model == "gpt-4o-mini"
assert config.show_logo is True
assert config.show_tips is True
assert config.show_status_bar is True
assert config.compact_mode is False
assert config.multiline is False
assert config.vi_mode is False
def test_custom_values(self):
"""Test custom TUI configuration values."""
from praisonai.cli.interactive.tui_app import TUIConfig
config = TUIConfig(
model="gpt-4o",
show_logo=False,
compact_mode=True,
)
assert config.model == "gpt-4o"
assert config.show_logo is False
assert config.compact_mode is True
class TestMessage:
"""Tests for Message dataclass."""
def test_message_creation(self):
"""Test message creation."""
from praisonai.cli.interactive.tui_app import Message
msg = Message(role="user", content="Hello")
assert msg.role == "user"
assert msg.content == "Hello"
assert msg.timestamp is not None
assert msg.metadata == {}
def test_message_with_metadata(self):
"""Test message with metadata."""
from praisonai.cli.interactive.tui_app import Message
msg = Message(role="assistant", content="Hi", metadata={"tokens": 10})
assert msg.metadata == {"tokens": 10}
class TestGetLogo:
"""Tests for get_logo function."""
def test_logo_wide_terminal(self):
"""Test logo for wide terminal."""
from praisonai.cli.interactive.tui_app import get_logo, LOGO_SMALL
logo = get_logo(100)
assert logo == LOGO_SMALL
def test_logo_medium_terminal(self):
"""Test logo for medium terminal."""
from praisonai.cli.interactive.tui_app import get_logo, LOGO_TINY
logo = get_logo(65)
assert logo == LOGO_TINY
def test_logo_narrow_terminal(self):
"""Test logo for narrow terminal."""
from praisonai.cli.interactive.tui_app import get_logo, LOGO_MINIMAL
logo = get_logo(50)
assert logo == LOGO_MINIMAL
class TestPraisonTUI:
"""Tests for PraisonTUI class."""
def test_init_default(self):
"""Test default initialization."""
from praisonai.cli.interactive.tui_app import PraisonTUI
tui = PraisonTUI()
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.tui_app import PraisonTUI, TUIConfig
config = TUIConfig(model="gpt-4o", show_logo=False)
tui = PraisonTUI(config=config)
assert tui.config.model == "gpt-4o"
assert tui.config.show_logo is False
def test_render_logo_disabled(self):
"""Test render_logo when disabled."""
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
config = TUIConfig(show_logo=False)
tui = PraisonTUI(config=config)
logo = tui._render_logo()
assert logo == ""
def test_render_logo_enabled(self):
"""Test render_logo when enabled."""
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
config = TUIConfig(show_logo=True, model="gpt-4o-mini")
tui = PraisonTUI(config=config)
logo = tui._render_logo()
assert "PRAISON" in logo or "gpt-4o-mini" in logo
def test_render_welcome(self):
"""Test render_welcome."""
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
config = TUIConfig(show_logo=True, show_tips=True)
tui = PraisonTUI(config=config)
welcome = tui._render_welcome()
assert "Type your message" in welcome or len(welcome) > 0
def test_render_status_bar(self):
"""Test render_status_bar."""
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
config = TUIConfig(show_status_bar=True, model="gpt-4o")
tui = PraisonTUI(config=config)
status = tui._render_status_bar()
assert "gpt-4o" in status
def test_render_status_bar_disabled(self):
"""Test render_status_bar when disabled."""
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
config = TUIConfig(show_status_bar=False)
tui = PraisonTUI(config=config)
status = tui._render_status_bar()
assert status == ""
def test_format_message_user(self):
"""Test format_message for user role."""
from praisonai.cli.interactive.tui_app import PraisonTUI, Message
tui = PraisonTUI()
msg = Message(role="user", content="Hello")
formatted = tui._format_message(msg)
assert "Hello" in formatted
assert "›" in formatted
def test_format_message_assistant(self):
"""Test format_message for assistant role."""
from praisonai.cli.interactive.tui_app import PraisonTUI, Message
tui = PraisonTUI()
msg = Message(role="assistant", content="Hi there")
formatted = tui._format_message(msg)
assert "Hi there" in formatted
assert "●" in formatted
def test_format_message_tool(self):
"""Test format_message for tool role."""
from praisonai.cli.interactive.tui_app import PraisonTUI, Message
tui = PraisonTUI()
msg = Message(role="tool", content="Running command")
formatted = tui._format_message(msg)
assert "Running command" in formatted
assert "⚙" in formatted
def test_handle_command_exit(self):
"""Test handling exit command."""
from praisonai.cli.interactive.tui_app import PraisonTUI
tui = PraisonTUI()
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.tui_app import PraisonTUI
tui = PraisonTUI()
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.tui_app import PraisonTUI
tui = PraisonTUI()
result = tui._handle_command("/help")
assert result is True
# Help should add a system message
assert len(tui.messages) == 1
assert tui.messages[0].role == "system"
def test_handle_command_clear(self):
"""Test handling clear command."""
from praisonai.cli.interactive.tui_app import PraisonTUI, Message
tui = PraisonTUI()
tui.messages = [Message(role="user", content="test")]
result = tui._handle_command("/clear")
assert result is True
assert len(tui.messages) == 0
def test_handle_command_model_show(self):
"""Test handling model command (show)."""
from praisonai.cli.interactive.tui_app import PraisonTUI
tui = PraisonTUI()
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.tui_app import PraisonTUI
tui = PraisonTUI()
result = tui._handle_command("/model gpt-4o")
assert result is True
assert tui.config.model == "gpt-4o"
def test_handle_command_cost(self):
"""Test handling cost command."""
from praisonai.cli.interactive.tui_app import PraisonTUI
tui = PraisonTUI()
tui._total_tokens = 100
tui._total_cost = 0.01
result = tui._handle_command("/cost")
assert result is True
assert len(tui.messages) == 1
assert "100" in tui.messages[0].content
def test_handle_command_compact(self):
"""Test handling compact command."""
from praisonai.cli.interactive.tui_app import PraisonTUI
tui = PraisonTUI()
original = tui.config.compact_mode
result = tui._handle_command("/compact")
assert result is True
assert tui.config.compact_mode != original
def test_handle_command_multiline(self):
"""Test handling multiline command."""
from praisonai.cli.interactive.tui_app import PraisonTUI
tui = PraisonTUI()
original = tui.config.multiline
result = tui._handle_command("/multiline")
assert result is True
assert tui.config.multiline != original
def test_handle_command_unknown(self):
"""Test handling unknown command."""
from praisonai.cli.interactive.tui_app import PraisonTUI
tui = PraisonTUI()
result = tui._handle_command("/unknown")
assert result is True
assert len(tui.messages) == 1
assert "Unknown" in tui.messages[0].content
class TestStartTUI:
"""Tests for start_tui function."""
def test_function_exists(self):
"""Test that start_tui function exists."""
from praisonai.cli.interactive.tui_app import start_tui
assert callable(start_tui)
class TestASCIILogos:
"""Tests for ASCII logo constants."""
def test_logo_large_exists(self):
"""Test LOGO_LARGE constant exists."""
from praisonai.cli.interactive.tui_app import LOGO_LARGE
assert len(LOGO_LARGE) > 0
assert "PRAISON" in LOGO_LARGE or "██" in LOGO_LARGE
def test_logo_small_exists(self):
"""Test LOGO_SMALL constant exists."""
from praisonai.cli.interactive.tui_app import LOGO_SMALL
assert len(LOGO_SMALL) > 0
def test_logo_tiny_exists(self):
"""Test LOGO_TINY constant exists."""
from praisonai.cli.interactive.tui_app import LOGO_TINY
assert len(LOGO_TINY) > 0
def test_logo_minimal_exists(self):
"""Test LOGO_MINIMAL constant exists."""
from praisonai.cli.interactive.tui_app import LOGO_MINIMAL
assert "PraisonAI" in LOGO_MINIMAL
+4 -4
View File
@@ -4276,7 +4276,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.1.0" },
{ name = "praisonaiagents", specifier = ">=0.11.2" },
{ name = "praisonaiagents", specifier = ">=0.11.3" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4324,7 +4324,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.11.2"
version = "0.11.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4333,9 +4333,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/a4/d0/84583da326312957eb887e32d6df4549889494b736b2f3925f3b80fc7e70/praisonaiagents-0.11.2.tar.gz", hash = "sha256:a32e1063404d61cac71be4ee447f6888f7b9566a298ac63793aeaec41eeeeab1", size = 972927, upload-time = "2026-01-10T19:35:37.032Z" }
sdist = { url = "https://files.pythonhosted.org/packages/96/4a/ab73d7381b0e9139f722dad51f9cb63575032c3d11aa83bc281f76bfd13a/praisonaiagents-0.11.3.tar.gz", hash = "sha256:5aa024935184eaa66578ff4c888e06b052d55568c68bfb702dd35c8a7951e10b", size = 973113, upload-time = "2026-01-11T07:29:29.515Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/43/e8/3c13e6f5a3bcd4b01c0a373a2d3fdd797eeea3f46f673fc6a14ba081e1f2/praisonaiagents-0.11.2-py3-none-any.whl", hash = "sha256:e5d666d23d59e53b498beefc30edef3d28ca42fda7b0cd862d1859e59410067c", size = 954207, upload-time = "2026-01-10T19:35:35.034Z" },
{ url = "https://files.pythonhosted.org/packages/a2/05/f142e42a88cd61d0af4391be3e97c5ecfe660deeed96ebcdc31f95180977/praisonaiagents-0.11.3-py3-none-any.whl", hash = "sha256:a78dcbaa3eb66cfab52a4c9789654c58eb71b0ea34b2f0dcca54baff91917e7d", size = 954407, upload-time = "2026-01-11T07:29:26.456Z" },
]
[[package]]