Compare commits

...
1 Commits
Author SHA1 Message Date
Mervin Praison 2c4dd0e2d4 Release v3.7.8 2026-01-12 07:39:39 +00:00
36 changed files with 2045 additions and 253 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.7" \
"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.7" \
"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.7" \
"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.7.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.7.7.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]"
+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."""
+12 -2
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)"),
@@ -174,6 +176,13 @@ def chat_main(
# TODO: Pass memory_value to TUI when memory support is added
_parse_memory_flag(memory, no_memory)
# 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"
# Use the async TUI (non-blocking, scrollable output)
from praisonai.cli.interactive.async_tui import AsyncTUI, AsyncTUIConfig
@@ -183,6 +192,7 @@ def chat_main(
show_status_bar=not compact,
session_id=session_id,
workspace=workspace,
debug=debug,
)
tui = AsyncTUI(config=tui_config)
@@ -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
@@ -13,6 +13,8 @@ Key fixes:
- Proper app.invalidate() for UI refresh
"""
import logging
import os
import shutil
import threading
import time
@@ -22,44 +24,41 @@ from typing import List, Optional
from datetime import datetime
# ============================================================================
# ASCII Art Logo - "Praison AI" branding
# Debug Logging - Only enabled with --debug flag or PRAISON_DEBUG=1
# ============================================================================
LOGO = r"""
██████╗ ██████╗ █████╗ ██╗███████╗ ██████╗ ███╗ ██╗ █████╗ ██╗
██╔══██╗██╔══██╗██╔══██╗██║██╔════╝██╔═══██╗████╗ ██║ ██╔══██╗██║
██████╔╝██████╔╝███████║██║███████╗██║ ██║██╔██╗ ██║ ███████║██║
██╔═══╝ ██╔══██╗██╔══██║██║╚════██║██║ ██║██║╚██╗██║ ██╔══██║██║
██║ ██║ ██║██║ ██║██║███████║╚██████╔╝██║ ╚████║ ██║ ██║██║
╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═╝ ╚═╝╚═╝
"""
logger = logging.getLogger(__name__)
_debug_initialized = False
LOGO_SMALL = r"""
╔═╗┬─┐┌─┐┬┌─┐┌─┐┌┐┌ ╔═╗╦
╠═╝├┬┘├─┤│└─┐│ ││││ ╠═╣║
╩ ┴└─┴ ┴┴└─┘└─┘┘└┘ ╩ ╩╩
"""
def _init_debug_logging():
"""Initialize debug logging to file. Only called when debug mode is enabled."""
global _debug_initialized
if _debug_initialized:
return
_debug_log_file = os.path.expanduser("~/.praisonai/async_tui_debug.log")
os.makedirs(os.path.dirname(_debug_log_file), exist_ok=True)
_file_handler = logging.FileHandler(_debug_log_file, mode='a')
_file_handler.setLevel(logging.DEBUG)
_file_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
))
logger.setLevel(logging.DEBUG)
logger.addHandler(_file_handler)
_debug_initialized = True
logger.debug("Debug logging initialized")
LOGO_MINIMAL = "▶ Praison AI"
# Check if debug mode is enabled via environment variable
if os.environ.get("PRAISON_DEBUG", "").lower() in ("1", "true", "yes"):
_init_debug_logging()
# ============================================================================
# Branding - Import from unified source
# ============================================================================
def get_logo(width: int = 80) -> str:
"""Get appropriate logo based on terminal width."""
if width >= 75:
return LOGO
elif width >= 40:
return LOGO_SMALL
else:
return LOGO_MINIMAL
def get_version() -> str:
"""Get PraisonAI version."""
try:
from praisonai import __version__
return __version__
except Exception:
return "1.0.0"
from praisonai.cli.branding import get_logo, get_version
# ============================================================================
@@ -77,6 +76,8 @@ class AsyncTUIConfig:
history_file: Optional[str] = None
compact_mode: bool = False
multiline_mode: bool = False
autonomy_mode: bool = False # Enable autonomous task delegation
debug: bool = False # Enable debug logging to file (~/.praisonai/async_tui_debug.log)
# ============================================================================
@@ -108,6 +109,11 @@ class AsyncTUI:
def __init__(self, config: Optional[AsyncTUIConfig] = None):
self.config = config or AsyncTUIConfig()
# Initialize debug logging if enabled
if self.config.debug:
_init_debug_logging()
self.messages: List[ChatMessage] = []
self._running = False
self._agent = None
@@ -183,21 +189,34 @@ class AsyncTUI:
def _get_agent(self):
"""Lazy-load the agent with tools."""
if self._agent is None:
logger.debug("Creating new agent...")
try:
from praisonaiagents import Agent
# Load interactive tools (read_file, write_file, execute_command, etc.)
tools = self._load_tools()
logger.debug(f"Tools for agent: {len(tools) if tools else 0}")
self._agent = Agent(
name="Praison",
role="AI Assistant",
goal="Help the user with their requests",
instructions="You are Praison AI, a helpful assistant. Be concise and helpful. You have access to tools for file operations, code execution, and web search.",
llm=self.config.model,
tools=tools if tools else None,
)
# Build agent config
agent_config = {
"name": "Praison",
"role": "AI Assistant",
"goal": "Help the user with their requests",
"instructions": "You are Praison AI, a helpful assistant. Be concise and helpful. You have access to tools for file operations, code execution, and web search.",
"llm": self.config.model,
"tools": tools if tools else None,
}
# Enable autonomy for complex task handling when autonomy_mode is on
if self.config.autonomy_mode:
agent_config["autonomy"] = True
logger.debug("Autonomy mode enabled")
logger.debug(f"Agent config: model={self.config.model}, tools={len(tools) if tools else 0}")
self._agent = Agent(**agent_config)
logger.debug("Agent created successfully")
except ImportError as e:
logger.error(f"Failed to import praisonaiagents: {e}")
raise RuntimeError(f"Failed to import praisonaiagents: {e}")
return self._agent
@@ -205,8 +224,10 @@ class AsyncTUI:
async def _start_runtime(self):
"""Start the InteractiveRuntime with ACP/LSP servers."""
if self._runtime_started:
logger.debug("Runtime already started, skipping")
return
logger.debug("Starting runtime...")
try:
from praisonai.cli.features.interactive_runtime import create_runtime
@@ -216,28 +237,36 @@ class AsyncTUI:
acp=True,
approval="auto", # Auto-approve for interactive mode
)
logger.debug("Runtime created")
# Start runtime (this starts ACP/LSP servers)
await self._runtime.start()
self._runtime_started = True
lsp_status = "ready" if self._runtime.lsp_ready else "failed"
acp_status = "ready" if self._runtime.acp_ready else "failed"
logger.debug(f"Runtime started: LSP={lsp_status}, ACP={acp_status}")
# Return status
return {
"lsp": "ready" if self._runtime.lsp_ready else "failed",
"acp": "ready" if self._runtime.acp_ready else "failed",
"lsp": lsp_status,
"acp": acp_status,
"read_only": self._runtime.read_only
}
except ImportError as e:
# Runtime not available, continue without it
logger.debug(f"Runtime import error: {e}")
self._runtime_started = True # Mark as attempted
return {"error": f"Runtime not available: {e}"}
except Exception as e:
logger.error(f"Runtime start error: {e}", exc_info=True)
self._runtime_started = True
return {"error": str(e)}
def _load_tools(self):
"""Load interactive tools for the agent."""
tools = []
logger.debug("Starting tool loading...")
# Try to load all interactive tools (basic + ACP + LSP)
try:
@@ -247,13 +276,15 @@ class AsyncTUI:
workspace=self.config.workspace,
)
if tools:
logger.debug(f"Loaded {len(tools)} tools from interactive_tools: {[t.__name__ for t in tools]}")
return tools
except ImportError:
pass
except Exception:
pass
except ImportError as e:
logger.debug(f"interactive_tools import failed: {e}")
except Exception as e:
logger.debug(f"interactive_tools error: {e}")
# Fallback: try to load basic tools directly from praisonaiagents
logger.debug("Falling back to direct praisonaiagents.tools import...")
try:
from praisonaiagents.tools import (
read_file,
@@ -262,32 +293,37 @@ class AsyncTUI:
execute_command,
)
tools = [read_file, write_file, list_files, execute_command]
except ImportError:
pass
logger.debug(f"Loaded basic tools: {[t.__name__ for t in tools]}")
except ImportError as e:
logger.debug(f"Basic tools import failed: {e}")
# Try to add internet search
try:
from praisonaiagents.tools import internet_search
tools.append(internet_search)
except ImportError:
pass
logger.debug("Added internet_search tool")
except ImportError as e:
logger.debug(f"internet_search import failed: {e}")
logger.debug(f"Final tools loaded: {len(tools)}")
return tools
def _format_output(self) -> str:
"""Format all messages for the output pane."""
lines = []
# Logo (only if no messages yet)
if self.config.show_logo and not self.messages:
# Logo - ALWAYS show at top (like Claude Code, OpenCode)
if self.config.show_logo:
logo = get_logo(self.term_width)
lines.append(logo.strip())
lines.append("")
lines.append(f" v{get_version()} · Model: {self.config.model}")
lines.append("")
lines.append(" Type your message and press Enter. Use /help for commands.")
lines.append(" Use PageUp/PageDown or Ctrl+Up/Down to scroll.")
lines.append("")
# Only show tips if no conversation yet
if not self.messages:
lines.append(" Type your message and press Enter. Use /help for commands.")
lines.append(" Use PageUp/PageDown or Ctrl+Up/Down to scroll.")
lines.append("")
# Messages
for msg in self.messages:
@@ -303,6 +339,9 @@ class AsyncTUI:
elif msg.role == "status":
lines.append(f" ⏳ {msg.content}")
lines.append("")
elif msg.role == "tool":
lines.append(f" ⚙ {msg.content}")
lines.append("")
# Processing indicator
if self._processing:
@@ -352,6 +391,10 @@ class AsyncTUI:
/import <file> Import conversation from file
/cost Show token usage and cost
/status Show ACP/LSP runtime status
/auto Toggle autonomy mode (auto-delegate complex tasks)
/debug Toggle debug logging to ~/.praisonai/async_tui_debug.log
/plan <task> Create a step-by-step plan for a task
/handoff <type> <task> Delegate to specialized agent (code/research/review/docs)
/compact Toggle compact output mode
/multiline Toggle multiline input mode
/files List workspace files for @ mentions
@@ -366,7 +409,8 @@ Keyboard Shortcuts:
Tips:
Use @filename to include file contents in your prompt
Type multiple prompts while AI is thinking (queued)"""
Type multiple prompts while AI is thinking (queued)
Use --debug flag or /debug command to enable debug logging"""
self.messages.append(ChatMessage(role="system", content=help_text))
return True
@@ -548,6 +592,102 @@ Tips:
self.messages.append(ChatMessage(role="system", content="Runtime not initialized."))
return True
elif cmd == "auto":
# Toggle autonomy mode for complex task delegation
self.config.autonomy_mode = not self.config.autonomy_mode
mode = "enabled" if self.config.autonomy_mode else "disabled"
self.messages.append(ChatMessage(
role="system",
content=f"Autonomy mode {mode}. Agent will {'auto-delegate complex tasks' if self.config.autonomy_mode else 'handle tasks directly'}."
))
# Recreate agent with autonomy setting
self._agent = None
return True
elif cmd == "debug":
# Toggle debug logging mode
self.config.debug = not self.config.debug
if self.config.debug:
_init_debug_logging()
logger.debug("Debug mode enabled via /debug command")
mode = "enabled" if self.config.debug else "disabled"
log_file = os.path.expanduser("~/.praisonai/async_tui_debug.log")
self.messages.append(ChatMessage(
role="system",
content=f"Debug mode {mode}. Logs written to: {log_file}"
))
return True
elif cmd == "plan":
# Planning mode - create a plan for a complex task
if not args:
self.messages.append(ChatMessage(
role="system",
content="Usage: /plan <task description>\nCreates a step-by-step plan before execution."
))
else:
# Execute with planning enabled
self.messages.append(ChatMessage(role="system", content=f"📋 Creating plan for: {args}"))
self._update_output()
# Use planning agent to create plan
planning_prompt = f"""Create a detailed step-by-step plan for the following task.
Do NOT execute anything yet, just analyze and plan.
Task: {args}
Provide:
1. Analysis of what needs to be done
2. Step-by-step plan with clear actions
3. Potential risks or considerations
4. Estimated complexity (simple/medium/complex)"""
self._queue_or_execute(planning_prompt)
return True
elif cmd == "handoff":
# Handoff to a specialized sub-agent
if not args:
self.messages.append(ChatMessage(
role="system",
content="""Usage: /handoff <agent_type> <task>
Available agent types:
- code: Code analysis and generation
- research: Web research and information gathering
- review: Code review and quality analysis
- docs: Documentation generation
Example: /handoff code "refactor the auth module" """
))
else:
parts = args.split(maxsplit=1)
if len(parts) < 2:
self.messages.append(ChatMessage(role="system", content="Please specify both agent type and task."))
else:
agent_type, task = parts
agent_type = agent_type.lower()
# Map agent types to specialized prompts
agent_prompts = {
"code": f"You are a specialized code agent. Focus on code analysis, generation, and refactoring. Task: {task}",
"research": f"You are a research agent. Focus on gathering information and providing comprehensive analysis. Task: {task}",
"review": f"You are a code review agent. Focus on identifying issues, suggesting improvements, and ensuring quality. Task: {task}",
"docs": f"You are a documentation agent. Focus on creating clear, comprehensive documentation. Task: {task}",
}
if agent_type in agent_prompts:
self.messages.append(ChatMessage(
role="system",
content=f"🔄 Handing off to {agent_type} agent..."
))
self._update_output()
self._queue_or_execute(agent_prompts[agent_type])
else:
self.messages.append(ChatMessage(
role="system",
content=f"Unknown agent type: {agent_type}. Use: code, research, review, or docs"
))
return True
else:
self.messages.append(ChatMessage(role="system", content=f"Unknown command: /{cmd}. Use /help for available commands."))
return True
@@ -558,14 +698,20 @@ Tips:
import io
import asyncio
logger.debug(f"Executing prompt: {prompt[:100]}...")
try:
agent = self._get_agent()
logger.debug(f"Agent loaded: {agent.name if hasattr(agent, 'name') else 'unnamed'}")
logger.debug(f"Agent tools: {[t.__name__ if hasattr(t, '__name__') else str(t) for t in (agent.tools or [])]}")
# Suppress agent's Rich output
# Capture agent's Rich output for logging
old_stdout = sys.stdout
old_stderr = sys.stderr
sys.stdout = io.StringIO()
sys.stderr = io.StringIO()
captured_stdout = io.StringIO()
captured_stderr = io.StringIO()
sys.stdout = captured_stdout
sys.stderr = captured_stderr
try:
# Try async execution first for better non-blocking behavior
@@ -575,21 +721,62 @@ Tips:
asyncio.set_event_loop(loop)
response = loop.run_until_complete(agent.astart(prompt))
loop.close()
except Exception:
logger.debug("Used async execution (astart)")
except Exception as e:
logger.debug(f"Async execution failed: {e}, falling back to sync")
# Fallback to sync
response = agent.start(prompt)
else:
response = agent.start(prompt)
logger.debug("Used sync execution (start)")
finally:
sys.stdout = old_stdout
sys.stderr = old_stderr
# Log captured output
stdout_content = captured_stdout.getvalue()
stderr_content = captured_stderr.getvalue()
if stdout_content:
logger.debug(f"Agent stdout: {stdout_content[:500]}")
if stderr_content:
logger.debug(f"Agent stderr: {stderr_content[:500]}")
logger.debug(f"Response received: {str(response)[:200] if response else 'None'}")
return str(response) if response else None
except Exception as e:
logger.error(f"Error executing prompt: {e}", exc_info=True)
return f"Error: {e}"
def _execute_in_background(self, prompt: str):
"""Execute prompt in background thread (non-blocking)."""
# Track tool calls for visibility
tool_calls = []
def tool_call_callback(message):
"""Callback triggered when a tool is called."""
if "Calling function:" in message:
parts = message.split("Calling function:")
if len(parts) > 1:
tool_name = parts[1].strip()
if tool_name and tool_name not in tool_calls:
tool_calls.append(tool_name)
self._status_text = f"⚙ Using {tool_name}..."
self._update_output()
elif "Function " in message and " returned:" in message:
self._status_text = "Processing result..."
self._update_output()
# Register callback for tool visibility
_sync_display_callbacks = None
_register_display_callback = None
try:
from praisonaiagents import register_display_callback, sync_display_callbacks
_sync_display_callbacks = sync_display_callbacks
_register_display_callback = register_display_callback
_register_display_callback('tool_call', tool_call_callback)
except ImportError:
pass
def run():
self._processing = True
self._status_text = "Praison AI is thinking..."
@@ -633,10 +820,19 @@ Tips:
llm_thread.join()
# Cleanup callback
if _sync_display_callbacks is not None and 'tool_call' in _sync_display_callbacks:
del _sync_display_callbacks['tool_call']
# Done processing current prompt
self._processing = False
self._status_text = ""
# Show tool calls summary if any were made
if tool_calls:
tools_used = ", ".join(tool_calls)
self.messages.append(ChatMessage(role="system", content=f"Tools used: {tools_used}"))
if error[0]:
self.messages.append(ChatMessage(role="assistant", content=f"Error: {error[0]}"))
self._conversation_history.append({"role": "assistant", "content": f"Error: {error[0]}"})
@@ -680,18 +876,10 @@ Tips:
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
runtime_status = loop.run_until_complete(self._start_runtime())
loop.run_until_complete(self._start_runtime())
loop.close()
# Show runtime status
if runtime_status and "error" not in runtime_status:
lsp = runtime_status.get("lsp", "unknown")
acp = runtime_status.get("acp", "unknown")
if lsp == "ready" or acp == "ready":
self.messages.append(ChatMessage(
role="system",
content=f"Runtime started: LSP={lsp}, ACP={acp}"
))
# Runtime status logged to debug file only (not shown in UI)
# Tools are available silently when runtime is ready
except Exception as e:
# Continue without runtime
self.messages.append(ChatMessage(
@@ -759,8 +947,8 @@ Tips:
# Create completer for commands and files
commands = ["help", "exit", "quit", "clear", "new", "model", "session", "sessions",
"continue", "history", "export", "import", "cost", "status", "compact",
"multiline", "files", "queue"]
"continue", "history", "export", "import", "cost", "status", "auto",
"debug", "plan", "handoff", "compact", "multiline", "files", "queue"]
workspace_files = self._workspace_files
class PraisonCompleter(Completer):
@@ -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"
@@ -19,44 +19,10 @@ from typing import List, Optional
from datetime import datetime
# ============================================================================
# ASCII Art Logo - "Praison AI" branding
# Branding - Import from unified source
# ============================================================================
LOGO = r"""
██████╗ ██████╗ █████╗ ██╗███████╗ ██████╗ ███╗ ██╗ █████╗ ██╗
██╔══██╗██╔══██╗██╔══██╗██║██╔════╝██╔═══██╗████╗ ██║ ██╔══██╗██║
██████╔╝██████╔╝███████║██║███████╗██║ ██║██╔██╗ ██║ ███████║██║
██╔═══╝ ██╔══██╗██╔══██║██║╚════██║██║ ██║██║╚██╗██║ ██╔══██║██║
██║ ██║ ██║██║ ██║██║███████║╚██████╔╝██║ ╚████║ ██║ ██║██║
╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚═╝╚══════╝ ╚═════╝ ╚═╝ ╚═══╝ ╚═╝ ╚═╝╚═╝
"""
LOGO_SMALL = r"""
╔═╗┬─┐┌─┐┬┌─┐┌─┐┌┐┌ ╔═╗╦
╠═╝├┬┘├─┤│└─┐│ ││││ ╠═╣║
╩ ┴└─┴ ┴┴└─┘└─┘┘└┘ ╩ ╩╩
"""
LOGO_MINIMAL = "▶ Praison AI"
def get_logo(width: int = 80) -> str:
"""Get appropriate logo based on terminal width."""
if width >= 75:
return LOGO
elif width >= 40:
return LOGO_SMALL
else:
return LOGO_MINIMAL
def get_version() -> str:
"""Get PraisonAI version."""
try:
from praisonai import __version__
return __version__
except Exception:
return "1.0.0"
from praisonai.cli.branding import get_logo, get_version
# ============================================================================
@@ -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.7 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.7"
__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",
+261 -13
View File
@@ -58,30 +58,30 @@ class TestGetLogo:
def test_logo_wide_terminal(self):
"""Test logo for wide terminal."""
from praisonai.cli.interactive.async_tui import get_logo, LOGO
from praisonai.cli.branding import get_logo, LOGO_LARGE
logo = get_logo(100)
assert logo == LOGO
assert logo == LOGO_LARGE
def test_logo_medium_terminal(self):
"""Test logo for medium terminal."""
from praisonai.cli.interactive.async_tui import get_logo, LOGO_SMALL
from praisonai.cli.branding import get_logo, LOGO_MEDIUM
logo = get_logo(50)
assert logo == LOGO_SMALL
assert logo == LOGO_MEDIUM
def test_logo_narrow_terminal(self):
"""Test logo for narrow terminal."""
from praisonai.cli.interactive.async_tui import get_logo, LOGO_MINIMAL
from praisonai.cli.branding import get_logo, LOGO_SMALL
logo = get_logo(30)
assert logo == LOGO_MINIMAL
assert logo == LOGO_SMALL
def test_logo_contains_praison_ai(self):
"""Test that logo contains 'Praison AI' branding."""
from praisonai.cli.interactive.async_tui import LOGO_MINIMAL
from praisonai.cli.branding import LOGO_SMALL
assert "Praison AI" in LOGO_MINIMAL
assert "Praison AI" in LOGO_SMALL
class TestAsyncTUI:
@@ -246,18 +246,18 @@ class TestASCIILogos:
def test_logo_exists(self):
"""Test LOGO constant exists and has content."""
from praisonai.cli.interactive.async_tui import LOGO
assert len(LOGO) > 0
assert "██" in LOGO
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.interactive.async_tui import LOGO_SMALL
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.interactive.async_tui import LOGO_MINIMAL
from praisonai.cli.branding import LOGO_MINIMAL
assert "Praison AI" in LOGO_MINIMAL
@@ -598,3 +598,251 @@ class TestRuntimeIntegration:
# 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
+13 -13
View File
@@ -58,30 +58,30 @@ class TestGetLogo:
def test_logo_wide_terminal(self):
"""Test logo for wide terminal."""
from praisonai.cli.interactive.split_tui import get_logo, LOGO
from praisonai.cli.branding import get_logo, LOGO_LARGE
logo = get_logo(100)
assert logo == LOGO
assert logo == LOGO_LARGE
def test_logo_medium_terminal(self):
"""Test logo for medium terminal."""
from praisonai.cli.interactive.split_tui import get_logo, LOGO_SMALL
from praisonai.cli.branding import get_logo, LOGO_MEDIUM
logo = get_logo(50)
assert logo == LOGO_SMALL
assert logo == LOGO_MEDIUM
def test_logo_narrow_terminal(self):
"""Test logo for narrow terminal."""
from praisonai.cli.interactive.split_tui import get_logo, LOGO_MINIMAL
from praisonai.cli.branding import get_logo, LOGO_SMALL
logo = get_logo(30)
assert logo == LOGO_MINIMAL
assert logo == LOGO_SMALL
def test_logo_contains_praison_ai(self):
"""Test that logo contains 'Praison AI' branding."""
from praisonai.cli.interactive.split_tui import LOGO_MINIMAL
from praisonai.cli.branding import LOGO_SMALL
assert "Praison AI" in LOGO_MINIMAL
assert "Praison AI" in LOGO_SMALL
class TestSplitTUI:
@@ -233,18 +233,18 @@ class TestASCIILogos:
def test_logo_exists(self):
"""Test LOGO constant exists and has content."""
from praisonai.cli.interactive.split_tui import LOGO
assert len(LOGO) > 0
assert "██" in LOGO # Contains block characters
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.interactive.split_tui import LOGO_SMALL
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.interactive.split_tui import LOGO_MINIMAL
from praisonai.cli.branding import LOGO_MINIMAL
assert "Praison AI" in LOGO_MINIMAL
+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]]