Compare commits

...
3 Commits
Author SHA1 Message Date
Mervin Praison 2c4dd0e2d4 Release v3.7.8 2026-01-12 07:39:39 +00:00
Mervin Praison fd23d62841 Release v3.7.7 2026-01-11 21:30:04 +00:00
Mervin Praison 5f3bcb78ac Release v3.7.6 2026-01-11 08:23:08 +00:00
37 changed files with 4285 additions and 131 deletions

No files matched your search

+2 -1
View File
@@ -83,4 +83,5 @@ issue_details.json
# MCP Registry tokens
.mcpregistry_*
*.log
*.log
.internal
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.5" \
"praisonai>=3.7.8" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.5" \
"praisonai>=3.7.8" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.7.5" \
"praisonai>=3.7.8" \
"praisonai[ui]" \
"praisonai[crewai]"
@@ -447,14 +447,27 @@ class Agent:
# ─────────────────────────────────────────────────────────────────────
# Resolve OUTPUT param using unified resolver
# Supports: None, str preset, list [preset, overrides], OutputConfig
# DEFAULT: "actions" mode (minimal overhead, multi-agent safe)
# Can be overridden by PRAISONAI_OUTPUT env var
# ─────────────────────────────────────────────────────────────────────
from ..config.presets import DEFAULT_OUTPUT_MODE
# Check for env var override
env_output = os.environ.get('PRAISONAI_OUTPUT', '').lower()
if env_output and env_output in OUTPUT_PRESETS and output is None:
output = env_output
# Use default output mode if not specified
if output is None:
output = DEFAULT_OUTPUT_MODE
_output_config = resolve(
value=output,
param_name="output",
config_class=OutputConfig,
presets=OUTPUT_PRESETS,
array_mode=ArrayMode.PRESET_OVERRIDE,
default=OutputConfig(),
default=OutputConfig(), # OutputConfig defaults to actions mode
)
if _output_config:
verbose = _output_config.verbose
@@ -463,8 +476,24 @@ class Agent:
metrics = _output_config.metrics
reasoning_steps = _output_config.reasoning_steps
output_style = getattr(_output_config, 'style', None)
actions_trace = getattr(_output_config, 'actions_trace', True) # Default True
json_output = getattr(_output_config, 'json_output', False)
else:
verbose, markdown, stream, metrics, reasoning_steps = True, True, False, False, False
# Fallback defaults match actions mode
verbose, markdown, stream, metrics, reasoning_steps = False, False, False, False, False
actions_trace = True
json_output = False
# Enable actions output mode if configured
# This registers callbacks to capture tool calls and final output
if actions_trace:
try:
from ..output.actions import enable_actions_mode, is_actions_mode_enabled
if not is_actions_mode_enabled():
output_format = "jsonl" if json_output else "text"
enable_actions_mode(redact=True, use_color=True, format=output_format)
except ImportError:
pass # Actions module not available
# ─────────────────────────────────────────────────────────────────────
# Resolve EXECUTION param using unified resolver
@@ -1065,6 +1094,9 @@ Your Goal: {self.goal}
self._context_param = context # Store raw param for lazy init
self._context_manager = None # Lazy initialized on first use
self._context_manager_initialized = False
# Action trace mode - handled via display callbacks, not separate emitter
self._actions_trace = actions_trace
@property
def auto_memory(self):
@@ -391,10 +391,21 @@ class OutputConfig:
Consolidates: verbose, markdown, stream, metrics, reasoning_steps, output_style
DEFAULT: output="actions" (minimal overhead, multi-agent safe)
Usage:
# Simple preset
# Default is actions mode (tool calls + final output only)
Agent(instructions="...") # Uses output="actions"
# Verbose mode with Rich streaming
Agent(output="verbose")
# Plain mode (final output only, no action trace)
Agent(output="plain")
# JSON mode for piping
Agent(output="json")
# With config
Agent(output=OutputConfig(
verbose=True,
@@ -404,11 +415,11 @@ class OutputConfig:
reasoning_steps=True,
))
"""
# Verbosity
verbose: bool = True
# Verbosity - False by default for actions mode
verbose: bool = False
# Formatting
markdown: bool = True
# Formatting - False by default for actions mode
markdown: bool = False
# Streaming
stream: bool = False
@@ -422,6 +433,14 @@ class OutputConfig:
# Output style (custom styling)
style: Optional[Any] = None
# Actions trace mode - shows tool calls, agent lifecycle, final output
# When True, suppresses verbose output and shows action-level trace instead
# DEFAULT: True for minimal overhead
actions_trace: bool = True
# JSON output mode - emit JSONL events for piping
json_output: bool = False
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
return {
@@ -430,6 +449,8 @@ class OutputConfig:
"stream": self.stream,
"metrics": self.metrics,
"reasoning_steps": self.reasoning_steps,
"actions_trace": self.actions_trace,
"json_output": self.json_output,
}
@@ -41,6 +41,27 @@ MEMORY_URL_SCHEMES: Dict[str, str] = {
# =============================================================================
OUTPUT_PRESETS: Dict[str, Dict[str, Any]] = {
# Actions preset - DEFAULT for SDK
# Shows action trace (tool calls, agent lifecycle) + final output
# No verbose panels, no streaming tokens, just action-level trace
# Minimal overhead, multi-agent safe
"actions": {
"verbose": False,
"markdown": False,
"stream": False,
"metrics": False,
"reasoning_steps": False,
"actions_trace": True, # Special flag for action trace mode
},
# Plain preset - final output only, no action trace
"plain": {
"verbose": False,
"markdown": False,
"stream": False,
"metrics": False,
"reasoning_steps": False,
"actions_trace": False,
},
"minimal": {
"verbose": False,
"markdown": False,
@@ -84,8 +105,21 @@ OUTPUT_PRESETS: Dict[str, Dict[str, Any]] = {
"metrics": False,
"reasoning_steps": False,
},
# JSON preset - JSONL output for piping
"json": {
"verbose": False,
"markdown": False,
"stream": False,
"metrics": False,
"reasoning_steps": False,
"actions_trace": True,
"json_output": True, # Output as JSONL
},
}
# Default output mode - can be overridden by PRAISONAI_OUTPUT env var
DEFAULT_OUTPUT_MODE = "actions"
# =============================================================================
# Execution Presets
@@ -0,0 +1,68 @@
"""
Trace Module for PraisonAI Agents.
Provides lightweight action tracing for the `output="actions"` mode.
Shows agent lifecycle events, tool calls, and final output without
verbose internal details.
Zero Performance Impact:
- All imports are lazy loaded via __getattr__
- NoOpSink is the default (zero overhead when not used)
- Disabled emitter has near-zero overhead
Usage:
from praisonaiagents import Agent
# Simple usage - shows action trace
agent = Agent(instructions="...", output="actions")
agent.start("Do something")
# Advanced - capture to file
from praisonaiagents.trace import ActionTraceConfig
agent = Agent(
instructions="...",
output="actions",
trace=ActionTraceConfig(
sink_type="jsonl",
file_path="trace.jsonl",
)
)
"""
__all__ = [
# Core types
"ActionEvent",
"ActionEventType",
"ActionTraceConfig",
# Sinks
"TraceSink",
"NoOpSink",
"ListSink",
# Emitter
"TraceEmitter",
# Redaction
"redact_dict",
"REDACT_KEYS",
]
def __getattr__(name: str):
"""Lazy load module components to avoid import overhead."""
if name in ("ActionEvent", "ActionEventType", "ActionTraceConfig"):
from .protocol import ActionEvent, ActionEventType, ActionTraceConfig
return locals()[name]
if name in ("TraceSink", "NoOpSink", "ListSink"):
from .protocol import TraceSink, NoOpSink, ListSink
return locals()[name]
if name == "TraceEmitter":
from .protocol import TraceEmitter
return TraceEmitter
if name in ("redact_dict", "REDACT_KEYS"):
from .redact import redact_dict, REDACT_KEYS
return locals()[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -0,0 +1,398 @@
"""
Action Trace Protocol for PraisonAI Agents.
Lightweight protocol for tracing agent actions. Uses only stdlib
to ensure zero dependencies and minimal overhead.
Schema Version: 1.0
"""
import json
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
# Schema version for forward compatibility
SCHEMA_VERSION = "1.0"
class ActionEventType(str, Enum):
"""Types of action events."""
AGENT_START = "agent_start"
AGENT_END = "agent_end"
TOOL_START = "tool_start"
TOOL_END = "tool_end"
ERROR = "error"
OUTPUT = "output"
@dataclass
class ActionEvent:
"""
Lightweight action trace event.
Represents a single action in the agent execution trace.
Designed for minimal memory footprint and fast serialization.
"""
event_type: str
timestamp: float
schema_version: str = SCHEMA_VERSION
agent_id: Optional[str] = None
agent_name: Optional[str] = None
tool_name: Optional[str] = None
tool_args: Optional[Dict[str, Any]] = None
tool_result_summary: Optional[str] = None
duration_ms: Optional[float] = None
status: Optional[str] = None # ok, error
error_message: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary, excluding None values."""
result = {
"schema_version": self.schema_version,
"event_type": self.event_type,
"timestamp": self.timestamp,
}
# Add optional fields only if set
if self.agent_id is not None:
result["agent_id"] = self.agent_id
if self.agent_name is not None:
result["agent_name"] = self.agent_name
if self.tool_name is not None:
result["tool_name"] = self.tool_name
if self.tool_args is not None:
result["tool_args"] = self.tool_args
if self.tool_result_summary is not None:
result["tool_result_summary"] = self.tool_result_summary
if self.duration_ms is not None:
result["duration_ms"] = self.duration_ms
if self.status is not None:
result["status"] = self.status
if self.error_message is not None:
result["error_message"] = self.error_message
if self.metadata:
result["metadata"] = self.metadata
return result
def to_json(self) -> str:
"""Convert to JSON string."""
return json.dumps(self.to_dict(), default=str)
@runtime_checkable
class TraceSink(Protocol):
"""
Protocol for trace event sinks.
Implementations receive action events and handle them
(e.g., print to stdout, write to file, send to server).
"""
def emit(self, event: ActionEvent) -> None:
"""Emit an action event."""
...
def flush(self) -> None:
"""Flush any buffered events."""
...
def close(self) -> None:
"""Close the sink and release resources."""
...
class NoOpSink:
"""
No-operation sink that discards all events.
Used as the default sink when tracing is disabled.
Has near-zero overhead.
"""
__slots__ = ()
def emit(self, event: ActionEvent) -> None:
"""Discard the event."""
pass
def flush(self) -> None:
"""No-op."""
pass
def close(self) -> None:
"""No-op."""
pass
class ListSink:
"""
Sink that collects events in a list.
Useful for testing and programmatic access to trace events.
"""
__slots__ = ("_events",)
def __init__(self):
self._events: List[ActionEvent] = []
def emit(self, event: ActionEvent) -> None:
"""Add event to the list."""
self._events.append(event)
def flush(self) -> None:
"""No-op for list sink."""
pass
def close(self) -> None:
"""No-op for list sink."""
pass
def get_events(self) -> List[ActionEvent]:
"""Get all collected events."""
return self._events.copy()
def clear(self) -> None:
"""Clear all collected events."""
self._events.clear()
@dataclass
class ActionTraceConfig:
"""
Configuration for action tracing.
Usage:
# Default (disabled)
config = ActionTraceConfig()
# Enable with stdout
config = ActionTraceConfig(enabled=True, sink_type="stdout")
# Enable with JSONL file
config = ActionTraceConfig(
enabled=True,
sink_type="jsonl",
file_path="trace.jsonl",
)
"""
enabled: bool = True
redact: bool = True
compact: bool = False
sink_type: str = "noop" # noop, stdout, jsonl, file
file_path: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary."""
return {
"enabled": self.enabled,
"redact": self.redact,
"compact": self.compact,
"sink_type": self.sink_type,
"file_path": self.file_path,
}
class TraceEmitter:
"""
Emitter for action trace events.
Provides a high-level API for emitting trace events.
Handles redaction and sink management.
Usage:
sink = ListSink()
emitter = TraceEmitter(sink=sink)
emitter.agent_start("researcher")
emitter.tool_start("search", {"query": "AI"})
emitter.tool_end("search", duration_ms=100, status="ok")
emitter.agent_end("researcher", duration_ms=500)
"""
__slots__ = ("_sink", "_enabled", "_redact", "_agent_starts")
def __init__(
self,
sink: Optional[TraceSink] = None,
enabled: bool = True,
redact: bool = True,
):
self._sink = sink or NoOpSink()
self._enabled = enabled
self._redact = redact
self._agent_starts: Dict[str, float] = {} # Track agent start times
def _emit(self, event: ActionEvent) -> None:
"""Internal emit with enabled check."""
if not self._enabled:
return
self._sink.emit(event)
def _redact_args(self, args: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""Redact sensitive data from args if enabled."""
if args is None or not self._redact:
return args
from .redact import redact_dict
return redact_dict(args)
def agent_start(
self,
agent_name: str,
agent_id: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""Emit agent start event."""
ts = time.time()
self._agent_starts[agent_name] = ts
self._emit(ActionEvent(
event_type=ActionEventType.AGENT_START.value,
timestamp=ts,
agent_name=agent_name,
agent_id=agent_id,
metadata=metadata or {},
))
def agent_end(
self,
agent_name: str,
agent_id: Optional[str] = None,
duration_ms: Optional[float] = None,
status: str = "ok",
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""Emit agent end event."""
ts = time.time()
# Calculate duration if not provided
if duration_ms is None and agent_name in self._agent_starts:
start_ts = self._agent_starts.pop(agent_name, None)
if start_ts:
duration_ms = (ts - start_ts) * 1000
self._emit(ActionEvent(
event_type=ActionEventType.AGENT_END.value,
timestamp=ts,
agent_name=agent_name,
agent_id=agent_id,
duration_ms=duration_ms,
status=status,
metadata=metadata or {},
))
def tool_start(
self,
tool_name: str,
tool_args: Optional[Dict[str, Any]] = None,
agent_name: Optional[str] = None,
agent_id: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""Emit tool start event."""
self._emit(ActionEvent(
event_type=ActionEventType.TOOL_START.value,
timestamp=time.time(),
tool_name=tool_name,
tool_args=self._redact_args(tool_args),
agent_name=agent_name,
agent_id=agent_id,
metadata=metadata or {},
))
def tool_end(
self,
tool_name: str,
duration_ms: float,
status: str = "ok",
result_summary: Optional[str] = None,
agent_name: Optional[str] = None,
agent_id: Optional[str] = None,
error_message: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""Emit tool end event."""
self._emit(ActionEvent(
event_type=ActionEventType.TOOL_END.value,
timestamp=time.time(),
tool_name=tool_name,
duration_ms=duration_ms,
status=status,
tool_result_summary=result_summary,
agent_name=agent_name,
agent_id=agent_id,
error_message=error_message,
metadata=metadata or {},
))
def error(
self,
message: str,
tool_name: Optional[str] = None,
agent_name: Optional[str] = None,
agent_id: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""Emit error event."""
self._emit(ActionEvent(
event_type=ActionEventType.ERROR.value,
timestamp=time.time(),
error_message=message,
tool_name=tool_name,
agent_name=agent_name,
agent_id=agent_id,
status="error",
metadata=metadata or {},
))
def output(
self,
content: str,
agent_name: Optional[str] = None,
agent_id: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> None:
"""Emit final output event."""
# Truncate long output for summary
summary = content[:500] + "..." if len(content) > 500 else content
self._emit(ActionEvent(
event_type=ActionEventType.OUTPUT.value,
timestamp=time.time(),
tool_result_summary=summary,
agent_name=agent_name,
agent_id=agent_id,
metadata=metadata or {},
))
def flush(self) -> None:
"""Flush the sink."""
self._sink.flush()
def close(self) -> None:
"""Close the sink."""
self._sink.close()
# Global default emitter (NoOp by default)
_default_emitter: Optional[TraceEmitter] = None
def get_default_emitter() -> TraceEmitter:
"""Get the default trace emitter."""
global _default_emitter
if _default_emitter is None:
_default_emitter = TraceEmitter(sink=NoOpSink(), enabled=False)
return _default_emitter
def set_default_emitter(emitter: TraceEmitter) -> None:
"""Set the default trace emitter."""
global _default_emitter
_default_emitter = emitter
@@ -0,0 +1,149 @@
"""
Redaction utilities for action trace.
Provides deterministic redaction of sensitive data in tool arguments
and results. Uses only stdlib for zero dependencies.
"""
from typing import Any, Dict, Set
import re
# Keys that should always be redacted (case-insensitive matching)
REDACT_KEYS: Set[str] = {
# API keys and tokens
"api_key",
"apikey",
"api-key",
"secret",
"secret_key",
"secretkey",
"private_key",
"privatekey",
# Authentication
"password",
"passwd",
"pwd",
"token",
"bearer",
"authorization",
"auth",
"auth_token",
"access_token",
"refresh_token",
# Session and cookies
"cookie",
"session",
"session_id",
"sessionid",
"credential",
"credentials",
# Database
"connection_string",
"connectionstring",
"database_url",
"db_url",
"dburl",
"db_password",
"db_pass",
# Cloud provider specific
"aws_secret",
"aws_secret_access_key",
"azure_key",
"gcp_key",
"openai_api_key",
"anthropic_api_key",
}
# Compiled regex for case-insensitive matching
_REDACT_PATTERN = re.compile(
r"(" + "|".join(re.escape(k) for k in REDACT_KEYS) + r")",
re.IGNORECASE
)
REDACTED_VALUE = "[REDACTED]"
def _should_redact(key: str) -> bool:
"""Check if a key should be redacted."""
# Normalize key for comparison
normalized = key.lower().replace("-", "_").replace(" ", "_")
# Check exact match first
if normalized in REDACT_KEYS:
return True
# Check if any redact key is contained in the normalized key
for redact_key in REDACT_KEYS:
if redact_key in normalized:
return True
return False
def redact_dict(data: Dict[str, Any], enabled: bool = True) -> Dict[str, Any]:
"""
Redact sensitive values from a dictionary.
Args:
data: Dictionary to redact
enabled: If False, returns data unchanged
Returns:
New dictionary with sensitive values redacted
"""
if not enabled:
return data
if not data:
return {}
return _redact_value(data)
def _redact_value(value: Any) -> Any:
"""Recursively redact sensitive values."""
if isinstance(value, dict):
return {k: _redact_key_value(k, v) for k, v in value.items()}
elif isinstance(value, list):
return [_redact_value(item) for item in value]
else:
return value
def _redact_key_value(key: str, value: Any) -> Any:
"""Redact a value if its key is sensitive."""
if _should_redact(key):
return REDACTED_VALUE
# Recursively process nested structures
return _redact_value(value)
def redact_string(text: str, enabled: bool = True) -> str:
"""
Redact potential secrets from a string.
This is a best-effort function that looks for common patterns
like "api_key=xxx" or "password: xxx" in strings.
Args:
text: String to redact
enabled: If False, returns text unchanged
Returns:
String with potential secrets redacted
"""
if not enabled or not text:
return text
# Pattern for key=value or key: value
patterns = [
(r'(["\']?)(' + '|'.join(re.escape(k) for k in REDACT_KEYS) + r')(["\']?\s*[:=]\s*["\']?)([^"\'\s,}\]]+)',
r'\1\2\3[REDACTED]'),
]
result = text
for pattern, replacement in patterns:
result = re.sub(pattern, replacement, result, flags=re.IGNORECASE)
return result
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.11.3"
version = "0.11.4"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -0,0 +1 @@
"""Tests for the trace module."""
@@ -0,0 +1,450 @@
"""
Tests for the trace protocol module.
TDD tests for ActionEvent, TraceSink protocol, and redaction.
"""
import time
import json
class TestActionEvent:
"""Tests for ActionEvent dataclass."""
def test_action_event_creation_minimal(self):
"""Test creating ActionEvent with minimal fields."""
from praisonaiagents.trace.protocol import ActionEvent
event = ActionEvent(
event_type="agent_start",
timestamp=1736665496.123,
)
assert event.event_type == "agent_start"
assert event.timestamp == 1736665496.123
assert event.schema_version == "1.0"
assert event.agent_id is None
assert event.agent_name is None
assert event.tool_name is None
def test_action_event_creation_full(self):
"""Test creating ActionEvent with all fields."""
from praisonaiagents.trace.protocol import ActionEvent
event = ActionEvent(
event_type="tool_end",
timestamp=1736665497.456,
agent_id="agent-123",
agent_name="researcher",
tool_name="web_search",
tool_args={"query": "AI trends"},
tool_result_summary="Found 10 results",
duration_ms=800.5,
status="ok",
error_message=None,
metadata={"provider": "duckduckgo"},
)
assert event.event_type == "tool_end"
assert event.agent_name == "researcher"
assert event.tool_name == "web_search"
assert event.tool_args == {"query": "AI trends"}
assert event.duration_ms == 800.5
assert event.status == "ok"
assert event.metadata == {"provider": "duckduckgo"}
def test_action_event_to_dict(self):
"""Test ActionEvent.to_dict() method."""
from praisonaiagents.trace.protocol import ActionEvent
event = ActionEvent(
event_type="agent_start",
timestamp=1736665496.123,
agent_name="writer",
)
d = event.to_dict()
assert d["schema_version"] == "1.0"
assert d["event_type"] == "agent_start"
assert d["timestamp"] == 1736665496.123
assert d["agent_name"] == "writer"
# None values should be excluded
assert "tool_name" not in d or d["tool_name"] is None
def test_action_event_to_json(self):
"""Test ActionEvent.to_json() method."""
from praisonaiagents.trace.protocol import ActionEvent
event = ActionEvent(
event_type="tool_start",
timestamp=1736665497.0,
tool_name="read_file",
tool_args={"path": "/tmp/test.txt"},
)
json_str = event.to_json()
parsed = json.loads(json_str)
assert parsed["event_type"] == "tool_start"
assert parsed["tool_name"] == "read_file"
assert parsed["tool_args"] == {"path": "/tmp/test.txt"}
def test_action_event_types(self):
"""Test all valid event types."""
from praisonaiagents.trace.protocol import ActionEvent, ActionEventType
valid_types = [
ActionEventType.AGENT_START,
ActionEventType.AGENT_END,
ActionEventType.TOOL_START,
ActionEventType.TOOL_END,
ActionEventType.ERROR,
ActionEventType.OUTPUT,
]
for event_type in valid_types:
event = ActionEvent(
event_type=event_type.value,
timestamp=time.time(),
)
assert event.event_type == event_type.value
class TestRedaction:
"""Tests for redaction functionality."""
def test_redact_api_key(self):
"""Test redacting api_key field."""
from praisonaiagents.trace.redact import redact_dict
data = {"api_key": "sk-1234567890abcdef", "query": "hello"}
redacted = redact_dict(data)
assert redacted["api_key"] == "[REDACTED]"
assert redacted["query"] == "hello"
def test_redact_multiple_sensitive_keys(self):
"""Test redacting multiple sensitive keys."""
from praisonaiagents.trace.redact import redact_dict
data = {
"api_key": "sk-secret",
"password": "mypassword123",
"token": "bearer-token",
"authorization": "Bearer xyz",
"secret": "top-secret",
"normal_field": "visible",
}
redacted = redact_dict(data)
assert redacted["api_key"] == "[REDACTED]"
assert redacted["password"] == "[REDACTED]"
assert redacted["token"] == "[REDACTED]"
assert redacted["authorization"] == "[REDACTED]"
assert redacted["secret"] == "[REDACTED]"
assert redacted["normal_field"] == "visible"
def test_redact_nested_dict(self):
"""Test redacting nested dictionaries."""
from praisonaiagents.trace.redact import redact_dict
data = {
"config": {
"api_key": "sk-nested",
"endpoint": "https://api.example.com",
},
"name": "test",
}
redacted = redact_dict(data)
assert redacted["config"]["api_key"] == "[REDACTED]"
assert redacted["config"]["endpoint"] == "https://api.example.com"
assert redacted["name"] == "test"
def test_redact_list_of_dicts(self):
"""Test redacting list containing dictionaries."""
from praisonaiagents.trace.redact import redact_dict
data = {
"items": [
{"name": "item1", "secret": "hidden1"},
{"name": "item2", "secret": "hidden2"},
]
}
redacted = redact_dict(data)
assert redacted["items"][0]["name"] == "item1"
assert redacted["items"][0]["secret"] == "[REDACTED]"
assert redacted["items"][1]["secret"] == "[REDACTED]"
def test_redact_case_insensitive(self):
"""Test that redaction is case-insensitive."""
from praisonaiagents.trace.redact import redact_dict
data = {
"API_KEY": "key1",
"Api_Key": "key2",
"apiKey": "key3",
"APIKEY": "key4",
}
redacted = redact_dict(data)
assert redacted["API_KEY"] == "[REDACTED]"
assert redacted["Api_Key"] == "[REDACTED]"
assert redacted["apiKey"] == "[REDACTED]"
assert redacted["APIKEY"] == "[REDACTED]"
def test_redact_connection_strings(self):
"""Test redacting database connection strings."""
from praisonaiagents.trace.redact import redact_dict
data = {
"connection_string": "postgresql://user:pass@host/db",
"database_url": "mysql://root:secret@localhost/mydb",
"db_url": "redis://default:password@redis:6379",
}
redacted = redact_dict(data)
assert redacted["connection_string"] == "[REDACTED]"
assert redacted["database_url"] == "[REDACTED]"
assert redacted["db_url"] == "[REDACTED]"
def test_redact_preserves_original(self):
"""Test that redaction doesn't modify original dict."""
from praisonaiagents.trace.redact import redact_dict
original = {"api_key": "secret", "name": "test"}
redacted = redact_dict(original)
assert original["api_key"] == "secret"
assert redacted["api_key"] == "[REDACTED]"
def test_redact_empty_dict(self):
"""Test redacting empty dictionary."""
from praisonaiagents.trace.redact import redact_dict
assert redact_dict({}) == {}
def test_redact_none_values(self):
"""Test handling None values."""
from praisonaiagents.trace.redact import redact_dict
data = {"api_key": None, "name": "test"}
redacted = redact_dict(data)
# None values for sensitive keys should still be redacted
assert redacted["api_key"] == "[REDACTED]"
assert redacted["name"] == "test"
def test_redact_disabled(self):
"""Test disabling redaction."""
from praisonaiagents.trace.redact import redact_dict
data = {"api_key": "visible", "name": "test"}
result = redact_dict(data, enabled=False)
assert result["api_key"] == "visible"
assert result["name"] == "test"
class TestTraceSink:
"""Tests for TraceSink protocol."""
def test_noop_sink(self):
"""Test NoOpSink does nothing."""
from praisonaiagents.trace.protocol import NoOpSink, ActionEvent
sink = NoOpSink()
event = ActionEvent(event_type="agent_start", timestamp=time.time())
# Should not raise
sink.emit(event)
sink.flush()
sink.close()
def test_list_sink(self):
"""Test ListSink collects events."""
from praisonaiagents.trace.protocol import ListSink, ActionEvent
sink = ListSink()
event1 = ActionEvent(event_type="agent_start", timestamp=1.0)
event2 = ActionEvent(event_type="tool_start", timestamp=2.0)
sink.emit(event1)
sink.emit(event2)
events = sink.get_events()
assert len(events) == 2
assert events[0].event_type == "agent_start"
assert events[1].event_type == "tool_start"
def test_list_sink_clear(self):
"""Test ListSink.clear() method."""
from praisonaiagents.trace.protocol import ListSink, ActionEvent
sink = ListSink()
sink.emit(ActionEvent(event_type="test", timestamp=1.0))
assert len(sink.get_events()) == 1
sink.clear()
assert len(sink.get_events()) == 0
class TestTraceEmitter:
"""Tests for TraceEmitter class."""
def test_emitter_with_noop_sink(self):
"""Test TraceEmitter with NoOpSink."""
from praisonaiagents.trace.protocol import TraceEmitter, NoOpSink
emitter = TraceEmitter(sink=NoOpSink())
# Should not raise
emitter.agent_start("test-agent", agent_id="123")
emitter.tool_start("web_search", {"query": "test"})
emitter.tool_end("web_search", duration_ms=100, status="ok")
emitter.agent_end("test-agent", duration_ms=500)
def test_emitter_with_list_sink(self):
"""Test TraceEmitter collects events."""
from praisonaiagents.trace.protocol import TraceEmitter, ListSink
sink = ListSink()
emitter = TraceEmitter(sink=sink)
emitter.agent_start("researcher")
emitter.tool_start("search", {"q": "AI"})
emitter.tool_end("search", duration_ms=100, status="ok")
emitter.agent_end("researcher", duration_ms=200)
events = sink.get_events()
assert len(events) == 4
assert events[0].event_type == "agent_start"
assert events[1].event_type == "tool_start"
assert events[2].event_type == "tool_end"
assert events[3].event_type == "agent_end"
def test_emitter_redacts_by_default(self):
"""Test TraceEmitter redacts sensitive data by default."""
from praisonaiagents.trace.protocol import TraceEmitter, ListSink
sink = ListSink()
emitter = TraceEmitter(sink=sink, redact=True)
emitter.tool_start("api_call", {"api_key": "sk-secret", "query": "test"})
events = sink.get_events()
assert events[0].tool_args["api_key"] == "[REDACTED]"
assert events[0].tool_args["query"] == "test"
def test_emitter_disabled(self):
"""Test TraceEmitter when disabled."""
from praisonaiagents.trace.protocol import TraceEmitter, ListSink
sink = ListSink()
emitter = TraceEmitter(sink=sink, enabled=False)
emitter.agent_start("test")
emitter.tool_start("tool", {})
# No events should be emitted
assert len(sink.get_events()) == 0
def test_emitter_error_event(self):
"""Test TraceEmitter error event."""
from praisonaiagents.trace.protocol import TraceEmitter, ListSink
sink = ListSink()
emitter = TraceEmitter(sink=sink)
emitter.error("Something went wrong", tool_name="failing_tool")
events = sink.get_events()
assert len(events) == 1
assert events[0].event_type == "error"
assert events[0].error_message == "Something went wrong"
assert events[0].tool_name == "failing_tool"
def test_emitter_output_event(self):
"""Test TraceEmitter output event for final agent output."""
from praisonaiagents.trace.protocol import TraceEmitter, ListSink
sink = ListSink()
emitter = TraceEmitter(sink=sink)
emitter.output("The analysis shows three major trends...")
events = sink.get_events()
assert len(events) == 1
assert events[0].event_type == "output"
assert "three major trends" in events[0].tool_result_summary
class TestActionTraceConfig:
"""Tests for ActionTraceConfig."""
def test_config_defaults(self):
"""Test ActionTraceConfig default values."""
from praisonaiagents.trace.protocol import ActionTraceConfig
config = ActionTraceConfig()
assert config.enabled is True
assert config.redact is True
assert config.compact is False
assert config.sink_type == "noop"
assert config.file_path is None
def test_config_custom(self):
"""Test ActionTraceConfig custom values."""
from praisonaiagents.trace.protocol import ActionTraceConfig
config = ActionTraceConfig(
enabled=True,
redact=False,
compact=True,
sink_type="jsonl",
file_path="/tmp/trace.jsonl",
)
assert config.redact is False
assert config.compact is True
assert config.sink_type == "jsonl"
assert config.file_path == "/tmp/trace.jsonl"
class TestPerformance:
"""Performance tests for trace module."""
def test_noop_sink_overhead(self):
"""Test that NoOpSink has minimal overhead."""
from praisonaiagents.trace.protocol import TraceEmitter, NoOpSink, ActionEvent
emitter = TraceEmitter(sink=NoOpSink())
start = time.perf_counter()
for _ in range(10000):
emitter.tool_start("test", {"arg": "value"})
emitter.tool_end("test", duration_ms=1, status="ok")
elapsed = time.perf_counter() - start
# Should complete 10000 iterations in under 100ms
assert elapsed < 0.1, f"NoOpSink too slow: {elapsed:.3f}s for 10000 iterations"
def test_disabled_emitter_overhead(self):
"""Test that disabled emitter has near-zero overhead."""
from praisonaiagents.trace.protocol import TraceEmitter, ListSink
emitter = TraceEmitter(sink=ListSink(), enabled=False)
start = time.perf_counter()
for _ in range(10000):
emitter.tool_start("test", {"arg": "value"})
elapsed = time.perf_counter() - start
# Should complete 10000 iterations in under 500ms (very conservative)
assert elapsed < 0.5, f"Disabled emitter too slow: {elapsed:.3f}s"
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.11.3"
version = "0.11.4"
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.5.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.5.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.8.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.8.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+4 -4
View File
@@ -143,15 +143,15 @@ def main_callback(
# If no command provided, start interactive mode
if ctx.invoked_subcommand is None:
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
tui_config = TUIConfig(
tui_config = AsyncTUIConfig(
model="gpt-4o-mini",
show_logo=True,
compact_mode=state.output_format == OutputFormat.json,
show_status_bar=state.output_format != OutputFormat.json,
)
tui = PraisonTUI(config=tui_config)
tui = AsyncTUI(config=tui_config)
tui.run()
+94
View File
@@ -0,0 +1,94 @@
"""
PraisonAI Branding - Single Source of Truth.
This module provides unified branding assets (logo, version, product name)
for all interactive UIs. All frontends should import from here.
Usage:
from praisonai.cli.branding import get_logo, get_version, PRODUCT_NAME
"""
# Product name
PRODUCT_NAME = "Praison AI"
# ASCII Art Logos - responsive to terminal width
LOGO_LARGE = r"""
██████╗ ██████╗ █████╗ ██╗███████╗ ██████╗ ███╗ ██╗ █████╗ ██╗
██╔══██╗██╔══██╗██╔══██╗██║██╔════╝██╔═══██╗████╗ ██║ ██╔══██╗██║
██████╔╝██████╔╝███████║██║███████╗██║ ██║██╔██╗ ██║ ███████║██║
██╔═══╝ ██╔══██╗██╔══██║██║╚════██║██║ ██║██║╚██╗██║ ██╔══██║██║
██║ ██║ ██║██║ ██║██║███████║╚██████╔╝██║ ╚████║ ██║ ██║██║
╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═╝ ╚═╝╚═╝
"""
LOGO_MEDIUM = r"""
╔═╗┬─┐┌─┐┬┌─┐┌─┐┌┐┌ ╔═╗╦
╠═╝├┬┘├─┤│└─┐│ ││││ ╠═╣║
╩ ┴└─┴ ┴┴└─┘└─┘┘└┘ ╩ ╩╩
"""
LOGO_SMALL = "▶ Praison AI"
LOGO_MINIMAL = "Praison AI"
def get_version() -> str:
"""Get PraisonAI version string."""
try:
from praisonai import __version__
return __version__
except Exception:
return "1.0.0"
def get_logo(width: int = 80) -> str:
"""
Get appropriate logo based on terminal width.
Args:
width: Terminal width in characters
Returns:
ASCII art logo string appropriate for the width
"""
if width >= 75:
return LOGO_LARGE
elif width >= 40:
return LOGO_MEDIUM
else:
return LOGO_SMALL
def get_banner(width: int = 80, show_version: bool = True, model: str = None) -> str:
"""
Get full banner with logo, version, and optional model info.
Args:
width: Terminal width
show_version: Whether to include version
model: Optional model name to display
Returns:
Complete banner string
"""
lines = []
# Logo
logo = get_logo(width)
lines.append(logo.strip())
lines.append("")
# Version and model
version_line = f" v{get_version()}"
if model:
version_line += f" · Model: {model}"
lines.append(version_line)
lines.append("")
return "\n".join(lines)
def get_welcome_tips() -> str:
"""Get welcome tips for new sessions."""
return """ Type your message and press Enter. Use /help for commands.
Use PageUp/PageDown or Ctrl+Up/Down to scroll."""
+20 -11
View File
@@ -67,6 +67,7 @@ def chat_main(
workspace: Optional[str] = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
no_acp: bool = typer.Option(False, "--no-acp", help="Disable ACP tools"),
no_lsp: bool = typer.Option(False, "--no-lsp", help="Disable LSP tools"),
safe_mode: bool = typer.Option(False, "--safe", help="Safe mode: require approval for file writes and commands"),
autonomy: bool = typer.Option(True, "--autonomy/--no-autonomy", help="Enable agent autonomy for complex tasks"),
# NEW: Agent-like consolidated params for ALL GREEN consistency
knowledge: Optional[str] = typer.Option(
@@ -107,8 +108,8 @@ def chat_main(
flag_value="true",
),
output: Optional[str] = typer.Option(
None, "--output",
help="Output preset. Use --output=verbose, --output=minimal, --output=silent",
None, "--output", "-o",
help="Output mode (default: actions). Options: actions, plain, verbose, json, silent",
),
execution: Optional[str] = typer.Option(
None, "--execution",
@@ -126,6 +127,7 @@ def chat_main(
),
profile: bool = typer.Option(False, "--profile", help="Enable CLI profiling (timing breakdown)"),
profile_deep: bool = typer.Option(False, "--profile-deep", help="Enable deep profiling (cProfile stats, higher overhead)"),
debug: bool = typer.Option(False, "--debug", help="Enable debug logging to ~/.praisonai/async_tui_debug.log"),
# UI backend selection
ui_backend: str = typer.Option("auto", "--ui-backend", help="UI backend: auto, plain, rich, mg (middle-ground)"),
json_output: bool = typer.Option(False, "--json", help="Output JSON (forces plain backend)"),
@@ -171,30 +173,37 @@ def chat_main(
typer.echo(" Use: praisonai chat \"your prompt\" --profile", err=True)
# Parse memory flag: --no-memory takes precedence, then --memory value
memory_value = _parse_memory_flag(memory, no_memory)
# TODO: Pass memory_value to TUI when memory support is added
_parse_memory_flag(memory, no_memory)
# Use the new TUI application (Aider/Claude Code style)
from praisonai.cli.interactive.tui_app import PraisonTUI, TUIConfig
# Set approval mode based on --safe flag
import os
if safe_mode:
os.environ["PRAISON_APPROVAL_MODE"] = "prompt"
else:
os.environ["PRAISON_APPROVAL_MODE"] = "auto"
tui_config = TUIConfig(
# 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_tips=not compact,
show_status_bar=not compact,
compact_mode=compact,
session_id=session_id,
workspace=workspace,
debug=debug,
)
tui = PraisonTUI(config=tui_config)
tui = AsyncTUI(config=tui_config)
if prompt:
# Single prompt mode
# Single prompt mode - direct response, no streaming
response = tui.run_single(prompt)
if response:
print(response)
else:
# Interactive TUI mode
# Interactive split-pane TUI mode
tui.run()
@@ -23,6 +23,7 @@ def code_main(
file: Optional[List[str]] = typer.Option(None, "--file", "-f", help="Attach file(s) to context"),
no_acp: bool = typer.Option(False, "--no-acp", help="Disable ACP tools (file operations)"),
no_lsp: bool = typer.Option(False, "--no-lsp", help="Disable LSP tools (code intelligence)"),
safe_mode: bool = typer.Option(False, "--safe", help="Safe mode: require approval for file writes and commands"),
session_id: Optional[str] = typer.Option(None, "--session", "-s", help="Session ID to resume"),
continue_session: bool = typer.Option(False, "--continue", "-c", help="Continue last session"),
autonomy: bool = typer.Option(True, "--autonomy/--no-autonomy", help="Enable agent autonomy for complex tasks"),
@@ -54,6 +55,12 @@ def code_main(
# Enable code-specific environment
os.environ["PRAISONAI_CODE_MODE"] = "true"
# Set approval mode based on --safe flag
if safe_mode:
os.environ["PRAISON_APPROVAL_MODE"] = "prompt"
else:
os.environ["PRAISON_APPROVAL_MODE"] = "auto"
# Handle profiling for single prompt mode
if prompt and (profile or profile_deep):
_run_profiled_code(
+30 -1
View File
@@ -29,6 +29,7 @@ def run_main(
max_tokens: int = typer.Option(16000, "--max-tokens", help="Maximum output tokens"),
profile: bool = typer.Option(False, "--profile", help="Enable CLI profiling (timing breakdown)"),
profile_deep: bool = typer.Option(False, "--profile-deep", help="Enable deep profiling (cProfile stats, higher overhead)"),
output_mode: Optional[str] = typer.Option(None, "--output", "-o", help="Output mode: actions, verbose, silent, stream"),
):
"""
Run agents from a file or prompt.
@@ -108,6 +109,7 @@ def run_main(
memory=memory,
tools=tools,
max_tokens=max_tokens,
output_mode=output_mode,
)
else:
# Run as prompt
@@ -120,6 +122,7 @@ def run_main(
memory=memory,
tools=tools,
max_tokens=max_tokens,
output_mode=output_mode,
)
@@ -134,6 +137,7 @@ def _run_from_file(
memory: bool = False,
tools: Optional[str] = None,
max_tokens: int = 16000,
output_mode: Optional[str] = None,
):
"""Run agents from a YAML file."""
output = get_output_controller()
@@ -178,12 +182,37 @@ def _run_prompt(
memory: bool = False,
tools: Optional[str] = None,
max_tokens: int = 16000,
output_mode: Optional[str] = None,
):
"""Run a direct prompt."""
output = get_output_controller()
try:
# Use existing handle_direct_prompt
# If output_mode is "actions", use direct Agent with actions preset
if output_mode == "actions":
from praisonaiagents import Agent
agent_config = {
"name": "RunAgent",
"role": "Assistant",
"goal": "Complete the task",
"output": "actions", # Use actions preset
}
if model:
agent_config["llm"] = model
agent = Agent(**agent_config)
result = agent.start(prompt)
output.emit_result(
message="Prompt completed",
data={"result": str(result) if result else None}
)
# Don't print result again - actions mode already shows output
return
# Use existing handle_direct_prompt for other modes
from praisonai.cli.main import PraisonAI
praison = PraisonAI()
@@ -160,6 +160,8 @@ class ActionOrchestrator:
async def _create_fallback_plan(self, prompt: str, plan_id: str) -> ActionResult:
"""Create a fallback plan when ACP is not available for planning."""
import re # Import at function level for all pattern matching
# This is a simplified fallback - in production, this would use LLM
steps = []
@@ -168,7 +170,6 @@ class ActionOrchestrator:
if "create" in prompt_lower or "new file" in prompt_lower:
# Extract file path if mentioned
import re
file_match = re.search(r'(?:create|new)\s+(?:a\s+)?(?:file\s+)?(?:called\s+)?([^\s]+\.\w+)', prompt_lower)
target = file_match.group(1) if file_match else "new_file.py"
@@ -188,7 +189,8 @@ class ActionOrchestrator:
))
elif "edit" in prompt_lower or "modify" in prompt_lower or "change" in prompt_lower:
file_match = re.search(r'(?:edit|modify|change)\s+([^\s]+\.\w+)', prompt_lower)
# Match patterns like "edit file X", "edit X", "modify X"
file_match = re.search(r'(?:edit|modify|change)\s+(?:file\s+)?([^\s]+(?:\.\w+)?)', prompt_lower)
target = file_match.group(1) if file_match else ""
if target:
@@ -197,11 +199,13 @@ class ActionOrchestrator:
action_type=ActionType.FILE_EDIT,
description=f"Edit file: {target}",
target=target,
params={"changes": prompt}
params={"changes": prompt},
status=ActionStatus.APPROVED # Auto-approve for fallback
))
elif "delete" in prompt_lower or "remove" in prompt_lower:
file_match = re.search(r'(?:delete|remove)\s+([^\s]+\.\w+)', prompt_lower)
# Match patterns like "delete file X", "delete X", "remove X"
file_match = re.search(r'(?:delete|remove)\s+(?:file\s+)?([^\s]+(?:\.\w+)?)', prompt_lower)
target = file_match.group(1) if file_match else ""
if target:
@@ -209,7 +213,8 @@ class ActionOrchestrator:
id=self._generate_step_id(plan_id, 0),
action_type=ActionType.FILE_DELETE,
description=f"Delete file: {target}",
target=target
target=target,
status=ActionStatus.APPROVED # Auto-approve for fallback
))
elif "rename" in prompt_lower:
@@ -218,19 +223,22 @@ class ActionOrchestrator:
action_type=ActionType.REFACTOR,
description="Rename operation",
target="",
params={"prompt": prompt}
params={"prompt": prompt},
status=ActionStatus.APPROVED # Auto-approve for fallback
))
elif "run" in prompt_lower or "execute" in prompt_lower:
cmd_match = re.search(r'(?:run|execute)\s+[`"\']?([^`"\']+)[`"\']?', prompt_lower)
command = cmd_match.group(1) if cmd_match else ""
elif "run" in prompt_lower or "execute" in prompt_lower or "command" in prompt_lower:
# Match patterns like "run command: X", "execute X", "run X"
cmd_match = re.search(r'(?:run|execute)\s+(?:command[:\s]+)?[`"\']?(.+?)[`"\']?$', prompt_lower)
command = cmd_match.group(1).strip() if cmd_match else ""
if command:
steps.append(ActionStep(
id=self._generate_step_id(plan_id, 0),
action_type=ActionType.SHELL_COMMAND,
description=f"Execute: {command}",
target=command
target=command,
status=ActionStatus.APPROVED # Auto-approve for fallback
))
if not steps:
@@ -596,20 +596,34 @@ class CodeIntelligenceRouter:
def _extract_symbol_from_query(self, query: str) -> Optional[str]:
"""Extract symbol name from query."""
# Common patterns for symbol extraction
# Common patterns for symbol extraction (order matters - more specific first)
patterns = [
r'(?:definition\s+of|where\s+is|find|show)\s+["\']?(\w+)["\']?',
# "find all references to X" - capture X after "to"
r'references?\s+to\s+["\']?(\w+)["\']?',
# "definition of X" - capture X after "of"
r'definition\s+of\s+["\']?(\w+)["\']?',
# "where is X defined/used"
r'where\s+is\s+["\']?(\w+)["\']?\s+(?:defined|used)',
# Quoted or backticked symbols
r'["\'](\w+)["\']',
r'`(\w+)`',
# "X is defined/used"
r'(\w+)\s+(?:is\s+)?(?:defined|used)',
# Generic "find/show X" - but filter common words
r'(?:find|show)\s+["\']?(\w+)["\']?',
]
# Words to filter out
filter_words = {'the', 'a', 'an', 'is', 'are', 'in', 'of', 'to', 'for',
'all', 'any', 'some', 'this', 'that', 'where', 'what',
'definition', 'references', 'reference', 'symbol', 'symbols'}
for pattern in patterns:
match = re.search(pattern, query, re.IGNORECASE)
if match:
symbol = match.group(1)
# Filter out common words
if symbol.lower() not in ['the', 'a', 'an', 'is', 'are', 'in', 'of', 'to', 'for']:
if symbol.lower() not in filter_words:
return symbol
return None
@@ -40,7 +40,7 @@ class RuntimeConfig:
workspace: str = "."
lsp_enabled: bool = True
acp_enabled: bool = True
approval_mode: str = "manual" # manual, auto, scoped
approval_mode: str = "auto" # auto (full privileges), manual, scoped
trace_enabled: bool = False
trace_file: Optional[str] = None
json_output: bool = False
@@ -146,15 +146,21 @@ class InteractiveRuntime:
def read_only(self) -> bool:
"""Check if runtime is in read-only mode.
Note: When PRAISON_APPROVAL_MODE=auto, we allow writes even without ACP
to support non-interactive automation and testing scenarios.
Note: When approval_mode=auto (via config or env), we allow writes even
without ACP to support non-interactive automation and testing scenarios.
"""
# Auto-approval mode bypasses read-only for automation
import os as _os
approval_mode = _os.environ.get("PRAISON_APPROVAL_MODE", "").lower()
if approval_mode == "auto":
# Check config approval_mode first (takes precedence)
if self.config.approval_mode == "auto":
return False
return self._read_only or not self.acp_ready
# Then check environment variable as fallback
import os as _os
env_approval_mode = _os.environ.get("PRAISON_APPROVAL_MODE", "").lower()
if env_approval_mode == "auto":
return False
# Only enforce read-only if explicitly set or ACP not ready
return self._read_only
def get_status(self) -> Dict[str, Any]:
"""Get runtime status."""
@@ -72,7 +72,7 @@ class ToolConfig:
enable_acp: bool = True
enable_lsp: bool = True
enable_basic: bool = True
approval_mode: str = "auto" # auto, manual, scoped
approval_mode: str = "auto" # auto (full privileges), manual, scoped
lsp_enabled: bool = True
acp_enabled: bool = True
File diff suppressed because it is too large. Load diff
@@ -50,8 +50,9 @@ class InteractiveConfig:
autonomy: bool = True # Enable autonomy by default for complex tasks
autonomy_config: Optional[dict] = None # Custom autonomy config
# Approval mode
approval_mode: Union[str, ApprovalMode] = "prompt"
# Approval mode - default to "auto" for full privileges in interactive mode
# Use "prompt" for safer mode that asks before each action
approval_mode: Union[str, ApprovalMode] = "auto"
# File attachments
files: List[str] = field(default_factory=list)
@@ -24,6 +24,9 @@ class InteractiveEventType(Enum):
TOOL_START = "tool.start"
TOOL_END = "tool.end"
# Stage transitions (PLAN → EXPLORE → BUILD → REVIEW)
STAGE_CHANGE = "stage.change"
# Approval/permission flow
APPROVAL_ASKED = "approval.asked"
APPROVAL_ANSWERED = "approval.answered"
@@ -0,0 +1,456 @@
"""
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
# ============================================================================
# Branding - Import from unified source
# ============================================================================
from praisonai.cli.branding import get_logo, get_version
# ============================================================================
# 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()
@@ -45,63 +45,10 @@ def _check_rich():
# ============================================================================
# ASCII Art Logo
# Branding - Import from unified source
# ============================================================================
# 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"
from praisonai.cli.branding import get_logo, get_version
# ============================================================================
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==3.7.5 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==3.7.8 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "3.7.5"
__version__ = "3.7.8"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.11.3",
"praisonaiagents>=0.11.4",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
@@ -0,0 +1,848 @@
"""
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.branding import get_logo, LOGO_LARGE
logo = get_logo(100)
assert logo == LOGO_LARGE
def test_logo_medium_terminal(self):
"""Test logo for medium terminal."""
from praisonai.cli.branding import get_logo, LOGO_MEDIUM
logo = get_logo(50)
assert logo == LOGO_MEDIUM
def test_logo_narrow_terminal(self):
"""Test logo for narrow terminal."""
from praisonai.cli.branding import get_logo, LOGO_SMALL
logo = get_logo(30)
assert logo == LOGO_SMALL
def test_logo_contains_praison_ai(self):
"""Test that logo contains 'Praison AI' branding."""
from praisonai.cli.branding import LOGO_SMALL
assert "Praison AI" in LOGO_SMALL
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.branding import LOGO_LARGE
assert len(LOGO_LARGE) > 0
assert "██" in LOGO_LARGE
def test_logo_small_exists(self):
"""Test LOGO_SMALL constant exists."""
from praisonai.cli.branding import LOGO_SMALL
assert len(LOGO_SMALL) > 0
def test_logo_minimal_branding(self):
"""Test LOGO_MINIMAL has correct branding."""
from praisonai.cli.branding 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)
class TestLogoDisplay:
"""Tests for logo display functionality."""
def test_logo_always_shows_when_enabled(self):
"""Test logo shows even when messages exist."""
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig, ChatMessage
config = AsyncTUIConfig(show_logo=True)
tui = AsyncTUI(config=config)
# Add a message
tui.messages.append(ChatMessage(role="system", content="Test message"))
# Logo should still be in output
output = tui._format_output()
assert "Praison" in output or "██" in output # Logo contains these
def test_logo_hidden_when_disabled(self):
"""Test logo is hidden when show_logo=False."""
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
config = AsyncTUIConfig(show_logo=False)
tui = AsyncTUI(config=config)
output = tui._format_output()
# Should not contain the large logo characters
assert "██████" not in output
def test_tips_hidden_after_first_message(self):
"""Test tips are hidden after first message but logo remains."""
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig, ChatMessage
config = AsyncTUIConfig(show_logo=True)
tui = AsyncTUI(config=config)
# Before messages - tips should show
output_before = tui._format_output()
assert "Type your message" in output_before
# Add a message
tui.messages.append(ChatMessage(role="user", content="Hello"))
# After messages - tips should be hidden but logo remains
output_after = tui._format_output()
assert "Type your message" not in output_after
# Logo should still be there
assert "Model:" in output_after
class TestToolCallVisibility:
"""Tests for tool call visibility."""
def test_tool_role_in_format_output(self):
"""Test tool role messages are formatted correctly."""
from praisonai.cli.interactive.async_tui import AsyncTUI, ChatMessage
tui = AsyncTUI()
tui.messages.append(ChatMessage(role="tool", content="read_file completed"))
output = tui._format_output()
assert "⚙" in output
assert "read_file completed" in output
class TestDebugMode:
"""Tests for debug mode (/debug command and --debug flag)."""
def test_debug_command_toggles_mode(self):
"""Test /debug command toggles debug mode."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
assert tui.config.debug is False
result = tui._handle_command("/debug")
assert result is True
assert tui.config.debug is True
assert "enabled" in tui.messages[0].content.lower()
assert "async_tui_debug.log" in tui.messages[0].content
def test_debug_flag_in_config(self):
"""Test debug flag can be set via config."""
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
config = AsyncTUIConfig(debug=True)
tui = AsyncTUI(config=config)
assert tui.config.debug is True
class TestPlanningIntegration:
"""Tests for planning integration (/plan command)."""
def test_plan_command_no_args(self):
"""Test /plan command without arguments shows usage."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/plan")
assert result is True
assert len(tui.messages) == 1
assert "Usage:" in tui.messages[0].content
assert "/plan" in tui.messages[0].content
def test_plan_command_with_task(self):
"""Test /plan command with a task description."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
# Mock _queue_or_execute to prevent actual execution
executed_prompts = []
tui._queue_or_execute = lambda p: executed_prompts.append(p)
result = tui._handle_command("/plan refactor the auth module")
assert result is True
assert len(tui.messages) == 1
assert "Creating plan" in tui.messages[0].content
assert len(executed_prompts) == 1
assert "refactor the auth module" in executed_prompts[0]
class TestHandoffSubagents:
"""Tests for handoff/subagents (/handoff command)."""
def test_handoff_command_no_args(self):
"""Test /handoff command without arguments shows usage."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/handoff")
assert result is True
assert len(tui.messages) == 1
assert "Usage:" in tui.messages[0].content
assert "code" in tui.messages[0].content
assert "research" in tui.messages[0].content
def test_handoff_command_missing_task(self):
"""Test /handoff command with only agent type."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command("/handoff code")
assert result is True
assert "specify both" in tui.messages[0].content.lower()
def test_handoff_command_code_agent(self):
"""Test /handoff code command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
executed_prompts = []
tui._queue_or_execute = lambda p: executed_prompts.append(p)
result = tui._handle_command('/handoff code "fix the bug"')
assert result is True
assert "Handing off" in tui.messages[0].content
assert len(executed_prompts) == 1
assert "code" in executed_prompts[0].lower()
def test_handoff_command_research_agent(self):
"""Test /handoff research command."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
executed_prompts = []
tui._queue_or_execute = lambda p: executed_prompts.append(p)
result = tui._handle_command('/handoff research "find best practices"')
assert result is True
assert len(executed_prompts) == 1
assert "research" in executed_prompts[0].lower()
def test_handoff_command_unknown_agent(self):
"""Test /handoff with unknown agent type."""
from praisonai.cli.interactive.async_tui import AsyncTUI
tui = AsyncTUI()
result = tui._handle_command('/handoff unknown "some task"')
assert result is True
assert "Unknown agent type" in tui.messages[0].content
class TestInteractiveRuntimeReadOnly:
"""Tests for InteractiveRuntime read_only property."""
def test_read_only_false_when_config_approval_auto(self):
"""Test read_only is False when config.approval_mode is 'auto'."""
from praisonai.cli.features.interactive_runtime import InteractiveRuntime, RuntimeConfig
config = RuntimeConfig(approval_mode="auto")
runtime = InteractiveRuntime(config)
# Even without ACP ready, should not be read-only in auto mode
assert runtime.read_only is False
def test_read_only_false_when_env_approval_auto(self):
"""Test read_only is False when PRAISON_APPROVAL_MODE env is 'auto'."""
import os
from praisonai.cli.features.interactive_runtime import InteractiveRuntime, RuntimeConfig
# Set env var
old_val = os.environ.get("PRAISON_APPROVAL_MODE")
os.environ["PRAISON_APPROVAL_MODE"] = "auto"
try:
config = RuntimeConfig(approval_mode="manual") # Config says manual
runtime = InteractiveRuntime(config)
# Config takes precedence, so this should be read-only
# But env var should also work as fallback
# Actually config.approval_mode="manual" so env var is checked
# Wait, config says manual, so it won't return False on first check
# Then env var check should return False
assert runtime.read_only is False
finally:
if old_val is None:
os.environ.pop("PRAISON_APPROVAL_MODE", None)
else:
os.environ["PRAISON_APPROVAL_MODE"] = old_val
def test_read_only_true_when_manual_mode(self):
"""Test read_only respects _read_only flag in manual mode."""
import os
from praisonai.cli.features.interactive_runtime import InteractiveRuntime, RuntimeConfig
# Clear env var
old_val = os.environ.get("PRAISON_APPROVAL_MODE")
os.environ.pop("PRAISON_APPROVAL_MODE", None)
try:
config = RuntimeConfig(approval_mode="manual")
runtime = InteractiveRuntime(config)
runtime._read_only = True # Explicitly set read-only
assert runtime.read_only is True
finally:
if old_val is not None:
os.environ["PRAISON_APPROVAL_MODE"] = old_val
@@ -0,0 +1,117 @@
"""
Tests for the unified branding module.
This module provides the single source of truth for logo/version/branding
across all interactive UIs.
"""
import pytest
class TestBrandingConstants:
"""Tests for branding constants."""
def test_product_name(self):
"""Test PRODUCT_NAME constant."""
from praisonai.cli.branding import PRODUCT_NAME
assert PRODUCT_NAME == "Praison AI"
def test_logo_large_exists(self):
"""Test LOGO_LARGE constant exists and has ASCII art."""
from praisonai.cli.branding import LOGO_LARGE
assert len(LOGO_LARGE) > 0
assert "██" in LOGO_LARGE
def test_logo_medium_exists(self):
"""Test LOGO_MEDIUM constant exists."""
from praisonai.cli.branding import LOGO_MEDIUM
assert len(LOGO_MEDIUM) > 0
def test_logo_small_exists(self):
"""Test LOGO_SMALL constant exists and has branding."""
from praisonai.cli.branding import LOGO_SMALL
assert len(LOGO_SMALL) > 0
assert "Praison AI" in LOGO_SMALL
def test_logo_minimal_exists(self):
"""Test LOGO_MINIMAL constant exists and has branding."""
from praisonai.cli.branding import LOGO_MINIMAL
assert LOGO_MINIMAL == "Praison AI"
class TestGetLogo:
"""Tests for get_logo function."""
def test_get_logo_wide_terminal(self):
"""Test get_logo returns LOGO_LARGE for wide terminals."""
from praisonai.cli.branding import get_logo, LOGO_LARGE
assert get_logo(100) == LOGO_LARGE
assert get_logo(80) == LOGO_LARGE
assert get_logo(75) == LOGO_LARGE
def test_get_logo_medium_terminal(self):
"""Test get_logo returns LOGO_MEDIUM for medium terminals."""
from praisonai.cli.branding import get_logo, LOGO_MEDIUM
assert get_logo(74) == LOGO_MEDIUM
assert get_logo(50) == LOGO_MEDIUM
assert get_logo(40) == LOGO_MEDIUM
def test_get_logo_narrow_terminal(self):
"""Test get_logo returns LOGO_SMALL for narrow terminals."""
from praisonai.cli.branding import get_logo, LOGO_SMALL
assert get_logo(39) == LOGO_SMALL
assert get_logo(30) == LOGO_SMALL
assert get_logo(20) == LOGO_SMALL
class TestGetVersion:
"""Tests for get_version function."""
def test_get_version_returns_string(self):
"""Test get_version returns a string."""
from praisonai.cli.branding import get_version
version = get_version()
assert isinstance(version, str)
assert len(version) > 0
def test_get_version_format(self):
"""Test get_version returns a valid version format."""
from praisonai.cli.branding import get_version
version = get_version()
# Should be in format X.Y.Z or similar
parts = version.split(".")
assert len(parts) >= 2
class TestGetBanner:
"""Tests for get_banner function."""
def test_get_banner_includes_logo(self):
"""Test get_banner includes logo."""
from praisonai.cli.branding import get_banner
banner = get_banner(100)
assert "██" in banner or "Praison" in banner
def test_get_banner_includes_version(self):
"""Test get_banner includes version when requested."""
from praisonai.cli.branding import get_banner, get_version
banner = get_banner(100, show_version=True)
version = get_version()
assert version in banner
def test_get_banner_includes_model(self):
"""Test get_banner includes model when provided."""
from praisonai.cli.branding import get_banner
banner = get_banner(100, model="gpt-4o")
assert "gpt-4o" in banner
class TestGetWelcomeTips:
"""Tests for get_welcome_tips function."""
def test_get_welcome_tips_content(self):
"""Test get_welcome_tips returns helpful tips."""
from praisonai.cli.branding import get_welcome_tips
tips = get_welcome_tips()
assert "Enter" in tips or "help" in tips
assert len(tips) > 0
@@ -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.branding import get_logo, LOGO_LARGE
logo = get_logo(100)
assert logo == LOGO_LARGE
def test_logo_medium_terminal(self):
"""Test logo for medium terminal."""
from praisonai.cli.branding import get_logo, LOGO_MEDIUM
logo = get_logo(50)
assert logo == LOGO_MEDIUM
def test_logo_narrow_terminal(self):
"""Test logo for narrow terminal."""
from praisonai.cli.branding import get_logo, LOGO_SMALL
logo = get_logo(30)
assert logo == LOGO_SMALL
def test_logo_contains_praison_ai(self):
"""Test that logo contains 'Praison AI' branding."""
from praisonai.cli.branding import LOGO_SMALL
assert "Praison AI" in LOGO_SMALL
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.branding import LOGO_LARGE
assert len(LOGO_LARGE) > 0
assert "██" in LOGO_LARGE # Contains block characters
def test_logo_small_exists(self):
"""Test LOGO_SMALL constant exists."""
from praisonai.cli.branding import LOGO_SMALL
assert len(LOGO_SMALL) > 0
def test_logo_minimal_branding(self):
"""Test LOGO_MINIMAL has correct branding."""
from praisonai.cli.branding 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
+14 -14
View File
@@ -76,24 +76,24 @@ class TestGetLogo:
def test_logo_wide_terminal(self):
"""Test logo for wide terminal."""
from praisonai.cli.interactive.tui_app import get_logo, LOGO_SMALL
from praisonai.cli.branding import get_logo, LOGO_LARGE
logo = get_logo(100)
assert logo == LOGO_SMALL
assert logo == LOGO_LARGE
def test_logo_medium_terminal(self):
"""Test logo for medium terminal."""
from praisonai.cli.interactive.tui_app import get_logo, LOGO_TINY
from praisonai.cli.branding import get_logo, LOGO_MEDIUM
logo = get_logo(65)
assert logo == LOGO_TINY
assert logo == LOGO_MEDIUM
def test_logo_narrow_terminal(self):
"""Test logo for narrow terminal."""
from praisonai.cli.interactive.tui_app import get_logo, LOGO_MINIMAL
from praisonai.cli.branding import get_logo, LOGO_SMALL
logo = get_logo(50)
assert logo == LOGO_MINIMAL
logo = get_logo(30)
assert logo == LOGO_SMALL
class TestPraisonTUI:
@@ -324,21 +324,21 @@ class TestASCIILogos:
def test_logo_large_exists(self):
"""Test LOGO_LARGE constant exists."""
from praisonai.cli.interactive.tui_app import LOGO_LARGE
from praisonai.cli.branding 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
from praisonai.cli.branding 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
"""Test LOGO_MEDIUM constant exists (was LOGO_TINY)."""
from praisonai.cli.branding import LOGO_MEDIUM
assert len(LOGO_MEDIUM) > 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
from praisonai.cli.branding import LOGO_MINIMAL
assert "Praison AI" 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.3" },
{ name = "praisonaiagents", specifier = ">=0.11.4" },
{ 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.3"
version = "0.11.4"
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/96/4a/ab73d7381b0e9139f722dad51f9cb63575032c3d11aa83bc281f76bfd13a/praisonaiagents-0.11.3.tar.gz", hash = "sha256:5aa024935184eaa66578ff4c888e06b052d55568c68bfb702dd35c8a7951e10b", size = 973113, upload-time = "2026-01-11T07:29:29.515Z" }
sdist = { url = "https://files.pythonhosted.org/packages/8a/c0/1deb76b4c422df0e7f46799ecdd618592d3909465bee00f2762dcf4a1d99/praisonaiagents-0.11.4.tar.gz", hash = "sha256:3cf03310d6e453abf06021ca33683c658ee260a7da3521bdeb6de9cda76ca890", size = 981161, upload-time = "2026-01-12T07:31:44.077Z" }
wheels = [
{ 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" },
{ url = "https://files.pythonhosted.org/packages/60/da/daa98ae0b2ed7e5bb9132f21cc7574b72db987872791d96ee67ce46f9ae6/praisonaiagents-0.11.4-py3-none-any.whl", hash = "sha256:8155d0912412a4a24ad9c9141fba5d7c922f990294cb473393f9b58e25e51e32", size = 964101, upload-time = "2026-01-12T07:31:41.673Z" },
]
[[package]]