Compare commits

..
Author SHA1 Message Date
praisonai-triage-agent[bot]andMervin Praison 251470489f fix: address valkey integration issues from code review
- Fix JSON encoding consistency in ValkeyStateStore to prevent type confusion
- Add TTL guards for ExpirySet/ExpiryType to prevent cryptic RuntimeError
- Add warning for score threshold limitation (cosine distance only)
- Fix metadata persistence in KnowledgeDocument (now properly round-tripped)
- Improve exception handling - only catch specific 'index not found' errors
- Fix URL parameter handling in registry factories (reject with clear error)
- Add input validation for vector dimensions and k parameter

Co-authored-by: Mervin Praison <[email protected]>
2026-05-20 17:45:00 +00:00
174 changed files with 2525 additions and 10202 deletions

No files matched your search

@@ -1,158 +0,0 @@
name: Cross-Repo Integration Tests
on:
push:
branches: [ main ]
paths:
- 'src/praisonai/praisonai/integration/**'
- 'src/praisonai/tests/integration/test_aiui_*'
- 'src/praisonai/pyproject.toml'
pull_request:
branches: [ main ]
paths:
- 'src/praisonai/praisonai/integration/**'
- 'src/praisonai/tests/integration/test_aiui_*'
- 'src/praisonai/pyproject.toml'
workflow_dispatch:
jobs:
cross-repo-integration:
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
- name: Checkout PraisonAI
uses: actions/checkout@v4
with:
path: PraisonAI
ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
- name: Checkout PraisonAIUI
uses: actions/checkout@v4
with:
repository: MervinPraison/PraisonAIUI
path: PraisonAIUI
ref: main
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
# Install PraisonAI packages in development mode
cd PraisonAI/src/praisonai-agents
pip install -e .
cd ../praisonai
pip install -e ".[ui,dev]"
# Install PraisonAIUI in development mode
cd ../../../PraisonAIUI
pip install -e .
- name: Verify versions
run: |
python -c "import praisonaiagents; print(f'praisonaiagents: {praisonaiagents.__version__}')"
python -c "import praisonai; print(f'praisonai: {praisonai.__version__}')"
python -c "import praisonaiui; print(f'praisonaiui: {praisonaiui.__version__}')"
- name: Run PraisonAI integration tests
run: |
cd PraisonAI/src/praisonai
python -m pytest tests/integration/test_aiui_* -v --timeout=300
- name: Run PraisonAIUI integration tests
run: |
cd PraisonAIUI
if [ -f tests/integration/test_agentic_roundtrip.py ]; then
python -m pytest tests/integration/test_agentic_roundtrip.py -v --timeout=300
fi
if [ -f tests/test_feature_sdk_backends.py ]; then
python -m pytest tests/test_feature_sdk_backends.py -v --timeout=300
fi
- name: Test Pattern C CLI
run: |
cd PraisonAI/src/praisonai
# Test that ui-gateway command exists
python -m praisonai serve --help | grep ui-gateway
- name: Test public API exports
run: |
cd PraisonAI/src/praisonai
python -c "from praisonai import run_integrated_gateway; print('✓ run_integrated_gateway imported')"
python -c "from praisonai import configure_host; print('✓ configure_host imported')"
- name: Integration smoke test
env:
PRAISONAI_TEST_MODE: "1"
PYTHONPATH: ${{ github.workspace }}/PraisonAI/src/praisonai:${{ github.workspace }}/PraisonAI/src/praisonai-agents:${{ github.workspace }}/PraisonAIUI
run: |
cd PraisonAI/src/praisonai
python -c "
import sys
sys.path.insert(0, '.')
from praisonai.integration.host_app import configure_host
from praisonai import run_integrated_gateway
print('✓ Integration imports successful')
# Test configure_host with new parameters
try:
configure_host(
style='dashboard',
context_paths=['AGENTS.md'],
title='Test App'
)
print('✓ configure_host with new parameters works')
except Exception as e:
print(f'× configure_host failed: {e}')
sys.exit(1)
"
# Optional job with API key for agentic tests (if secret is available)
agentic-integration:
runs-on: ubuntu-latest
timeout-minutes: 30
if: ${{ vars.RUN_AGENTIC_TESTS == 'true' }}
needs: cross-repo-integration
steps:
- name: Checkout PraisonAI
uses: actions/checkout@v4
with:
path: PraisonAI
ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
- name: Checkout PraisonAIUI
uses: actions/checkout@v4
with:
repository: MervinPraison/PraisonAIUI
path: PraisonAIUI
ref: main
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
cd PraisonAI/src/praisonai-agents
pip install -e .
cd ../praisonai
pip install -e ".[ui,dev]"
cd ../../../PraisonAIUI
pip install -e .
- name: Run agentic integration test
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
cd PraisonAI/src/praisonai
if [ -n "$OPENAI_API_KEY" ]; then
python -m pytest tests/integration/test_aiui_host_agentic.py -v --timeout=600
else
echo "OPENAI_API_KEY not available, skipping agentic tests"
fi
+1 -1
View File
@@ -382,7 +382,7 @@ praisonai flow
Open **http://localhost:7861** — use the **Agent** and **Agent Team** components to create sequential or parallel workflows. Connect Chat Input → Agent Team → Chat Output for instant multi-agent pipelines.
> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/agentflow) — visual agent building, component reference, and deployment
> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/flow) — visual agent building, component reference, and deployment
### 8. PraisonAI UI 🤖 (Clean Chat)
+1 -16
View File
@@ -40,7 +40,6 @@ Methods:
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">chat_with_context</a>(message: str, context: 'ContextPack', **kwargs) -> str</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">checkpoints</a>() -> Optional[bool]</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">checkpoints</a>(value: Optional[bool]) -> None</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">clone_for_channel</a>() -> 'Agent'</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">close</a>() -> None</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">console</a>() -> Optional[Any]</code>
* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">context_manager</a>() -> Optional[Any]</code>
@@ -172,7 +171,6 @@ Methods:
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">safe_run</a>(**kwargs) -> ToolResult</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate</a>() -> bool</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_class</a>() -> bool</code>
* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_schema_roundtrip</a>() -> bool</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">__call__</a>(*args, **kwargs) -> Any</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">check_availability</a>() -> tuple[bool, str]</code>
* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">injected_params</a>() -> Dict[str, Any]</code>
@@ -479,10 +477,8 @@ Methods:
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">edit_skill</a>(name: str, content: str) -> dict</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_allowed_tools</a>(name: str) -> List[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_available_skills</a>() -> List[SkillMetadata]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_available_skills_by_state</a>(state: SkillState) -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_instructions</a>(name: str) -> Optional[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_skill</a>(name: str) -> Optional[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_skills_diagnostics</a>() -> Dict[str, ValidationResult]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_user_invocable_skills</a>() -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">invoke</a>(name: str, raw_args: str = '', session_id: Optional[str] = None, shell_exec: bool = False) -> Optional[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">load_resources</a>(name: str) -> bool</code>
@@ -491,7 +487,6 @@ Methods:
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">skill_names</a>() -> List[str]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">skills</a>() -> List[LoadedSkill]</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">to_prompt</a>() -> str</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">validate_skill_capabilities</a>(skill_name: str, force_refresh: bool = False) -> ValidationResult</code>
* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">write_skill_file</a>(name: str, file_path: str, file_content: str) -> dict</code>
* <code title="class SkillMetadata">SkillMetadata.<a href="./src/praisonai-agents/praisonaiagents/skills/models.py">from_properties</a>(props: SkillProperties) -> 'SkillMetadata'</code>
* <code title="class SkillProperties">SkillProperties.<a href="./src/praisonai-agents/praisonaiagents/skills/models.py">to_dict</a>() -> dict</code>
@@ -607,13 +602,6 @@ Methods:
* <code title="class ContextManager">ContextManager.<a href="./src/praisonai-agents/praisonaiagents/context/manager.py">set_tool_budget</a>(tool_name: str, max_tokens: int, protected: bool = False) -> None</code>
* <code title="class ContextManager">ContextManager.<a href="./src/praisonai-agents/praisonaiagents/context/manager.py">truncate_tool_output</a>(tool_name: str, output: str) -> str</code>
# UI
Types:
```python
from praisonaiagents import A2UI
```
# Config
Types:
@@ -768,7 +756,7 @@ Methods:
Types:
```python
from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__, arun, run, run_integrated_gateway
from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__
```
# CLI
@@ -873,7 +861,6 @@ Methods:
* <code title="cli">praisonai doctor packaging <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor performance <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor selftest <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor skills <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor tools <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai doctor troubleshoot <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
* <code title="cli">praisonai endpoints list <a href="./src/praisonai/praisonai/cli/commands/endpoints.py">--help</a></code>
@@ -905,7 +892,6 @@ Methods:
* <code title="cli">praisonai gateway send <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway start <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway status <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway stop <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai gateway uninstall <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
* <code title="cli">praisonai github triage <a href="./src/praisonai/praisonai/cli/commands/github.py">--help</a></code>
* <code title="cli">praisonai hooks add <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
@@ -1068,7 +1054,6 @@ Methods:
* <code title="cli">praisonai serve status <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve stop <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve ui <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve ui-gateway <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai serve unified <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
* <code title="cli">praisonai session delete <a href="./src/praisonai/praisonai/cli/commands/session.py">--help</a></code>
* <code title="cli">praisonai session export <a href="./src/praisonai/praisonai/cli/commands/session.py">--help</a></code>
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.6.48" \
"praisonai>=4.6.43" \
"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>=4.6.48" \
"praisonai>=4.6.43" \
"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>=4.6.48" \
"praisonai>=4.6.43" \
"praisonai[ui]" \
"praisonai[crewai]"
-27
View File
@@ -1,27 +0,0 @@
#!/usr/bin/env python3
"""Minimal A2UI example — agent + send_a2ui_messages tool.
Install:
pip install praisonaiagents[a2ui]
Run:
python send_surface.py
For live rendering, use PraisonAIUI with AG-UI (/agui) — see integrate-a2ui-frontend docs.
"""
from praisonaiagents import Agent
from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
agent = Agent(
name="ui-assistant",
instructions=(
"When asked for UI, call send_a2ui_messages with A2UI v0.9 JSON "
"(createSurface on surface id 'main')."
),
tools=[send_a2ui_messages],
)
if __name__ == "__main__":
result = agent.start("Show a simple welcome card on surface main")
print(result)
-222
View File
@@ -1,222 +0,0 @@
# Multi-Channel Bot Deployment Guide
This guide explains how to set up multiple bots on the same platform (e.g., multiple Telegram bots) for different roles using PraisonAI, similar to Hermes-style AI workforces.
## Overview
PraisonAI supports running multiple bots on the same platform, each routing to different agents with specialized knowledge and capabilities. This is useful for:
- **Role-based bots**: `@company_cfo_bot`, `@company_ops_bot`, `@company_content_bot`
- **Department-specific agents**: Each with different knowledge brains and tool access
- **Workflow separation**: Different bots handle different types of requests
## Quick Start with Onboard Wizard
The easiest way to set up multiple channels is using the onboard wizard:
```bash
praisonai onboard
```
The wizard now supports:
1. **Multi-channel setup**: Add multiple bots for the same platform
2. **Role-based naming**: Automatically generates `TELEGRAM_<ROLE>_BOT_TOKEN` environment variables
3. **Agent routing**: Each channel routes to a different agent
### Onboard Wizard Flow
1. **Choose platform**: Start with your first platform (e.g., telegram)
2. **Add more channels**: Wizard asks "Add another bot channel?"
3. **Specify role**: Enter role name (e.g., "cfo", "ops", "content")
4. **Configure tokens**: Each channel gets its own environment variable
5. **Test connections**: Wizard tests each bot separately
## Environment Variable Convention
For multiple bots on the same platform, use this naming convention:
| Channel | Environment Variable | Bot Purpose |
|---------|---------------------|-------------|
| `telegram_cfo` | `TELEGRAM_CFO_BOT_TOKEN` | CFO-related tasks |
| `telegram_ops` | `TELEGRAM_OPS_BOT_TOKEN` | Operations tasks |
| `telegram_content` | `TELEGRAM_CONTENT_BOT_TOKEN` | Content creation |
| `telegram_support` | `TELEGRAM_SUPPORT_BOT_TOKEN` | Customer support |
### Setting Up Tokens
1. **Create bots in @BotFather**:
```text
/newbot
Bot name: Company CFO Bot
Username: @company_cfo_bot
-> Copy token
/newbot
Bot name: Company Ops Bot
Username: @company_ops_bot
-> Copy token
```
2. **Set environment variables**:
```bash
export TELEGRAM_CFO_BOT_TOKEN="123456789:ABC..."
export TELEGRAM_OPS_BOT_TOKEN="987654321:DEF..."
export TELEGRAM_CONTENT_BOT_TOKEN="456789123:GHI..."
```
3. **Or use `~/.praisonai/.env`**:
```env
TELEGRAM_CFO_BOT_TOKEN=123456789:ABC...
TELEGRAM_OPS_BOT_TOKEN=987654321:DEF...
TELEGRAM_CONTENT_BOT_TOKEN=456789123:GHI...
TELEGRAM_ALLOWED_USERS=123456789,987654321
```
## Example Configuration
See `multi-telegram-hermes-workforce.yaml` for a complete example.
### Key Features
1. **Multiple Channels**:
```yaml
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
routes:
default: cfo
telegram_ops:
platform: telegram
token: ${TELEGRAM_OPS_BOT_TOKEN}
routes:
default: ops
```
2. **Specialized Agents**:
```yaml
agents:
cfo:
instructions: "You are a CFO agent. Help with financial analysis..."
model: gpt-4o-mini
ops:
instructions: "You are an ops agent. Help with system monitoring..."
model: gpt-4o-mini
```
## Validation with Doctor
Use the doctor command to validate your multi-channel setup:
```bash
praisonai doctor
```
The doctor checks for:
- ✅ **Unique tokens**: Each channel uses a different bot token
- ✅ **Naming convention**: Follows `PLATFORM_ROLE_BOT_TOKEN` pattern
- ✅ **Missing tokens**: All referenced environment variables are set
- ❌ **Duplicate tokens**: Same token used by multiple channels (fails)
## Deployment
### Start Gateway
```bash
# Start the multi-bot gateway
praisonai gateway start --config multi-telegram-hermes-workforce.yaml
# Or install as daemon
praisonai onboard # Wizard installs daemon automatically
praisonai gateway status
```
### Verify Setup
1. **Check health**: `curl http://127.0.0.1:8765/health`
2. **View dashboard**: `praisonai claw` → http://127.0.0.1:8082
3. **Test bots**: Send messages to each bot to verify routing
## Common Patterns
### Basic Multi-Role Setup
```yaml
# Single platform, multiple roles
channels:
telegram_support:
token: ${TELEGRAM_SUPPORT_BOT_TOKEN}
routes:
default: support_agent
telegram_sales:
token: ${TELEGRAM_SALES_BOT_TOKEN}
routes:
default: sales_agent
```
### Cross-Platform Workforce
```yaml
# Multiple platforms, multiple roles
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
routes:
default: cfo
discord_ops:
platform: discord
token: ${DISCORD_OPS_BOT_TOKEN}
routes:
default: ops
slack_content:
platform: slack
token: ${SLACK_CONTENT_BOT_TOKEN}
routes:
default: content
```
## Troubleshooting
### Common Issues
1. **"Token is used by multiple channels"**
- Each bot needs its own unique token from @BotFather
- Check environment variables for duplicates
2. **"Channel skipped - no token"**
- Set the environment variable: `export TELEGRAM_CFO_BOT_TOKEN="..."`
- Or update `~/.praisonai/.env`
3. **"Bot not responding"**
- Verify token is correct: `praisonai doctor`
- Check bot is not already running elsewhere
- Ensure allowlist includes your user ID
### Debugging Commands
```bash
# Check configuration
praisonai doctor
# View logs
praisonai gateway logs
# Test connection
curl -H "Authorization: Bearer $GATEWAY_AUTH_TOKEN" \
http://127.0.0.1:8765/info
```
## Migration from Single-Bot Setup
If you currently have a single bot setup, you can add more channels:
1. **Run onboard wizard**: `praisonai onboard`
2. **Add new channels**: Wizard will preserve existing setup
3. **Update configuration**: New channels added to existing `bot.yaml`
4. **Restart gateway**: `praisonai gateway restart`
The wizard preserves backward compatibility with single-bot configurations.
-19
View File
@@ -1,19 +0,0 @@
# A2UI canvas agent — YAML example
#
# Usage:
# pip install praisonaiagents[a2ui]
# praisonai agent run --file a2ui-canvas.yaml --prompt "Show a welcome card"
#
# For live A2UI rendering, use PraisonAIUI (AG-UI / surfaces):
# pip install praisonaiui
# See examples/python/a2ui/send_surface.py and PraisonAIUI example 29.
agent:
name: "ui-assistant"
instructions: |
You are a helpful assistant. When the user asks for UI elements
(buttons, cards, forms), call send_a2ui_messages with valid A2UI v0.9 JSON.
Use surface id "main" unless told otherwise.
llm: gpt-4o-mini
tools:
- send_a2ui_messages
@@ -1,66 +0,0 @@
# Multi-Telegram Channel Hermes Workforce Example
# Generated by `praisonai onboard` with multi-channel support
# Start with: praisonai gateway start --config multi-telegram-hermes-workforce.yaml
gateway:
host: "127.0.0.1"
port: 8765
# Legacy single-agent block for backward compatibility
agent:
name: "assistant"
instructions: "You are a helpful AI assistant. You can search the web, manage schedules, and remember past conversations. Use your tools proactively."
model: "gpt-4o-mini"
# Multi-agent workforce with different specialized roles
agents:
cfo:
instructions: "You are a CFO agent. You are a helpful AI assistant. Use your tools proactively to help with CFO-related tasks like financial analysis, budget planning, and cost optimization."
model: gpt-4o-mini
memory: false
ops:
instructions: "You are an ops agent. You are a helpful AI assistant. Use your tools proactively to help with ops-related tasks like system monitoring, deployment, and infrastructure management."
model: gpt-4o-mini
memory: false
content:
instructions: "You are a content agent. You are a helpful AI assistant. Use your tools proactively to help with content-related tasks like writing, editing, and content strategy."
model: gpt-4o-mini
memory: false
# Multiple Telegram channels - each with unique tokens and routing
channels:
telegram_cfo:
platform: telegram
token: ${TELEGRAM_CFO_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: cfo
telegram_ops:
platform: telegram
token: ${TELEGRAM_OPS_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: ops
telegram_content:
platform: telegram
token: ${TELEGRAM_CONTENT_BOT_TOKEN}
group_policy: mention_only
allowed_users: ${TELEGRAM_ALLOWED_USERS}
# auto_approve_tools: true
# Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
routes:
default: content
# Legacy routing block
routing:
default: assistant
@@ -343,7 +343,6 @@ _LAZY_IMPORTS = {
# UI (optional)
'AGUI': ('praisonaiagents.ui.agui', 'AGUI'),
'A2A': ('praisonaiagents.ui.a2a', 'A2A'),
'A2UI': ('praisonaiagents.ui.a2ui', 'A2UI'),
# Feature configs
'MemoryConfig': ('praisonaiagents.config.feature_configs', 'MemoryConfig'),
@@ -815,6 +814,6 @@ def __dir__():
# - Workflow, Task, Route, Parallel, Loop, Repeat, Pipeline, etc.
# - MCP, FlowDisplay, track_workflow, FastContext, etc.
# - Plan, PlanStep, TodoList, PlanningAgent, ApprovalCallback, etc.
# - RAG, RAGConfig, RAGResult, AGUI, A2A, A2UI, etc.
# - RAG, RAGConfig, RAGResult, AGUI, A2A, etc.
# - All telemetry, display, and utility functions
# ============================================================================
@@ -1952,129 +1952,6 @@ Your Goal: {self.goal}
# Sandbox configuration - initialize SandboxMixin
super().__init__(sandbox=sandbox)
def __deepcopy__(self, memo: dict) -> "Agent":
"""Custom deepcopy that creates fresh threading primitives.
threading.RLock (self.__cache_lock) and threading.Lock (self._cost_lock)
cannot be deep-copied on CPython < 3.13. This hook deep-copies every
other attribute normally and replaces the locks with new instances so
that copy.deepcopy(agent) works in any Python version.
"""
import copy
cls = self.__class__
result = cls.__new__(cls)
memo[id(self)] = result
for k, v in self.__dict__.items():
if k in ("_Agent__cache_lock",):
object.__setattr__(result, k, threading.RLock())
elif k == "_cost_lock":
object.__setattr__(result, k, threading.Lock())
else:
object.__setattr__(result, k, copy.deepcopy(v, memo))
return result
def clone_for_channel(self) -> "Agent":
"""Return a fully independent copy of this agent for a gateway channel.
This method safely clones an agent for use in multi-channel scenarios
by avoiding deepcopy issues with tools/handoffs and creating fresh
instances of mutable objects like interrupt_controller.
Returns:
Agent: A new Agent instance with copied configuration but fresh
locks, interrupt controller, and shallow-copied tools.
"""
import copy
from ..tools.base import BaseTool
# Build kwargs for new Agent instance, copying core attributes
clone_kwargs = {
# Core identity
'name': self.name,
'role': self.role,
'goal': self.goal,
'backstory': self.backstory,
'instructions': self.instructions,
# LLM configuration
'llm': self.llm,
'base_url': getattr(self, 'base_url', None),
'api_key': getattr(self, 'api_key', None),
'auth': getattr(self, 'auth', None),
# Shallow copy tools to avoid deepcopy issues with nested objects
'tools': list(self.tools) if self.tools else None,
# Skip handoffs entirely - they shouldn't be shared across channels
# and can contain nested Agent instances that cause RLock issues
'handoffs': None,
# Feature configurations - check for actual stored config objects
'memory': getattr(self, '_memory_config', None),
'knowledge': getattr(self, '_knowledge_config', None),
'planning': getattr(self, '_planning_config', None),
'reflection': getattr(self, '_reflection_config', None),
'guardrails': getattr(self, '_guardrails_config', None),
'web': getattr(self, '_web_config', None),
'context': getattr(self, '_context_config', None),
'autonomy': getattr(self, '_autonomy_config', None),
'output': getattr(self, '_output_config', None),
'execution': getattr(self, '_execution_config', None),
'templates': getattr(self, '_template_config', None),
'caching': getattr(self, '_caching_config', None),
'hooks': getattr(self, '_hooks_config', None),
'skills': getattr(self, '_skills_config', None),
'approval': getattr(self, '_approval_config', None),
'learn': getattr(self, '_learn_config', None),
# Tool configuration
'tool_timeout': getattr(self, '_tool_timeout', None),
'parallel_tool_calls': getattr(self, 'parallel_tool_calls', False),
# CLI backend
'cli_backend': getattr(self, '_cli_backend', None),
# Create fresh interrupt controller to avoid shared state
'interrupt_controller': None, # Let new instance create its own
# Sandbox config
'sandbox': getattr(self, '_sandbox_config', None),
}
# Handle deprecated parameters for backward compatibility
import warnings
# Check for deprecated standalone attributes and emit warnings
deprecated_attrs = [
('allow_code_execution', 'execution'),
('code_execution_mode', 'execution'),
('auto_save', 'memory'),
('rate_limiter', 'execution'),
('allow_delegation', 'handoffs'),
('verification_hooks', 'autonomy')
]
for old_attr, new_param in deprecated_attrs:
if hasattr(self, old_attr):
value = getattr(self, old_attr)
if value is not None and value != False: # Skip None/False defaults
warnings.warn(
f"Deprecated attribute '{old_attr}' found in agent clone. "
f"Use '{new_param}=' parameter instead.",
DeprecationWarning,
stacklevel=2
)
# For basic compatibility, include in clone_kwargs if not already set
if old_attr == 'allow_code_execution' and clone_kwargs['execution'] is None:
from ..config.execution import ExecutionConfig
clone_kwargs['execution'] = ExecutionConfig(code_execution=value)
elif old_attr == 'auto_save' and clone_kwargs['memory'] is None:
from ..config.memory import MemoryConfig
clone_kwargs['memory'] = MemoryConfig(auto_save=value)
# Create new Agent instance
return self.__class__(**{k: v for k, v in clone_kwargs.items() if v is not None})
@property
def _telemetry(self):
"""Lazy-loaded telemetry instance for performance."""
@@ -4739,57 +4616,60 @@ Answer:"""
error=f"Agent guardrail validation error: {str(e)}"
)
def _validate_with_guardrail(self, response_text):
"""Validate response with guardrail. Returns (success, result, error)."""
def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic.
Args:
response_text: The response to validate
prompt: Original prompt for regeneration if needed
temperature: Temperature for regeneration
tools: Tools for regeneration
Returns:
str: The validated response text or None if validation fails after retries
"""
if not self._guardrail_fn:
return True, response_text, None
return response_text
from ..main import TaskOutput
task_output = TaskOutput(
description="Agent response output",
raw=response_text,
agent=self.name
)
guardrail_result = self._process_guardrail(task_output)
if guardrail_result.success:
# Return the potentially modified result
if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
return True, guardrail_result.result.raw, None
elif guardrail_result.result:
return True, str(guardrail_result.result), None
else:
return True, response_text, None
else:
return False, None, guardrail_result.error
def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic (sync version)."""
retry_count = 0
current_response = response_text
while retry_count <= self.max_guardrail_retries:
success, result, error = self._validate_with_guardrail(current_response)
# Create TaskOutput object
task_output = TaskOutput(
description="Agent response output",
raw=current_response,
agent=self.name
)
if success:
# Process guardrail
guardrail_result = self._process_guardrail(task_output)
if guardrail_result.success:
logging.info(f"Agent {self.name}: Guardrail validation passed")
return result
# Return the potentially modified result
if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
return guardrail_result.result.raw
elif guardrail_result.result:
return str(guardrail_result.result)
else:
return current_response
# Guardrail failed
if retry_count >= self.max_guardrail_retries:
raise Exception(
f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
f"Last error: {error}"
f"Last error: {guardrail_result.error}"
)
retry_count += 1
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {guardrail_result.error}")
# Regenerate response for retry
try:
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {guardrail_result.error}. Please provide an improved response."
response = self._chat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
if response and response.choices:
content = response.choices[0].message.content
@@ -4798,46 +4678,10 @@ Answer:"""
raise Exception("Failed to generate retry response")
except Exception as e:
logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
return current_response
async def _aapply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
"""Apply guardrail validation with retry logic (async version)."""
retry_count = 0
current_response = response_text
while retry_count <= self.max_guardrail_retries:
success, result, error = self._validate_with_guardrail(current_response)
if success:
logging.info(f"Agent {self.name}: Guardrail validation passed")
return result
# Guardrail failed
if retry_count >= self.max_guardrail_retries:
# If we can't regenerate, fail the guardrail
raise Exception(
f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
f"Last error: {error}"
f"Agent {self.name} guardrail retry failed: {e}"
)
retry_count += 1
logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
# Regenerate response for retry (async version)
try:
retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
response = await self._execute_unified_achat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
if response and hasattr(response, 'choices') and response.choices:
content = response.choices[0].message.content
current_response = content.strip() if content else ""
elif isinstance(response, str):
current_response = response.strip()
else:
raise Exception("Failed to generate retry response")
except Exception as e:
logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
return current_response
@@ -2079,7 +2079,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Apply guardrail validation for custom LLM response
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
validated_response = self._apply_guardrail_with_retry(response_text, prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(normalized_content, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(prompt, validated_response, start_time)
@@ -2187,216 +2187,22 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
task_id=task_id,
response_format=response_format
)
# Process response - mirror sync _chat_impl behavior (lines ~1544-1602)
if not response:
# Rollback chat history on response failure
self._truncate_chat_history(chat_history_length)
return None
# Extract response content using the same method as sync
response_text = self._extract_llm_response_content(response)
if isinstance(response_text, str):
response_text = response_text.strip()
# Handle output_json or output_pydantic if specified
if output_json or output_pydantic:
# Add to chat history and return raw response
# User message already added before LLM call via _build_messages
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
# Apply guardrail validation even for JSON output
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for JSON output: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# For regular responses (no self-reflection)
if not self.self_reflect:
# User message already added before LLM call via _build_messages
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB (non-reflect path)
self._persist_message("assistant", response_text)
if self.verbose:
logging.debug(f"Agent {self.name} final response: {response_text}")
# Return only reasoning content if reasoning_steps is True
if reasoning_steps and hasattr(response, 'choices') and response.choices and hasattr(response.choices[0].message, 'reasoning_content') and response.choices[0].message.reasoning_content:
# Apply guardrail to reasoning content
try:
validated_reasoning = await self._aapply_guardrail_with_retry(response.choices[0].message.reasoning_content, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_reasoning, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_reasoning, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed for reasoning content: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
else:
# Apply guardrail to regular response content
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# If self-reflection is enabled, implement reflection logic
if self.self_reflect:
# Implement async self-reflection similar to sync path
reflection_prompt = f"""
Reflect on your previous response: '{response_text}'.
{self.reflect_prompt if self.reflect_prompt else "Identify any flaws, improvements, or actions."}
Provide a "satisfactory" status ('yes' or 'no').
Output MUST be JSON with 'reflection' and 'satisfactory'.
"""
# Add reflection prompt to messages
reflection_messages = messages + [
{"role": "assistant", "content": response_text},
{"role": "user", "content": reflection_prompt}
]
reflection_count = 0
while True:
try:
# Check if OpenAI client is available for self-reflection
if self._openai_client is None:
# For custom LLMs, self-reflection with structured output is not supported
if self.verbose:
_get_display_functions()['display_self_reflection'](f"Agent {self.name}: Self-reflection with structured output is not supported for custom LLM providers. Skipping reflection.", console=self.console)
# Return the original response without reflection
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
reflection_response = await self._openai_client.async_client.beta.chat.completions.parse(
model=self.reflect_llm if self.reflect_llm else self.llm,
messages=reflection_messages,
temperature=temperature,
response_format=_get_display_functions()['ReflectionOutput']
)
reflection_output = reflection_response.choices[0].message.parsed
if self.verbose:
_get_display_functions()['display_self_reflection'](f"Agent {self.name} self reflection (using {self.reflect_llm if self.reflect_llm else self.llm}): reflection='{reflection_output.reflection}' satisfactory='{reflection_output.satisfactory}'", console=self.console)
# Only consider satisfactory after minimum reflections
if reflection_output.satisfactory == "yes" and reflection_count >= self.min_reflect - 1:
if self.verbose:
_get_display_functions()['display_self_reflection']("Agent marked the response as satisfactory after meeting minimum reflections", console=self.console)
# Add to chat history and return
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed after reflection: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# Check if we've hit max reflections
if reflection_count >= self.max_reflect - 1:
if self.verbose:
_get_display_functions()['display_self_reflection']("Maximum reflection count reached, returning current response", console=self.console)
# Add to chat history and return
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as e:
logging.error(f"Agent {self.name}: Guardrail validation failed after max reflections: {e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
# Regenerate response based on reflection
regenerate_messages = reflection_messages + [
{"role": "assistant", "content": f"Self Reflection: {reflection_output.reflection} Satisfactory?: {reflection_output.satisfactory}"},
{"role": "user", "content": "Now regenerate your response using the reflection you made"}
]
new_response = await self._execute_unified_achat_completion(
messages=regenerate_messages,
temperature=temperature,
tools=formatted_tools,
stream=stream,
reasoning_steps=reasoning_steps,
task_name=task_name,
task_description=task_description,
task_id=task_id,
response_format=response_format
)
if new_response:
new_response_text = self._extract_llm_response_content(new_response)
if isinstance(new_response_text, str):
response_text = new_response_text.strip()
# Update reflection_messages to include the new response for next iteration
reflection_messages = regenerate_messages + [
{"role": "assistant", "content": response_text}
]
reflection_count += 1
except Exception as e:
if self.verbose:
_get_display_functions()['display_error'](f"Error in parsing self-reflection json {e}. Retrying", console=self.console)
logging.error("Reflection parsing failed.", exc_info=True)
reflection_count += 1
if reflection_count >= self.max_reflect:
# Return original response after max reflection attempts
self._append_to_chat_history({"role": "assistant", "content": response_text})
# Persist assistant message to DB
self._persist_message("assistant", response_text)
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
except Exception as guard_e:
logging.error(f"Agent {self.name}: Guardrail validation failed after reflection error: {guard_e}")
# Rollback chat history on guardrail failure
self._truncate_chat_history(chat_history_length)
return None
continue
# This should never be reached due to the returns above
# But adding as safety fallback
logging.warning(f"Agent {self.name}: Unexpected code path reached in unified dispatch")
return None
return await self._finalize_unified_achat_response(
response=response,
original_prompt=original_prompt,
temperature=temperature,
tools=tools,
output_json=output_json,
output_pydantic=output_pydantic,
reasoning_steps=reasoning_steps,
stream=stream,
messages=messages,
chat_history_length=chat_history_length,
start_time=start_time,
task_name=task_name,
task_description=task_description,
task_id=task_id,
)
else:
# LEGACY: Check if OpenAI client is available
if self._openai_client is None:
@@ -2533,7 +2339,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Apply guardrail validation for OpenAI client response
try:
validated_response = await self._aapply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
validated_response = self._apply_guardrail_with_retry(response_text, original_prompt, temperature, tools, task_name, task_description, task_id)
# Execute callback after validation
self._execute_callback_and_display(original_prompt, validated_response, time.time() - start_time, task_name, task_description, task_id)
return await self._atrigger_after_agent_hook(original_prompt, validated_response, start_time)
@@ -403,33 +403,9 @@ class MemoryMixin:
self._session_store.add_user_message(self._session_id, content)
elif role == "assistant":
self._session_store.add_assistant_message(self._session_id, content)
self._persist_session_stats()
except Exception as e:
logging.warning(f"Failed to persist message to session store: {e}")
def _persist_session_stats(self):
"""Flush agent cost/token stats into session JSON metadata."""
store = getattr(self, "_session_store", None)
session_id = getattr(self, "_session_id", None)
if store is None or session_id is None:
return
if not hasattr(store, "update_session_metadata"):
return
try:
summary = getattr(self, "cost_summary", {}) or {}
llm = getattr(self, "llm", None)
model = str(llm) if llm is not None else None
store.update_session_metadata(
session_id,
model=model,
total_tokens=summary.get("tokens_in", 0) + summary.get("tokens_out", 0),
cost=summary.get("cost", 0),
source=getattr(self, "_session_source", "chat"),
agent_id=getattr(self, "agent_id", None) or getattr(self, "_agent_id", None),
)
except Exception as e:
logging.debug(f"Session stats persist skipped: {e}")
def session_id(self) -> Optional[str]:
"""Get the current session ID."""
return self._session_id
@@ -26,21 +26,6 @@ if TYPE_CHECKING:
class ToolExecutionMixin:
"""Mixin providing toolexecution methods for the Agent class."""
def _get_existing_stream_emitter(self):
"""Return an already-initialized stream emitter without creating one."""
emitter = getattr(self, "_stream_emitter", None)
if emitter is not None:
return emitter
# Support name-mangled private attributes across class renames/inheritance.
for cls in type(self).mro():
mangled = f"_{cls.__name__}__stream_emitter"
if hasattr(self, mangled):
emitter = getattr(self, mangled, None)
if emitter is not None:
return emitter
return None
def _resolve_tool_names(self, tool_names):
"""Resolve tool names to actual tool instances from registry.
@@ -175,9 +160,8 @@ class ToolExecutionMixin:
# Emit TOOL_CALL_START to stream_emitter (for AIUI/AG-UI consumers)
# Zero overhead when no callbacks registered
_stream_emitter = self._get_existing_stream_emitter()
if _stream_emitter is not None and _stream_emitter.has_callbacks:
_stream_emitter.emit(StreamEvent(
if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
type=StreamEventType.TOOL_CALL_START,
timestamp=_tool_start_perf,
tool_call={
@@ -291,10 +275,10 @@ class ToolExecutionMixin:
# Emit TOOL_CALL_RESULT to stream_emitter (for AIUI/AG-UI consumers)
# Zero overhead when no callbacks registered
if _stream_emitter is not None and _stream_emitter.has_callbacks:
if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
# Truncate result for stream event (keep it reasonable for UI display)
result_summary = str(result)[:500] if result else None
_stream_emitter.emit(StreamEvent(
getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
type=StreamEventType.TOOL_CALL_RESULT,
timestamp=_time.perf_counter(),
tool_call={
@@ -306,18 +290,6 @@ class ToolExecutionMixin:
agent_id=self.name,
metadata={"duration_ms": _duration_ms},
))
_stream_emitter.emit(StreamEvent(
type=StreamEventType.TOOL_CALL_END,
timestamp=_time.perf_counter(),
tool_call={
"name": function_name,
"arguments": arguments,
"result": result_summary,
"id": tool_call_id,
},
agent_id=self.name,
metadata={"duration_ms": _duration_ms},
))
# Trigger AFTER_TOOL hook
from ..hooks import HookEvent, AfterToolInput
@@ -954,3 +926,4 @@ class ToolExecutionMixin:
def pending_approval_count(self) -> int:
"""Number of approval requests still waiting."""
return len(self._pending_approvals)
@@ -15,9 +15,9 @@ from enum import Enum
# Import token tracking
try:
from ..telemetry.token_collector import get_token_collector
from ..telemetry.token_collector import _token_collector
except ImportError:
get_token_collector = None
_token_collector = None
# Import async utility for hot-path usage
try:
@@ -1609,25 +1609,25 @@ class AgentTeam:
original_on_task_start = self.on_task_start
original_on_task_complete = self.on_task_complete
def composed_on_task_start(task, task_id):
def composed_on_task_start(task):
try:
verbose_task_start_callback(task, task_id)
verbose_task_start_callback(task)
except Exception as e:
logging.debug(f"Error in verbose task start callback: {e}")
if original_on_task_start:
try:
original_on_task_start(task, task_id)
original_on_task_start(task)
except Exception as e:
logging.debug(f"Error in original task start callback: {e}")
def composed_on_task_complete(task, task_output):
def composed_on_task_complete(task):
try:
verbose_task_complete_callback(task, task_output)
verbose_task_complete_callback(task)
except Exception as e:
logging.debug(f"Error in verbose task complete callback: {e}")
if original_on_task_complete:
try:
original_on_task_complete(task, task_output)
original_on_task_complete(task)
except Exception as e:
logging.debug(f"Error in original task complete callback: {e}")
@@ -1826,19 +1826,18 @@ class AgentTeam:
def get_token_usage_summary(self) -> Dict[str, Any]:
"""Get a summary of token usage across all agents and tasks."""
if not get_token_collector:
if not _token_collector:
return {"error": "Token tracking not available"}
return get_token_collector().get_session_summary()
return _token_collector.get_session_summary()
def get_detailed_token_report(self) -> Dict[str, Any]:
"""Get a detailed token usage report."""
if not get_token_collector:
if not _token_collector:
return {"error": "Token tracking not available"}
collector = get_token_collector()
summary = collector.get_session_summary()
recent = collector.get_recent_interactions(limit=20)
summary = _token_collector.get_session_summary()
recent = _token_collector.get_recent_interactions(limit=20)
# Calculate cost estimates (example rates)
cost_per_1k_input = 0.0005 # $0.0005 per 1K input tokens
@@ -1862,11 +1861,11 @@ class AgentTeam:
def display_token_usage(self):
"""Display token usage in a formatted table."""
if not get_token_collector:
if not _token_collector:
print("Token tracking not available")
return
summary = get_token_collector().get_session_summary()
summary = _token_collector.get_session_summary()
print("\n" + "="*50)
print("TOKEN USAGE SUMMARY")
@@ -1,13 +0,0 @@
"""Hooks query protocol for UI/API consumers."""
from __future__ import annotations
from typing import Dict, List, Protocol, runtime_checkable
@runtime_checkable
class HooksQueryProtocol(Protocol):
"""Read path for registered hooks."""
def list_hooks_for_api(self) -> List[Dict]:
...
@@ -439,13 +439,3 @@ def has_hook(event: Union[str, HookEvent]) -> bool:
)
return get_default_registry().has_hooks(event)
def list_hooks_for_api() -> List[Dict]:
"""Flat hook list suitable for REST/UI consumers."""
registry = get_default_registry()
flat: List[Dict] = []
for event, hooks in registry.list_hooks().items():
for h in hooks:
flat.append({"event": event, "source": "sdk", **h})
return flat
@@ -68,14 +68,6 @@ class HookEvent(str, Enum):
SCHEDULE_REMOVE = "schedule_remove"
SCHEDULE_TRIGGER = "schedule_trigger"
# Kanban task lifecycle (wrapper dispatcher + tools emit these)
KANBAN_TASK_CREATED = "kanban_task_created"
KANBAN_TASK_CLAIMED = "kanban_task_claimed"
KANBAN_TASK_MOVED = "kanban_task_moved"
KANBAN_TASK_DONE = "kanban_task_done"
KANBAN_TASK_BLOCKED = "kanban_task_blocked"
KANBAN_TASK_FAILED = "kanban_task_failed"
# Claude Code parity events
USER_PROMPT_SUBMIT = "user_prompt_submit" # When user submits a prompt
NOTIFICATION = "notification" # When notification is sent
@@ -109,34 +101,6 @@ class HookInput:
}
@dataclass
class KanbanHookInput(HookInput):
"""Hook input for kanban task lifecycle events.
Used by wrapper dispatcher and tools when emitting kanban events.
Observability adapters can subscribe to these via the hook registry.
"""
task_id: str = ""
board: str = "default"
status: str = ""
assignee: Optional[str] = field(default=None)
from_status: Optional[str] = field(default=None)
to_status: Optional[str] = field(default=None)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
result = super().to_dict()
result.update({
"task_id": self.task_id,
"board": self.board,
"status": self.status,
"assignee": self.assignee,
"from_status": self.from_status,
"to_status": self.to_status,
})
return result
@dataclass
class HookOutput:
"""Base hook output - common fields for all events."""
@@ -1,23 +0,0 @@
"""
Kanban protocols and types for PraisonAI Agents.
This module provides the protocol contracts for kanban functionality,
allowing the wrapper (praisonai) and PraisonAIUI to share a stable
interface without coupling the core to SQLite or other heavy implementations.
"""
from praisonaiagents.kanban.protocols import (
KanbanStoreProtocol,
KanbanTaskProtocol,
KanbanCommentingProtocol,
KanbanLinkingProtocol,
VALID_KANBAN_STATUSES,
)
__all__ = [
"KanbanStoreProtocol",
"KanbanTaskProtocol",
"KanbanCommentingProtocol",
"KanbanLinkingProtocol",
"VALID_KANBAN_STATUSES",
]
@@ -1,221 +0,0 @@
"""
Kanban protocols for PraisonAI Agents.
Defines the protocol contracts that the wrapper (praisonai) and PraisonAIUI
can use to implement kanban functionality with a stable interface.
This follows AGENTS.md §3.2 Protocol-First Design:
- Protocols define WHAT (interface contract)
- Implementations define HOW (concrete behavior)
- Core SDK has protocols only, wrapper has heavy implementations
"""
from typing import Protocol, TypedDict, runtime_checkable
# Valid kanban statuses matching PraisonAIUI columns + archived
VALID_KANBAN_STATUSES: frozenset[str] = frozenset([
"triage",
"todo",
"ready",
"running",
"blocked",
"review",
"done",
"archived"
])
class KanbanTaskProtocol(TypedDict, total=False):
"""Typed dict shape for kanban task fields.
Defines the expected structure of task objects returned by
KanbanStoreProtocol implementations.
"""
id: str
title: str
body: str
status: str
assignee: str | None
priority: str | None
tenant: str | None
board: str
created_at: float
updated_at: float
@runtime_checkable
class KanbanStoreProtocol(Protocol):
"""Protocol contract for kanban store implementations.
This protocol defines the core interface that PraisonAIUI expects
for injected kanban stores. Wrapper implementations must
implement all methods to be compatible.
Duck typing contract matches InjectedKanbanStore in PraisonAIUI.
"""
def get_board(
self,
*,
board: str = "default",
tenant: str | None = None,
include_archived: bool = False,
) -> dict:
"""Get board data with tasks grouped by status.
Args:
board: Board name (default: "default")
tenant: Optional tenant filter
include_archived: Whether to include archived tasks
Returns:
Board data with tasks organized by status columns
"""
...
def get_task(self, task_id: str) -> dict | None:
"""Get a single task by ID.
Args:
task_id: Task identifier
Returns:
Task data or None if not found
"""
...
def create_task(self, data: dict) -> dict:
"""Create a new task.
Args:
data: Task creation data (title, body, status, etc.)
Returns:
Created task data with assigned ID
"""
...
def update_task(self, task_id: str, data: dict) -> dict | None:
"""Update an existing task.
Args:
task_id: Task identifier
data: Fields to update
Returns:
Updated task data or None if not found
"""
...
def move_task(self, task_id: str, status: str) -> dict | None:
"""Move a task to a different status.
Args:
task_id: Task identifier
status: New status (must be in VALID_KANBAN_STATUSES)
Returns:
Updated task data or None if not found
"""
...
def bulk_update(self, task_ids: list[str], status: str) -> dict:
"""Update multiple tasks to the same status.
Args:
task_ids: List of task identifiers
status: New status for all tasks
Returns:
Results summary with success/failure counts
"""
...
def delete_task(self, task_id: str) -> bool:
"""Delete a task.
Args:
task_id: Task identifier
Returns:
True if deleted, False if not found
"""
...
def list_events(self, since: float = 0.0, board: str = "default") -> list[dict]:
"""List kanban events since a timestamp.
Args:
since: Unix timestamp to filter events from
board: Board name to filter events
Returns:
List of event data
"""
...
def health(self) -> dict:
"""Check store health status.
Returns:
Health status information
"""
...
@runtime_checkable
class KanbanCommentingProtocol(Protocol):
"""Extension protocol for kanban task commenting functionality.
This protocol is implemented separately from KanbanStoreProtocol to allow
stores to optionally support commenting without breaking isinstance checks
on the core protocol.
"""
def add_comment(self, task_id: str, text: str, author: str | None = None) -> dict | None:
"""Add a comment to a task.
Args:
task_id: Task identifier
text: Comment text
author: Optional comment author
Returns:
Comment data or None if task not found
"""
...
@runtime_checkable
class KanbanLinkingProtocol(Protocol):
"""Extension protocol for kanban task linking functionality.
This protocol is implemented separately from KanbanStoreProtocol to allow
stores to optionally support task relationships without breaking isinstance
checks on the core protocol.
"""
def link_tasks(self, parent_id: str, child_id: str) -> bool:
"""Link two tasks in parent-child relationship.
Args:
parent_id: Parent task identifier
child_id: Child task identifier
Returns:
True if linked successfully, False otherwise
"""
...
def unlink_tasks(self, parent_id: str, child_id: str) -> bool:
"""Unlink parent-child task relationship.
Args:
parent_id: Parent task identifier
child_id: Child task identifier
Returns:
True if unlinked successfully, False otherwise
"""
...
@@ -174,14 +174,7 @@ class ChromaKnowledgeAdapter:
# Add where filter if any conditions specified
if where_filter:
# ChromaDB validation expects exactly one condition in the where clause
# When multiple conditions exist, use $and operator to combine them
if len(where_filter) == 1:
query_kwargs["where"] = where_filter
else:
# Convert multiple conditions to ChromaDB's $and format
and_conditions = [{k: v} for k, v in where_filter.items()]
query_kwargs["where"] = {"$and": and_conditions}
query_kwargs["where"] = where_filter
response = self.collection.query(**query_kwargs)
@@ -204,10 +197,6 @@ class ChromaKnowledgeAdapter:
return SearchResult(results=items)
except Exception as e:
# Log ChromaDB query errors for better debugging
import logging
logger = logging.getLogger(__name__)
logger.warning(f"ChromaDB search failed: {e}")
return SearchResult(results=[])
def add(self, content: Any, *, user_id: Optional[str] = None, agent_id: Optional[str] = None,
@@ -73,10 +73,10 @@ def _get_live():
# Import token tracking
try:
from ..telemetry.token_collector import TokenMetrics, get_token_collector
from ..telemetry.token_collector import TokenMetrics, _token_collector
except ImportError:
TokenMetrics = None
get_token_collector = None
_token_collector = None
# Logging is already configured in _logging.py via __init__.py
@@ -4519,7 +4519,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
def _track_token_usage(self, response: Dict[str, Any], model: str) -> Optional[TokenMetrics]:
"""Extract and track token usage from LLM response."""
if not TokenMetrics or not get_token_collector:
if not TokenMetrics or not _token_collector:
return None
# Note: metrics check moved to call sites for performance
@@ -4551,7 +4551,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
# Track in global collector
# Extract provider from model string (e.g., 'openai' from 'openai/gpt-4o-mini')
provider = model.split('/')[0] if '/' in model else 'litellm'
get_token_collector().track_tokens(
_token_collector.track_tokens(
model=model,
agent=self.current_agent_name,
metrics=metrics,
@@ -6,13 +6,7 @@ No heavy imports - only stdlib.
"""
import hashlib
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Union
if TYPE_CHECKING:
from ..knowledge.models import SearchResultItem
ResultItem = Union[Dict[str, Any], "SearchResultItem"]
from typing import Any, Dict, List, Optional, Set, Tuple
def _estimate_tokens(text: str) -> int:
@@ -25,39 +19,6 @@ def _estimate_tokens(text: str) -> int:
return len(text) // 4 + 1
def _extract_value(item: ResultItem, key: str, default: Any = None) -> Any:
"""
Extract value from either dict or object (e.g., SearchResultItem).
Args:
item: Dictionary or object with attributes
key: Key/attribute name to extract
default: Default value if key not found
Returns:
Extracted value or default
"""
if isinstance(item, dict):
return item.get(key, default)
else:
return getattr(item, key, default)
def _extract_metadata_value(
item: ResultItem,
metadata: Dict[str, Any],
key: str,
default: Any = "",
) -> Any:
"""
Extract metadata value with fallback to top-level item attribute.
"""
value = metadata.get(key)
if value is None:
return _extract_value(item, key, default)
return value
def _chunk_hash(text: str, source: Optional[str] = None) -> str:
"""
Generate stable hash for chunk deduplication.
@@ -74,38 +35,31 @@ def _chunk_hash(text: str, source: Optional[str] = None) -> str:
def deduplicate_chunks(
results: List[ResultItem],
results: List[Dict[str, Any]],
similarity_threshold: float = 0.9,
) -> List[ResultItem]:
) -> List[Dict[str, Any]]:
"""
Deduplicate chunks by content hash.
Handles both dict format and SearchResultItem objects.
Args:
results: List of search results (dicts or SearchResultItem objects)
results: List of search results with 'text' or 'memory' key
similarity_threshold: Not used currently (hash-based dedup)
Returns:
Deduplicated list of results
"""
seen_hashes: Set[str] = set()
unique_results: List[ResultItem] = []
unique_results: List[Dict[str, Any]] = []
for result in results:
# Skip None results
if result is None:
continue
# Handle different result formats (dict or SearchResultItem)
text = _extract_value(result, "text") or _extract_value(result, "memory", "")
# Handle different result formats
text = result.get("text") or result.get("memory", "")
# CRITICAL: Handle metadata=None from mem0 - ensure always dict
metadata = _extract_value(result, "metadata") or {}
if not isinstance(metadata, dict):
metadata = {}
source = _extract_metadata_value(result, metadata, "source", "")
metadata = result.get("metadata") or {}
source = metadata.get("source", "")
chunk_id = _chunk_hash(text, source)
@@ -152,19 +106,17 @@ def truncate_context(
def build_context(
results: List[ResultItem],
results: List[Dict[str, Any]],
max_tokens: int = 4000,
deduplicate: bool = True,
separator: str = "\n\n---\n\n",
include_source: bool = True,
) -> Tuple[str, List[ResultItem]]:
) -> Tuple[str, List[Dict[str, Any]]]:
"""
Build context string from retrieval results.
Handles both dict format and SearchResultItem objects.
Args:
results: List of search results (dicts or SearchResultItem objects)
results: List of search results
max_tokens: Maximum tokens for context
deduplicate: Whether to deduplicate chunks
separator: Separator between chunks
@@ -181,7 +133,7 @@ def build_context(
results = deduplicate_chunks(results)
context_parts: List[str] = []
used_results: List[ResultItem] = []
used_results: List[Dict[str, Any]] = []
current_tokens = 0
separator_tokens = _estimate_tokens(separator)
@@ -189,21 +141,17 @@ def build_context(
# Skip None results
if result is None:
continue
# Handle different result formats (dict or SearchResultItem)
text = _extract_value(result, "text") or _extract_value(result, "memory", "")
# Handle different result formats
text = result.get("text") or result.get("memory", "")
if not text:
continue
# Build chunk text with optional source
if include_source:
# CRITICAL: Handle metadata=None from mem0 - ensure always dict
metadata = _extract_value(result, "metadata") or {}
if not isinstance(metadata, dict):
metadata = {}
source = _extract_metadata_value(result, metadata, "source", "")
filename = _extract_metadata_value(result, metadata, "filename", "")
metadata = result.get("metadata") or {}
source = metadata.get("source", "")
filename = metadata.get("filename", "")
source_label = filename or source or f"Source {i + 1}"
chunk_text = f"[{source_label}]\n{text}"
else:
@@ -246,14 +194,11 @@ class DefaultContextBuilder:
def build(
self,
results: List[ResultItem],
results: List[Dict[str, Any]],
max_tokens: int = 4000,
deduplicate: bool = True,
) -> str:
"""Build context string from results.
Handles both dict format and SearchResultItem objects.
"""
"""Build context string from results."""
context, _ = build_context(
results=results,
max_tokens=max_tokens,
@@ -1,29 +0,0 @@
# Scheduler — runner vs loop
## ScheduleRunner (stateless)
`ScheduleRunner` is **stateless**. Call `get_due_jobs()` whenever you want to check for jobs that should fire. It does not run background threads.
Use when:
- CLI one-shot tick (`praisonai schedule tick`)
- Host integration that polls on its own timer
- Tests
## ScheduleLoop (optional daemon)
`ScheduleLoop` is an **optional background daemon**. You must call `.start()` explicitly; it polls the runner and invokes `on_trigger` callbacks.
Use when:
- Long-running host process (Pattern B/C via `setup_bridges()`)
- Standalone scheduler service
```python
from praisonaiagents.scheduler import FileScheduleStore, ScheduleRunner, ScheduleLoop
store = FileScheduleStore()
runner = ScheduleRunner(store)
loop = ScheduleLoop(runner)
loop.start() # optional daemon
```
The wrapper host calls `ensure_schedule_runner()` which starts `ScheduleLoop` when the scheduler optional dependency is available.
@@ -32,22 +32,7 @@ When you provide a `session_id` to an Agent:
### Default Storage Location
Sessions are stored in: `~/.praisonai/sessions/{session_id}.json`
### Session metadata contract (Issue #48)
List endpoints and UI dashboards may expose these optional root-level fields:
| Field | Type | Description |
|-------|------|-------------|
| `model` | string | LLM model used in the session |
| `total_tokens` | int | Cumulative input+output tokens |
| `cost` | float | Estimated USD cost |
| `agent_id` | string | Gateway or registry agent id |
| `source` | string | Origin: `chat`, `gateway`, `cli`, `api` |
| `agent_name` | string | Human-readable agent name |
Fields are persisted in session JSON `metadata` and mirrored at root on save. Populated automatically after each assistant turn when using the default session store.
Sessions are stored in: `~/.praison/sessions/{session_id}.json`
### Session File Format
@@ -60,13 +45,7 @@ Fields are persisted in session JSON `metadata` and mirrored at root on save. Po
],
"created_at": "2026-01-02T04:00:00+00:00",
"updated_at": "2026-01-02T04:01:00+00:00",
"agent_name": "Assistant",
"agent_id": "agent-abc",
"source": "chat",
"model": "gpt-4o-mini",
"total_tokens": 128,
"cost": 0.0004,
"metadata": {}
"agent_name": "Assistant"
}
```
@@ -144,35 +144,6 @@ class HierarchicalSessionStore(DefaultSessionStore):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._extended_cache: Dict[str, ExtendedSessionData] = {}
def _load_session_from_disk(self, session_id: str, filepath: str) -> ExtendedSessionData:
"""Load extended session JSON from disk (caller must hold FileLock)."""
if os.path.exists(filepath):
try:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
return ExtendedSessionData.from_dict(data)
except (json.JSONDecodeError, IOError):
pass
return ExtendedSessionData(session_id=session_id)
def _modify_session_locked(
self,
session_id: str,
mutator,
*,
error_label: str = "modify session",
) -> bool:
"""Locked read-modify-write preserving extended session fields."""
result = super()._modify_session_locked(
session_id, mutator, error_label=error_label
)
if result:
with self._lock:
cached = self._cache.get(session_id)
if isinstance(cached, ExtendedSessionData):
self._extended_cache[session_id] = cached
return result
def add_message(
self,
@@ -186,22 +157,25 @@ class HierarchicalSessionStore(DefaultSessionStore):
Overrides parent to preserve extended session data.
"""
from .store import SessionMessage
# Load extended session (force reload to get latest)
session = self._load_extended_session(session_id, force_reload=True)
message = SessionMessage(
role=role,
content=content,
timestamp=time.time(),
metadata=metadata or {},
)
def _apply(session: SessionData) -> None:
session.messages.append(message)
if len(session.messages) > self.max_messages:
session.messages = session.messages[-self.max_messages :]
return self._modify_session_locked(
session_id, _apply, error_label="add message to session"
)
session.messages.append(message)
# Trim messages if over limit
if len(session.messages) > self.max_messages:
session.messages = session.messages[-self.max_messages:]
return self._save_extended_session(session)
def _load_extended_session(self, session_id: str, force_reload: bool = False) -> ExtendedSessionData:
"""Load extended session from disk."""
@@ -129,16 +129,3 @@ class SessionStoreProtocol(Protocol):
True if the session exists.
"""
...
@runtime_checkable
class CheckpointQueryProtocol(Protocol):
"""Read path for session checkpoints / rollback snapshots."""
def list_checkpoints(self, session_id: str) -> List[Dict[str, Any]]:
"""Return checkpoint metadata for a session."""
...
def get_checkpoint(self, session_id: str, checkpoint_id: str) -> Optional[Dict[str, Any]]:
"""Return a single checkpoint payload."""
...
@@ -22,7 +22,7 @@ else:
_HAS_FCNTL = False
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Dict, List, Optional
from ..paths import get_sessions_dir
@@ -81,7 +81,7 @@ class SessionData:
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
data = {
return {
"session_id": self.session_id,
"messages": [m.to_dict() for m in self.messages],
"created_at": self.created_at,
@@ -92,10 +92,6 @@ class SessionData:
"gateway_session_id": self.gateway_session_id,
"agent_id": self.agent_id,
}
for key in ("model", "llm", "total_tokens", "token_count", "cost", "source"):
if key in self.metadata:
data[key] = self.metadata[key]
return data
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "SessionData":
@@ -141,8 +137,7 @@ class FileLock:
self._lock_path = filepath + ".lock"
def __enter__(self):
if not self.acquire():
raise IOError(f"Failed to acquire file lock for {self.filepath} after {self.timeout}s")
self.acquire()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
@@ -291,72 +286,6 @@ class DefaultSessionStore:
self._cache[session_id] = session
return session
def _load_session_from_disk(self, session_id: str, filepath: str) -> SessionData:
"""Load session JSON from disk (caller must hold FileLock)."""
if os.path.exists(filepath):
try:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
return SessionData.from_dict(data)
except (json.JSONDecodeError, IOError):
pass
return SessionData(session_id=session_id)
def _read_session_fresh(self, session_id: str) -> SessionData:
"""Reload session from disk and refresh the in-process cache."""
filepath = self._get_session_path(session_id)
with FileLock(filepath, self.lock_timeout):
session = self._load_session_from_disk(session_id, filepath)
with self._lock:
self._cache[session_id] = session
return session
def _modify_session_locked(
self,
session_id: str,
mutator: Callable[[SessionData], None],
*,
error_label: str = "modify session",
) -> bool:
"""Apply mutator after reloading from disk under FileLock."""
filepath = self._get_session_path(session_id)
with FileLock(filepath, self.lock_timeout):
session = self._load_session_from_disk(session_id, filepath)
mutator(session)
session.updated_at = datetime.now(timezone.utc).isoformat()
if len(session.messages) > self.max_messages:
session.messages = session.messages[-self.max_messages:]
try:
dir_path = os.path.dirname(filepath) or "."
os.makedirs(dir_path, exist_ok=True)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=dir_path,
delete=False,
suffix=".tmp",
) as f:
json.dump(session.to_dict(), f, indent=2, ensure_ascii=False)
temp_path = f.name
os.replace(temp_path, filepath)
with self._lock:
self._cache[session_id] = session
return True
except (IOError, OSError) as e:
logger.error(f"Failed to {error_label} {session_id}: {e}")
try:
if "temp_path" in locals():
os.remove(temp_path)
except (IOError, OSError):
pass
return False
def _save_session(self, session: SessionData) -> bool:
"""Save session to disk with atomic write."""
@@ -506,13 +435,13 @@ class DefaultSessionStore:
Returns:
List of {"role": "user/assistant", "content": "..."} dicts.
"""
session = self._read_session_fresh(session_id)
session = self._load_session(session_id)
limit = max_messages or self.max_messages
return session.get_chat_history(limit)
def get_session(self, session_id: str) -> SessionData:
"""Get full session data."""
return self._read_session_fresh(session_id)
return self._load_session(session_id)
def set_agent_info(
self,
@@ -521,42 +450,27 @@ class DefaultSessionStore:
user_id: Optional[str] = None,
) -> bool:
"""Set agent info for a session."""
def _apply(session: SessionData) -> None:
session = self._load_session(session_id)
with self._lock:
if agent_name:
session.agent_name = agent_name
if user_id:
session.user_id = user_id
return self._modify_session_locked(
session_id, _apply, error_label="set agent info for session"
)
self._cache[session_id] = session
return self._save_session(session)
def clear_session(self, session_id: str) -> bool:
"""Clear all messages from a session."""
return self._modify_session_locked(
session_id,
lambda session: session.messages.clear(),
error_label="clear session",
)
def update_session_metadata(self, session_id: str, **fields: Any) -> bool:
"""Merge run stats / metadata fields into a persisted session."""
if not fields:
return True
def _apply(session: SessionData) -> None:
for key, value in fields.items():
if value is None:
continue
session.metadata[key] = value
if key in ("agent_id", "agent_name", "user_id"):
setattr(session, key, value)
return self._modify_session_locked(
session_id, _apply, error_label="update session metadata"
)
session = self._load_session(session_id)
with self._lock:
session.messages.clear()
self._cache[session_id] = session
return self._save_session(session)
def delete_session(self, session_id: str) -> bool:
"""Delete a session completely."""
filepath = self._get_session_path(session_id)
@@ -586,16 +500,10 @@ class DefaultSessionStore:
data = json.load(f)
sessions.append({
"session_id": data.get("session_id", filename[:-5]),
"id": data.get("session_id", filename[:-5]),
"agent_name": data.get("agent_name"),
"agent_id": data.get("agent_id") or (data.get("metadata") or {}).get("agent_id"),
"source": data.get("source") or (data.get("metadata") or {}).get("source"),
"created_at": data.get("created_at"),
"updated_at": data.get("updated_at"),
"message_count": len(data.get("messages", [])),
"model": data.get("model") or data.get("llm") or (data.get("metadata") or {}).get("model"),
"total_tokens": data.get("total_tokens") or data.get("token_count") or (data.get("metadata") or {}).get("total_tokens"),
"cost": data.get("cost") or (data.get("metadata") or {}).get("cost"),
})
except (json.JSONDecodeError, IOError):
continue
@@ -655,7 +563,7 @@ class DefaultSessionStore:
List of SessionData objects for the specified agent
"""
session_ids = self.list_sessions_by_agent(agent_name, limit)
return [self._read_session_fresh(sid) for sid in session_ids]
return [self._load_session(sid) for sid in session_ids]
def get_agent_chat_history(
self,
@@ -709,15 +617,16 @@ class DefaultSessionStore:
Returns:
True if saved successfully
"""
def _apply(session: SessionData) -> None:
session = self._load_session(session_id)
with self._lock:
if gateway_session_id:
session.gateway_session_id = gateway_session_id
if agent_id:
session.agent_id = agent_id
return self._modify_session_locked(
session_id, _apply, error_label="set gateway info for session"
)
self._cache[session_id] = session
return self._save_session(session)
def get_by_gateway_session(self, gateway_session_id: str) -> Optional[SessionData]:
"""Get session data linked to a gateway session.
@@ -19,9 +19,7 @@ Usage:
__all__ = [
# Models
"SkillProperties",
"SkillMetadata",
"SkillRequirements",
"SkillState",
"SkillMetadata",
"ParseError",
"ValidationError",
# Parser
@@ -56,17 +54,13 @@ __all__ = [
"SkillActivatedEvent",
# Budget
"SkillPromptBudget",
# Capability validation
"CapabilityValidator",
"EnforcementLevel",
"ValidationResult",
]
def __getattr__(name: str):
"""Lazy load module components to avoid import overhead."""
if name in ("SkillProperties", "SkillMetadata", "SkillRequirements", "SkillState", "ParseError", "ValidationError"):
from .models import SkillProperties, SkillMetadata, SkillRequirements, SkillState, ParseError, ValidationError
if name in ("SkillProperties", "SkillMetadata", "ParseError", "ValidationError"):
from .models import SkillProperties, SkillMetadata, ParseError, ValidationError
return locals()[name]
if name in ("parse_frontmatter", "find_skill_md", "read_properties"):
@@ -136,9 +130,5 @@ def __getattr__(name: str):
return None
return load_skill
if name in ("CapabilityValidator", "EnforcementLevel", "ValidationResult"):
from .capability_validator import CapabilityValidator, EnforcementLevel, ValidationResult
return locals()[name]
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -1,236 +0,0 @@
"""Capability validation for Agent Skills.
This module implements the capability gate enforcement system for skills,
checking that required tools, servers, and environment variables are available
before allowing skill activation.
"""
import logging
import os
from dataclasses import dataclass
from typing import Dict, List, Optional, Set, Tuple
from enum import Enum
from .models import SkillProperties, SkillRequirements, SkillState
logger = logging.getLogger(__name__)
class EnforcementLevel(Enum):
"""Enforcement level for capability gates."""
DISABLED = "disabled" # No enforcement (existing behavior)
TELEMETRY = "telemetry" # Log only, no blocking
WARN = "warn" # Warning but allow activation
STRICT = "strict" # Hard failure, block activation
@dataclass
class ValidationResult:
"""Result of capability validation for a skill."""
skill_name: str
state: SkillState
satisfied_tools: List[str]
missing_tools: List[str]
satisfied_servers: List[str]
missing_servers: List[str]
satisfied_env_vars: List[str]
missing_env_vars: List[str]
warnings: List[str]
errors: List[str]
@property
def is_fully_satisfied(self) -> bool:
"""Check if all requirements are satisfied."""
return not any([
self.missing_tools,
self.missing_servers,
self.missing_env_vars,
self.errors
])
@property
def has_critical_missing(self) -> bool:
"""Check if critical requirements are missing (tools, servers)."""
return bool(self.missing_tools or self.missing_servers)
def to_dict(self) -> Dict:
"""Convert to dictionary for serialization."""
return {
"skill_name": self.skill_name,
"state": self.state.value,
"satisfied_tools": self.satisfied_tools,
"missing_tools": self.missing_tools,
"satisfied_servers": self.satisfied_servers,
"missing_servers": self.missing_servers,
"satisfied_env_vars": self.satisfied_env_vars,
"missing_env_vars": self.missing_env_vars,
"warnings": self.warnings,
"errors": self.errors,
"is_fully_satisfied": self.is_fully_satisfied,
"has_critical_missing": self.has_critical_missing
}
class CapabilityValidator:
"""Validates skill requirements against available capabilities.
This is the core capability gate enforcement engine. It checks skill
requirements against the runtime context (available tools, servers, etc.)
and determines whether a skill should be activated based on policy.
"""
def __init__(self, enforcement_level: EnforcementLevel = EnforcementLevel.WARN):
"""Initialize the validator.
Args:
enforcement_level: How strictly to enforce requirements
"""
self.enforcement_level = enforcement_level
self._tool_cache: Optional[Set[str]] = None
self._server_cache: Optional[Set[str]] = None
def validate_skill(
self,
skill: SkillProperties,
available_tools: Optional[Set[str]] = None,
available_servers: Optional[Set[str]] = None
) -> ValidationResult:
"""Validate a skill's requirements against available capabilities.
Args:
skill: Skill to validate
available_tools: Set of available tool names (auto-detected if None)
available_servers: Set of available server names (auto-detected if None)
Returns:
ValidationResult with validation status and details
"""
if not skill.requirements or skill.requirements.is_empty():
# No requirements specified - skill is always available
return ValidationResult(
skill_name=skill.name,
state=SkillState.ACTIVE,
satisfied_tools=[],
missing_tools=[],
satisfied_servers=[],
missing_servers=[],
satisfied_env_vars=[],
missing_env_vars=[],
warnings=[],
errors=[]
)
# Get available capabilities
if available_tools is None:
available_tools = self._get_available_tools()
if available_servers is None:
available_servers = self._get_available_servers()
requirements = skill.requirements
# Validate tools
satisfied_tools = [tool for tool in requirements.tools if tool in available_tools]
missing_tools = [tool for tool in requirements.tools if tool not in available_tools]
# Validate servers
satisfied_servers = [srv for srv in requirements.servers if srv in available_servers]
missing_servers = [srv for srv in requirements.servers if srv not in available_servers]
# Validate environment variables
satisfied_env_vars = [var for var in requirements.env_vars if os.getenv(var)]
missing_env_vars = [var for var in requirements.env_vars if not os.getenv(var)]
# Determine warnings and errors
warnings = []
errors = []
if missing_tools:
msg = f"Missing required tools: {', '.join(missing_tools)}"
if self.enforcement_level in (EnforcementLevel.STRICT,):
errors.append(msg)
else:
warnings.append(msg)
if missing_servers:
msg = f"Missing required servers: {', '.join(missing_servers)}"
if self.enforcement_level in (EnforcementLevel.STRICT,):
errors.append(msg)
else:
warnings.append(msg)
if missing_env_vars:
msg = f"Missing required environment variables: {', '.join(missing_env_vars)}"
if self.enforcement_level == EnforcementLevel.STRICT:
errors.append(msg)
else:
warnings.append(msg)
# Determine skill state
if errors:
state = SkillState.UNAVAILABLE
elif warnings:
state = SkillState.DEGRADED
else:
state = SkillState.ACTIVE
result = ValidationResult(
skill_name=skill.name,
state=state,
satisfied_tools=satisfied_tools,
missing_tools=missing_tools,
satisfied_servers=satisfied_servers,
missing_servers=missing_servers,
satisfied_env_vars=satisfied_env_vars,
missing_env_vars=missing_env_vars,
warnings=warnings,
errors=errors
)
# Log based on enforcement level
self._log_validation_result(result)
return result
def _get_available_tools(self) -> Set[str]:
"""Get set of available tool names from registry."""
if self._tool_cache is None:
try:
from ..tools.registry import get_registry
registry = get_registry()
self._tool_cache = set(registry.list_tools())
except ImportError:
logger.debug("Tool registry not available")
self._tool_cache = set()
return self._tool_cache
def _get_available_servers(self) -> Set[str]:
"""Get set of available MCP server names."""
if self._server_cache is None:
# TODO: Implement MCP server discovery
# For now, return empty set - this can be extended later
self._server_cache = set()
return self._server_cache
def _log_validation_result(self, result: ValidationResult) -> None:
"""Log validation result based on enforcement level."""
if self.enforcement_level == EnforcementLevel.DISABLED:
return
skill_name = result.skill_name
if result.state == SkillState.UNAVAILABLE:
if self.enforcement_level == EnforcementLevel.STRICT:
logger.error(f"Skill '{skill_name}' blocked: {'; '.join(result.errors)}")
else:
logger.warning(f"Skill '{skill_name}' has critical missing dependencies: {'; '.join(result.errors)}")
elif result.state == SkillState.DEGRADED:
logger.warning(f"Skill '{skill_name}' partially available: {'; '.join(result.warnings)}")
elif result.state == SkillState.ACTIVE:
logger.debug(f"Skill '{skill_name}' requirements satisfied")
def clear_cache(self) -> None:
"""Clear cached capability information."""
self._tool_cache = None
self._server_cache = None
@@ -108,7 +108,7 @@ class SkillLoader:
return False
try:
content = skill_md.read_text(encoding="utf-8")
content = skill_md.read_text()
_, body = parse_frontmatter(content)
skill.instructions = body
return True
@@ -136,7 +136,7 @@ class SkillLoader:
for script_file in scripts_dir.iterdir():
if script_file.is_file():
try:
scripts[script_file.name] = script_file.read_text(encoding="utf-8")
scripts[script_file.name] = script_file.read_text()
except Exception:
continue
except PermissionError:
@@ -166,7 +166,7 @@ class SkillLoader:
for ref_file in refs_dir.iterdir():
if ref_file.is_file():
try:
refs[ref_file.name] = ref_file.read_text(encoding="utf-8")
refs[ref_file.name] = ref_file.read_text()
except Exception:
continue
except PermissionError:
@@ -5,13 +5,12 @@ from typing import List, Optional, Dict
logger = logging.getLogger(__name__)
from .models import SkillMetadata, SkillState
from .models import SkillMetadata
from .discovery import discover_skills, get_default_skill_dirs
from .loader import SkillLoader, LoadedSkill
from .prompt import generate_skills_xml
from .substitution import render_skill_body
from .shell_render import render_shell_blocks
from .capability_validator import CapabilityValidator, EnforcementLevel, ValidationResult
class SkillManager:
@@ -35,21 +34,11 @@ class SkillManager:
manager.activate(skill)
"""
def __init__(self, enforcement_level: Optional[EnforcementLevel] = None):
"""Initialize the SkillManager.
Args:
enforcement_level: Capability enforcement level (defaults to environment-based)
"""
def __init__(self):
"""Initialize the SkillManager."""
self._skills: Dict[str, LoadedSkill] = {}
self._loader = SkillLoader()
self._discovered = False
self._validation_cache: Dict[str, ValidationResult] = {}
# Initialize capability validator
if enforcement_level is None:
enforcement_level = self._get_default_enforcement_level()
self._validator = CapabilityValidator(enforcement_level)
@property
def skills(self) -> List[LoadedSkill]:
@@ -137,29 +126,15 @@ class SkillManager:
def get_available_skills(self) -> List[SkillMetadata]:
"""Get metadata for all available skills.
This is used for system prompt injection. Skills are filtered by:
1. disable-model-invocation: true (excluded)
2. Capability enforcement (if strict mode, exclude UNAVAILABLE skills)
This is used for system prompt injection. Skills with
``disable-model-invocation: true`` are omitted so the LLM never
sees them and cannot auto-trigger them.
"""
available = []
for skill in self._skills.values():
# Skip if explicitly disabled for model invocation
if getattr(skill.properties, "disable_model_invocation", False):
continue
# Check capability requirements if enforcement is strict
if self._validator.enforcement_level == EnforcementLevel.STRICT:
try:
validation = self.validate_skill_capabilities(skill.properties.name)
if validation.state == SkillState.UNAVAILABLE:
continue # Skip unavailable skills in strict mode
except Exception as e:
logger.warning(f"Skipping skill '{skill.properties.name}' due to validation error: {e}")
continue
available.append(skill.metadata)
return available
return [
skill.metadata
for skill in self._skills.values()
if not getattr(skill.properties, "disable_model_invocation", False)
]
def get_user_invocable_skills(self) -> List[LoadedSkill]:
"""Return skills the *user* may invoke via slash-commands.
@@ -591,90 +566,9 @@ author: agent
logger.debug("Failed to clean up temp file %s", temp_path, exc_info=True)
raise
def validate_skill_capabilities(self, skill_name: str, force_refresh: bool = False) -> ValidationResult:
"""Validate a skill's capability requirements.
Args:
skill_name: Name of the skill to validate
force_refresh: If True, bypass cache and re-validate
Returns:
ValidationResult with capability status
Raises:
ValueError: If skill not found
"""
skill = self.get_skill(skill_name)
if skill is None:
raise ValueError(f"Skill '{skill_name}' not found")
# Check cache first
if not force_refresh and skill_name in self._validation_cache:
return self._validation_cache[skill_name]
# Perform validation
result = self._validator.validate_skill(skill.properties)
self._validation_cache[skill_name] = result
return result
def get_available_skills_by_state(self, state: SkillState) -> List[LoadedSkill]:
"""Get skills filtered by their capability validation state.
Args:
state: Desired skill state
Returns:
List of skills in the specified state
"""
matching = []
for skill in self._skills.values():
try:
result = self.validate_skill_capabilities(skill.properties.name)
if result.state == state:
matching.append(skill)
except Exception as e:
logger.warning(f"Failed to validate skill '{skill.properties.name}': {e}")
continue
return matching
def get_skills_diagnostics(self) -> Dict[str, ValidationResult]:
"""Get capability diagnostics for all skills.
Returns:
Dict mapping skill names to their validation results
"""
diagnostics = {}
for skill_name in self.skill_names:
try:
diagnostics[skill_name] = self.validate_skill_capabilities(skill_name)
except Exception as e:
logger.error(f"Failed to validate skill '{skill_name}': {e}")
return diagnostics
def _get_default_enforcement_level(self) -> EnforcementLevel:
"""Get default enforcement level from environment."""
import os
level_str = os.getenv('SKILL_CAPABILITY_ENFORCEMENT', 'warn').lower()
level_map = {
'disabled': EnforcementLevel.DISABLED,
'off': EnforcementLevel.DISABLED,
'telemetry': EnforcementLevel.TELEMETRY,
'log': EnforcementLevel.TELEMETRY,
'warn': EnforcementLevel.WARN,
'warning': EnforcementLevel.WARN,
'strict': EnforcementLevel.STRICT,
'hard': EnforcementLevel.STRICT,
'fail': EnforcementLevel.STRICT,
}
return level_map.get(level_str, EnforcementLevel.WARN)
def clear(self) -> None:
"""Clear all loaded skills."""
self._skills.clear()
self._validation_cache.clear()
self._validator.clear_cache()
self._discovered = False
def __len__(self) -> int:
@@ -1,9 +1,8 @@
"""Data models for Agent Skills."""
from dataclasses import dataclass, field
from typing import Optional, List, Union
from typing import Optional
from pathlib import Path
from enum import Enum
class ParseError(Exception):
@@ -16,105 +15,6 @@ class ValidationError(Exception):
pass
class SkillState(Enum):
"""Skill activation state based on requirement validation."""
ACTIVE = "active" # All requirements satisfied
DEGRADED = "degraded" # Some requirements missing (soft warn)
UNAVAILABLE = "unavailable" # Critical requirements missing (hard fail)
UNKNOWN = "unknown" # Requirements not yet validated
@dataclass
class SkillRequirements:
"""Capability requirements for a skill.
This represents the parsed and normalized requirements from skill frontmatter.
Supports both Hermes/OpenClaw conventions and PraisonAI-specific extensions.
Attributes:
servers: Required MCP servers or service endpoints
tools: Required tool names (from tool registry)
env_vars: Required environment variables (use sparingly)
openclaw_hints: OpenClaw-specific metadata (passthrough)
"""
servers: List[str] = field(default_factory=list)
tools: List[str] = field(default_factory=list)
env_vars: List[str] = field(default_factory=list)
openclaw_hints: dict = field(default_factory=dict)
@classmethod
def from_frontmatter(cls, metadata: dict) -> "SkillRequirements":
"""Parse requirements from skill frontmatter.
Normalizes various naming conventions:
- requires_servers, requires_tools (Hermes/OpenClaw style)
- allowed-tools (existing PraisonAI convention)
- openclaw (OpenClaw-specific hints)
Args:
metadata: Raw frontmatter dict
Returns:
SkillRequirements instance
"""
servers = []
tools = []
env_vars = []
openclaw_hints = {}
# Parse servers
if "requires_servers" in metadata:
servers = cls._normalize_list(metadata["requires_servers"])
elif "requires-servers" in metadata:
servers = cls._normalize_list(metadata["requires-servers"])
# Parse tools (multiple naming conventions)
if "requires_tools" in metadata:
tools.extend(cls._normalize_list(metadata["requires_tools"]))
elif "requires-tools" in metadata:
tools.extend(cls._normalize_list(metadata["requires-tools"]))
# Support existing allowed-tools convention for backward compatibility
if "allowed-tools" in metadata:
tools.extend(cls._normalize_list(metadata["allowed-tools"]))
# Parse environment variables
if "requires_env" in metadata:
env_vars = cls._normalize_list(metadata["requires_env"])
elif "requires-env" in metadata:
env_vars = cls._normalize_list(metadata["requires-env"])
# OpenClaw hints (passthrough)
if "openclaw" in metadata:
openclaw_hints = metadata["openclaw"] if isinstance(metadata["openclaw"], dict) else {}
return cls(
servers=servers,
tools=tools,
env_vars=env_vars,
openclaw_hints=openclaw_hints
)
@staticmethod
def _normalize_list(value: Union[str, List[str]]) -> List[str]:
"""Normalize string or list to list of strings."""
if isinstance(value, str):
# Support both comma-separated and space-separated strings
return [item.strip() for item in value.replace(',', ' ').split() if item.strip()]
elif isinstance(value, (list, tuple)):
return [str(item).strip() for item in value if str(item).strip()]
else:
return []
def is_empty(self) -> bool:
"""Check if no requirements are specified."""
return not any([self.servers, self.tools, self.env_vars, self.openclaw_hints])
def __bool__(self) -> bool:
"""Truth value: True if any requirements are specified."""
return not self.is_empty()
@dataclass
class SkillProperties:
"""Properties parsed from a skill's SKILL.md frontmatter.
@@ -149,8 +49,6 @@ class SkillProperties:
hooks: Optional[dict] = None
paths: Optional[object] = None # list[str] glob patterns
shell: Optional[str] = None # "bash" | "powershell"
# Capability requirements (new)
requirements: Optional[SkillRequirements] = None
def to_dict(self) -> dict:
"""Convert to dictionary, excluding None values and empty metadata."""
@@ -3,7 +3,7 @@
from pathlib import Path
from typing import Optional, Tuple
from .models import ParseError, ValidationError, SkillProperties, SkillRequirements
from .models import ParseError, ValidationError, SkillProperties
def find_skill_md(skill_dir: Path) -> Optional[Path]:
@@ -100,7 +100,7 @@ def read_properties(skill_dir: Path) -> SkillProperties:
if skill_md is None:
raise ParseError(f"SKILL.md not found in {skill_dir}")
content = skill_md.read_text(encoding="utf-8")
content = skill_md.read_text()
metadata, _ = parse_frontmatter(content)
if "name" not in metadata:
@@ -143,6 +143,4 @@ def read_properties(skill_dir: Path) -> SkillProperties:
hooks=metadata.get("hooks") if isinstance(metadata.get("hooks"), dict) else None,
paths=metadata.get("paths"),
shell=metadata.get("shell"),
# Parse capability requirements
requirements=SkillRequirements.from_frontmatter(metadata),
)
@@ -7,7 +7,7 @@ plugin bundle, enterprise policy) without touching the core.
from __future__ import annotations
from typing import Protocol, Iterable, Optional, runtime_checkable, List, Dict, Any
from typing import Protocol, Iterable, Optional, runtime_checkable
from .models import SkillProperties
@@ -45,8 +45,7 @@ class SkillMutatorProtocol(Protocol):
"""Agent-managed skill CRUD operations.
Allows agents to create, edit, and manage their own skills at runtime.
Implementations should provide safe-by-default behaviour: use ``propose=True``
(default) to stage mutations for human approval before writing to disk.
Implementations should provide safe-by-default behavior (e.g., propose mode).
"""
def create(self, name: str, content: str, category: Optional[str] = None,
@@ -164,30 +163,3 @@ class SkillMutatorProtocol(Protocol):
Status message describing the action taken
"""
...
@runtime_checkable
class SkillsCatalogProtocol(Protocol):
"""Metadata listing for skills — UI catalog pages."""
def list_skills(self) -> List[Dict[str, Any]]:
"""Return skill metadata dicts (name, description, location)."""
...
def list_skills_for_api() -> List[Dict[str, Any]]:
"""Default catalog adapter using SkillManager when available."""
try:
from .manager import SkillManager
manager = SkillManager()
return [
{
"name": s.name,
"description": s.description,
"location": getattr(s, "location", ""),
}
for s in manager.get_available_skills()
]
except Exception:
return []
@@ -191,11 +191,9 @@ def validate(skill_dir: Path, strict: bool = False) -> List[str]:
return ["Missing required file: SKILL.md"]
try:
content = skill_md.read_text(encoding="utf-8")
content = skill_md.read_text()
metadata, _ = parse_frontmatter(content)
except ParseError as e:
return [str(e)]
except UnicodeDecodeError as e:
return [f"SKILL.md is not valid UTF-8: {e}"]
return validate_metadata(metadata, skill_dir, strict=strict)
@@ -100,107 +100,3 @@ class InMemoryTokenUsageSink:
def clear(self) -> None:
"""Clear all records."""
self.records.clear()
@runtime_checkable
class UsageQueryProtocol(Protocol):
"""Read path for token usage — UI dashboards and analytics."""
def get_summary(self) -> Dict[str, Any]:
"""Aggregate totals and breakdowns by model/agent."""
...
def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
"""Recent usage records, newest last."""
...
def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
...
def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
...
class InMemoryUsageQuery:
"""Query adapter backed by InMemoryTokenUsageSink records."""
def __init__(self, sink: InMemoryTokenUsageSink):
self._sink = sink
def get_summary(self) -> Dict[str, Any]:
records = self._sink.records
total_in = 0
total_out = 0
total_all = 0
by_model: Dict[str, Dict[str, int]] = {}
by_agent: Dict[str, Dict[str, int]] = {}
# Single-pass optimization: calculate everything in one loop
for r in records:
r_in = r.get("input_tokens", 0)
r_out = r.get("output_tokens", 0)
r_total = r.get("total_tokens", r_in + r_out)
total_in += r_in
total_out += r_out
total_all += r_total
model = r.get("model", "unknown")
agent = r.get("agent_name", "unknown")
# Update by_model and by_agent efficiently
for bucket, key in ((by_model, model), (by_agent, agent)):
stats = bucket.setdefault(key, {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0})
stats["input_tokens"] += r_in
stats["output_tokens"] += r_out
stats["total_tokens"] += r_total
return {
"total_requests": len(records),
"total_input_tokens": total_in,
"total_output_tokens": total_out,
"total_tokens": total_all, # Use consistent calculation
"by_model": by_model,
"by_agent": by_agent,
}
def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
return self._sink.records[-limit:]
def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
return self._sink.get_by_task(task_id)
def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
return self._sink.get_by_agent(agent_name)
class TokenCollectorUsageQuery:
"""Query adapter reading from the global TokenCollector + optional sink."""
def __init__(self, collector: Any):
self._collector = collector
def get_summary(self) -> Dict[str, Any]:
summary = self._collector.get_session_summary()
return {
"total_requests": summary.get("total_interactions", 0),
"total_input_tokens": summary.get("total_metrics", {}).get("input_tokens", 0),
"total_output_tokens": summary.get("total_metrics", {}).get("output_tokens", 0),
"total_tokens": summary.get("total_tokens", 0),
"by_model": summary.get("by_model", {}),
"by_agent": summary.get("by_agent", {}),
}
def list_recent(self, limit: int = 50) -> List[Dict[str, Any]]:
return self._collector.get_recent_interactions(limit)
def get_by_task(self, task_id: str) -> List[Dict[str, Any]]:
sink = getattr(self._collector, "_sink", None)
if sink and hasattr(sink, "get_by_task"):
return sink.get_by_task(task_id)
return []
def get_by_agent(self, agent_name: str) -> List[Dict[str, Any]]:
sink = getattr(self._collector, "_sink", None)
if sink and hasattr(sink, "get_by_agent"):
return sink.get_by_agent(agent_name)
return []
@@ -191,9 +191,4 @@ class TokenCollector:
# Global token collector instance
_token_collector = TokenCollector()
def get_token_collector() -> TokenCollector:
"""Return the global TokenCollector singleton."""
return _token_collector
_token_collector = TokenCollector()
@@ -7,7 +7,7 @@ import logging
from praisonaiagents._logging import get_logger
# Import dependencies
from .token_collector import get_token_collector, TokenMetrics
from .token_collector import _token_collector, TokenMetrics
from .telemetry import get_telemetry
logger = get_logger(__name__)
@@ -56,12 +56,11 @@ class TokenTelemetryBridge:
def export_token_metrics(self) -> Dict[str, Any]:
"""Export token metrics for telemetry reporting."""
collector = get_token_collector()
if not collector:
if not _token_collector:
return {}
try:
metrics = collector.export_metrics()
metrics = _token_collector.export_metrics()
# Prepare telemetry-friendly format
return {
@@ -78,7 +77,8 @@ class TokenTelemetryBridge:
def reset_token_metrics(self):
"""Reset token metrics collection."""
get_token_collector().reset()
if _token_collector:
_token_collector.reset()
# Global telemetry bridge instance
_token_telemetry_bridge = TokenTelemetryBridge()
@@ -14,31 +14,10 @@ import json
from typing import Any, Dict, List, Union
from praisonaiagents.tools.decorator import tool
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
def _unwrap_messages(messages: Union[List[Dict[str, Any]], Dict[str, Any], str]) -> List[Dict[str, Any]]:
"""Minimal unwrap for common LLM arg shapes (not full UI normalisation)."""
if isinstance(messages, str):
try:
messages = json.loads(messages)
except json.JSONDecodeError as exc:
raise ValueError(f"messages JSON string is not valid JSON: {exc}") from exc
if isinstance(messages, dict):
if "messages" in messages:
messages = messages["messages"]
else:
messages = [messages]
if not isinstance(messages, list):
raise ValueError("messages must be a list of A2UI message dicts or a JSON string")
return messages
@tool
def send_a2ui_messages(messages: Union[List[Dict[str, Any]], Dict[str, Any], str]) -> Dict[str, Any]:
def send_a2ui_messages(messages: Union[List[Dict[str, Any]], str]) -> Dict[str, Any]:
"""
Send validated A2UI JSON messages to a connected A2UI renderer client.
@@ -53,14 +32,21 @@ def send_a2ui_messages(messages: Union[List[Dict[str, Any]], Dict[str, Any], str
"""
from praisonaiagents.ui.a2ui.adapter import create_a2ui_part
messages = _unwrap_messages(messages)
if isinstance(messages, str):
try:
messages = json.loads(messages)
except json.JSONDecodeError as exc:
raise ValueError(f"messages JSON string is not valid JSON: {exc}") from exc
if not isinstance(messages, list):
raise ValueError("messages must be a list of A2UI message dicts or a JSON string")
part = create_a2ui_part({"messages": messages})
data = getattr(getattr(part, "root", part), "data", messages)
metadata = getattr(getattr(part, "root", part), "metadata", {}) or {}
return {
"mime_type": metadata.get("mimeType", A2UI_MIME_TYPE),
"mime_type": metadata.get("mimeType", "application/json+a2ui"),
"messages": messages,
"a2ui_part": data,
}
@@ -15,7 +15,7 @@ Usage:
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional, Type, get_type_hints
from typing import Any, Dict, Optional, Type, get_type_hints
import inspect
import logging
@@ -234,81 +234,18 @@ class BaseTool(ABC):
if getattr(self.run, '__isabstractmethod__', False):
errors.append("Tool must implement the 'run()' method")
# Check parameters schema is valid and OpenAI-compatible
# Check parameters schema is valid
if self.parameters:
if not isinstance(self.parameters, dict):
errors.append("'parameters' must be a dictionary")
elif "type" not in self.parameters:
errors.append("'parameters' must have a 'type' field")
elif "properties" not in self.parameters:
errors.append("'parameters' must have a 'properties' field for OpenAI compatibility")
# Validate schema structure for OpenAI compatibility
try:
schema = self.get_schema()
if not isinstance(schema, dict):
errors.append("get_schema() must return a dictionary")
elif schema.get("type") != "function":
errors.append("get_schema() must return schema with type='function'")
elif "function" not in schema:
errors.append("get_schema() must have 'function' key")
else:
func = schema["function"]
if not isinstance(func.get("parameters"), dict):
errors.append("Schema function.parameters must be a dictionary")
elif "properties" not in func["parameters"]:
errors.append("Schema function.parameters must have 'properties' for OpenAI compatibility")
except Exception as e:
errors.append(f"Schema generation failed: {e}")
if errors:
raise ToolValidationError(f"Tool '{self.name}' validation failed: {'; '.join(errors)}")
return True
def validate_schema_roundtrip(self) -> bool:
"""Validate that schema can round-trip through OpenAI-style serialization.
This ensures the tool schema is compatible with LLM providers that expect
OpenAI function calling format and that parameters can be properly parsed.
Raises:
ToolValidationError: If round-trip fails
Returns:
True if round-trip validation passes
"""
try:
import json
# Get the schema
schema = self.get_schema()
# Serialize to JSON (what gets sent to LLM)
serialized = json.dumps(schema)
# Deserialize back
deserialized = json.loads(serialized)
# Validate structure matches expectations
if deserialized != schema:
raise ToolValidationError(f"Schema serialization round-trip failed for tool '{self.name}'")
# Validate OpenAI format requirements
func_schema = deserialized.get("function", {})
parameters = func_schema.get("parameters", {})
if not isinstance(parameters.get("properties"), dict):
raise ToolValidationError(f"Tool '{self.name}' parameters.properties must be a dict for OpenAI compatibility")
if "required" in parameters and not isinstance(parameters["required"], list):
raise ToolValidationError(f"Tool '{self.name}' parameters.required must be a list")
return True
except Exception as e:
raise ToolValidationError(f"Schema round-trip validation failed for tool '{self.name}': {e}")
@classmethod
def validate_class(cls) -> bool:
"""Validate a tool class before instantiation.
@@ -345,9 +282,7 @@ def validate_tool(tool: Any) -> bool:
ToolValidationError: If validation fails
"""
if isinstance(tool, BaseTool):
tool.validate()
tool.validate_schema_roundtrip()
return True
return tool.validate()
if callable(tool):
# For plain functions, check they have a name
@@ -359,67 +294,5 @@ def validate_tool(tool: Any) -> bool:
raise ToolValidationError(f"Invalid tool type: {type(tool)}")
def validate_tool_schema_consistency(tools: List[Any]) -> bool:
"""Validate a list of tools for schema consistency with OpenAI format.
This function ensures all tools in a list can be properly serialized
and have consistent schema structures for use with LLM providers.
Args:
tools: List of tool objects to validate
Returns:
True if all tools are valid and consistent
Raises:
ToolValidationError: If validation fails
"""
if not tools:
return True
import json
from .decorator import get_tool_schema
schemas = []
for i, tool in enumerate(tools):
try:
# Validate individual tool
validate_tool(tool)
# Get schema from different tool types
if isinstance(tool, BaseTool):
schema = tool.get_schema()
elif hasattr(tool, 'get_schema') and callable(getattr(tool, 'get_schema')):
schema = tool.get_schema()
elif callable(tool):
schema = get_tool_schema(tool)
else:
raise ToolValidationError(f"Cannot extract schema from tool at index {i}: {type(tool)}")
if not schema:
raise ToolValidationError(f"Tool at index {i} returned empty schema")
# Validate JSON serialization
try:
json.dumps(schema)
except (TypeError, ValueError) as e:
raise ToolValidationError(f"Tool schema at index {i} is not JSON serializable: {e}")
schemas.append(schema)
except Exception as e:
raise ToolValidationError(f"Tool validation failed at index {i}: {e}")
# Check for duplicate tool names
names = set()
for schema in schemas:
name = schema.get("function", {}).get("name")
if name in names:
raise ToolValidationError(f"Duplicate tool name '{name}' found in tool list")
names.add(name)
return True
# For backward compatibility - tools can also just be functions
# The @tool decorator (in decorator.py) wraps functions into BaseTool instances
@@ -214,13 +214,6 @@ def tool(
availability=availability
)
# Validate the tool at creation time for early error detection
try:
tool_instance.validate()
tool_instance.validate_schema_roundtrip()
except Exception as e:
logging.warning(f"Tool validation warning for {tool_instance.name}: {e}")
# Register with global registry if available
try:
from .registry import get_registry
@@ -253,7 +253,7 @@ class A2A:
from praisonaiagents.ui.a2a.conversion import (
extract_user_input,
praisonai_to_a2a_message,
create_response_artifact,
create_artifact,
)
# Validate params
@@ -286,8 +286,7 @@ class A2A:
else:
raise RuntimeError("No agent or agents configured")
response_msg = praisonai_to_a2a_message(str(response), task_id=task_id)
history = getattr(self.agent, "chat_history", None) if self.agent else None
artifact = create_response_artifact(str(response), chat_history=history)
artifact = create_artifact(str(response))
self.task_store.add_to_history(task_id, response_msg)
self.task_store.add_artifact(task_id, artifact)
self.task_store.update_status(task_id, TaskState.COMPLETED)
@@ -323,8 +322,7 @@ class A2A:
response_msg = praisonai_to_a2a_message(
str(response), task_id=task.id
)
history = getattr(self.agent, "chat_history", None) if self.agent else None
artifact = create_response_artifact(str(response), chat_history=history)
artifact = create_artifact(str(response))
# Update task store
self.task_store.add_to_history(task.id, response_msg)
@@ -16,7 +16,6 @@ from praisonaiagents.ui.a2a.types import (
DataPart,
Artifact,
)
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
def a2a_to_praisonai_messages(messages: List[Message]) -> List[Dict[str, Any]]:
@@ -108,75 +107,6 @@ def praisonai_to_a2a_message(
)
def find_a2ui_tool_result(
chat_history: Optional[List[Dict[str, Any]]],
) -> Optional[Dict[str, Any]]:
"""Return the last A2UI tool result dict from agent chat history, if any."""
if not chat_history:
return None
for msg in reversed(chat_history):
if msg.get("role") != "tool":
continue
content = msg.get("content")
if content is None:
continue
data: Any = content if isinstance(content, dict) else None
if data is None:
try:
data = json.loads(content)
except (json.JSONDecodeError, TypeError):
continue
if isinstance(data, dict) and data.get("mime_type") == A2UI_MIME_TYPE:
return data
return None
def create_a2ui_artifact(
result: Dict[str, Any],
artifact_id: Optional[str] = None,
name: Optional[str] = None,
description: Optional[str] = None,
) -> Artifact:
"""Create an A2A Artifact with a DataPart for an A2UI tool result."""
art_id = artifact_id or f"art-{uuid.uuid4().hex[:12]}"
messages = result.get("messages") or []
a2ui_data = result.get("a2ui_part")
if not isinstance(a2ui_data, dict):
a2ui_data = {"messages": messages}
elif "messages" not in a2ui_data:
a2ui_data = {**a2ui_data, "messages": messages}
return Artifact(
artifact_id=art_id,
name=name,
description=description,
parts=[
DataPart(
data=a2ui_data,
metadata={"mimeType": result.get("mime_type", A2UI_MIME_TYPE)},
)
],
)
def create_response_artifact(
content: str,
chat_history: Optional[List[Dict[str, Any]]] = None,
artifact_id: Optional[str] = None,
name: Optional[str] = None,
description: Optional[str] = None,
) -> Artifact:
"""Build artifact from response text, preferring A2UI DataPart when present in history."""
a2ui = find_a2ui_tool_result(chat_history)
if a2ui:
return create_a2ui_artifact(
a2ui, artifact_id=artifact_id, name=name, description=description
)
return create_artifact(
content, artifact_id=artifact_id, name=name, description=description
)
def create_artifact(
content: str,
artifact_id: Optional[str] = None,
@@ -12,12 +12,9 @@ from praisonaiagents.ui.a2ui.adapter import (
is_a2ui_part,
parse_a2ui_response,
)
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE, A2UIToolResultProtocol
__all__ = [
"A2UI",
"A2UI_MIME_TYPE",
"A2UIToolResultProtocol",
"create_a2ui_part",
"generate_a2ui_system_prompt",
"get_schema_manager",
@@ -12,7 +12,6 @@ from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Set
from praisonaiagents.ui.agui.types import (
BaseEvent,
CustomEvent,
TextMessageStartEvent,
TextMessageContentEvent,
TextMessageEndEvent,
@@ -27,7 +26,6 @@ from praisonaiagents.ui.agui.types import (
StepFinishedEvent,
StateSnapshotEvent,
)
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
@dataclass
@@ -294,41 +292,13 @@ def stream_event_to_agui_events(
if event.type == StreamEventType.TOOL_CALL_RESULT and event.tool_call:
tool_call_id = event.tool_call.get("id") or str(uuid.uuid4())
raw_result = event.tool_call.get("result")
content = event.content if event.content is not None else str(raw_result or "")
content = event.content if event.content is not None else str(event.tool_call.get("result", ""))
buffer.end_tool_call(tool_call_id)
events: List[BaseEvent] = [
create_tool_result_event(tool_call_id, str(content), message_id)
]
if isinstance(raw_result, dict) and raw_result.get("mime_type") == A2UI_MIME_TYPE:
events.append(
CustomEvent(
name="a2ui",
value={
"mime_type": A2UI_MIME_TYPE,
"messages": raw_result.get("messages", []),
"surface_id": _infer_a2ui_surface_id(raw_result),
},
)
)
return events
return [create_tool_result_event(tool_call_id, str(content), message_id)]
return []
def _infer_a2ui_surface_id(result: Dict[str, Any]) -> str:
"""Infer surface id from send_a2ui_messages result (inline, no UI-layer deps)."""
for msg in result.get("messages") or []:
if isinstance(msg, dict):
create = msg.get("createSurface")
if isinstance(create, dict):
sid = create.get("surfaceId") or create.get("id")
if sid:
return str(sid)
sid = result.get("surface_id") or result.get("surfaceId")
return str(sid) if sid else "main"
async def async_stream_agent_response(
agent,
user_input: str,
@@ -133,13 +133,6 @@ class StateSnapshotEvent(BaseEvent):
snapshot: Dict[str, Any]
class CustomEvent(BaseEvent):
"""Custom extension event (e.g. structured A2UI payloads)."""
type: Literal["CUSTOM"] = "CUSTOM"
name: str
value: Any
# Message Types
class FunctionCall(BaseModel):
"""Function call in a tool call."""
@@ -182,7 +175,6 @@ Event = Union[
ToolCallEndEvent,
ToolCallResultEvent,
StateSnapshotEvent,
CustomEvent,
RunStartedEvent,
RunFinishedEvent,
RunErrorEvent,
@@ -1,26 +0,0 @@
"""
UI protocol types — lightweight contracts for agent-to-UI integration.
Zero runtime cost: TypedDict and constants only. No optional dependencies.
"""
from __future__ import annotations
from typing import Any, Dict, List, TypedDict, Union
# Standard MIME type for A2UI payloads (Google A2UI / A2A transport).
A2UI_MIME_TYPE = "application/json+a2ui"
class A2UIToolResultProtocol(TypedDict, total=False):
"""
Output shape from ``send_a2ui_messages`` — the integrator contract for any UI.
Example:
if result.get("mime_type") == A2UI_MIME_TYPE:
render(result["messages"])
"""
mime_type: str
messages: List[Dict[str, Any]]
a2ui_part: Union[Dict[str, Any], List[Any]]
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "1.6.48"
version = "1.6.43"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -233,47 +233,6 @@ class TestA2AJsonRpc:
assert result["status"]["state"] == "completed"
assert result["artifacts"] is not None
assert len(result["artifacts"]) > 0
def test_post_a2a_message_send_a2ui_artifact(self):
"""message/send with A2UI tool result in chat_history → DataPart artifact."""
import json
from fastapi.testclient import TestClient
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
app, a2a = _make_app()
a2ui_payload = {
"mime_type": A2UI_MIME_TYPE,
"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}],
"a2ui_part": {"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}]},
}
a2a.agent.chat = MagicMock(return_value="Surface ready.")
a2a.agent.chat_history = [
{"role": "tool", "content": json.dumps(a2ui_payload)},
]
client = TestClient(app)
response = client.post("/a2a", json={
"jsonrpc": "2.0",
"method": "message/send",
"id": "a2ui-1",
"params": {
"message": {
"messageId": "msg-a2ui",
"role": "user",
"parts": [{"text": "Show UI"}],
}
},
})
assert response.status_code == 200
result = response.json()["result"]
artifact = result["artifacts"][0]
parts = artifact["parts"]
data_parts = [p for p in parts if "data" in p]
assert data_parts, "Expected DataPart in artifact"
assert data_parts[0]["metadata"]["mimeType"] == A2UI_MIME_TYPE
def test_post_a2a_message_send_creates_task_in_store(self):
"""message/send should create a task in TaskStore."""
@@ -1,66 +0,0 @@
"""Tests for A2UI artifact conversion in A2A transport."""
import json
import pytest
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
SAMPLE_A2UI_RESULT = {
"mime_type": A2UI_MIME_TYPE,
"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}],
"a2ui_part": {"messages": [{"createSurface": {"surfaceId": "main", "catalogId": "basic"}}]},
}
class TestFindA2uiToolResult:
def test_finds_last_a2ui_tool_message(self):
from praisonaiagents.ui.a2a.conversion import find_a2ui_tool_result
history = [
{"role": "user", "content": "show ui"},
{"role": "tool", "content": json.dumps({"mime_type": "text/plain", "data": "x"})},
{"role": "tool", "content": json.dumps(SAMPLE_A2UI_RESULT)},
]
result = find_a2ui_tool_result(history)
assert result is not None
assert result["mime_type"] == A2UI_MIME_TYPE
def test_returns_none_without_a2ui(self):
from praisonaiagents.ui.a2a.conversion import find_a2ui_tool_result
assert find_a2ui_tool_result([]) is None
assert find_a2ui_tool_result([{"role": "tool", "content": "plain text"}]) is None
class TestCreateA2uiArtifact:
def test_creates_datapart_with_mime(self):
from praisonaiagents.ui.a2a.conversion import create_a2ui_artifact
from praisonaiagents.ui.a2a.types import DataPart
artifact = create_a2ui_artifact(SAMPLE_A2UI_RESULT, name="ui-surface")
assert artifact.name == "ui-surface"
assert len(artifact.parts) == 1
part = artifact.parts[0]
assert isinstance(part, DataPart)
assert part.metadata.get("mimeType") == A2UI_MIME_TYPE
assert "messages" in part.data
class TestCreateResponseArtifact:
def test_prefers_a2ui_from_history(self):
from praisonaiagents.ui.a2a.conversion import create_response_artifact
from praisonaiagents.ui.a2a.types import DataPart
history = [{"role": "tool", "content": json.dumps(SAMPLE_A2UI_RESULT)}]
artifact = create_response_artifact("Done.", chat_history=history)
assert isinstance(artifact.parts[0], DataPart)
def test_falls_back_to_text(self):
from praisonaiagents.ui.a2a.conversion import create_response_artifact
from praisonaiagents.ui.a2a.types import TextPart
artifact = create_response_artifact("Hello", chat_history=[])
assert isinstance(artifact.parts[0], TextPart)
assert artifact.parts[0].text == "Hello"
@@ -1,51 +0,0 @@
"""Tests for send_a2ui_messages minimal arg unwrap."""
import json
import pytest
from praisonaiagents.tools.a2ui_tools import _unwrap_messages
try:
import a2ui # noqa: F401
HAS_A2UI_SDK = True
except ImportError:
HAS_A2UI_SDK = False
SAMPLE_MSG = {"createSurface": {"surfaceId": "main", "catalogId": "basic"}}
class TestUnwrapMessages:
def test_list_input(self):
assert _unwrap_messages([SAMPLE_MSG]) == [SAMPLE_MSG]
def test_json_string_list(self):
assert _unwrap_messages(json.dumps([SAMPLE_MSG])) == [SAMPLE_MSG]
def test_wrapped_messages_dict(self):
assert _unwrap_messages({"messages": [SAMPLE_MSG]}) == [SAMPLE_MSG]
def test_json_string_wrapped_object(self):
assert _unwrap_messages(json.dumps({"messages": [SAMPLE_MSG]})) == [SAMPLE_MSG]
def test_single_message_dict(self):
assert _unwrap_messages(SAMPLE_MSG) == [SAMPLE_MSG]
def test_invalid_json_raises(self):
with pytest.raises(ValueError, match="not valid JSON"):
_unwrap_messages("{not json")
def test_invalid_shape_raises(self):
with pytest.raises(ValueError, match="must be"):
_unwrap_messages(123)
@pytest.mark.skipif(not HAS_A2UI_SDK, reason="a2ui-agent-sdk not installed")
class TestSendA2uiMessagesIntegration:
def test_full_tool_call(self):
from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
result = send_a2ui_messages(messages=[SAMPLE_MSG])
assert result["mime_type"] == "application/json+a2ui"
assert len(result["messages"]) == 1
@@ -34,27 +34,3 @@ class TestA2UITool:
)
assert result["mime_type"] == "application/json+a2ui"
assert len(result["messages"]) == 1
class TestA2UIAdapterSmoke:
def test_is_a2ui_part_with_sdk(self):
pytest.importorskip("a2ui", reason="a2ui-agent-sdk not installed")
from praisonaiagents.ui.a2ui.adapter import create_a2ui_part, is_a2ui_part
part = create_a2ui_part({"messages": [{"createSurface": {"surfaceId": "m"}}]})
assert is_a2ui_part(part) is True
def test_missing_sdk_import_error_message(self):
from unittest.mock import patch
from praisonaiagents.ui.a2ui import adapter
def _raise(_name):
raise ImportError(
"A2UI support requires the optional dependency. "
"Install with: pip install praisonaiagents[a2ui]"
)
with patch.object(adapter, "_import", side_effect=_raise):
with pytest.raises(ImportError, match="praisonaiagents\\[a2ui\\]"):
adapter.create_a2ui_part({"messages": []})
@@ -1,27 +0,0 @@
"""Tests for UI protocol types (zero-dep contracts)."""
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE, A2UIToolResultProtocol
class TestA2UIProtocols:
def test_mime_type_constant(self):
assert A2UI_MIME_TYPE == "application/json+a2ui"
def test_tool_result_protocol_shape(self):
result: A2UIToolResultProtocol = {
"mime_type": A2UI_MIME_TYPE,
"messages": [{"createSurface": {"surfaceId": "main"}}],
"a2ui_part": {"messages": []},
}
assert result["mime_type"] == A2UI_MIME_TYPE
def test_exported_from_a2ui_package(self):
from praisonaiagents.ui.a2ui import A2UI_MIME_TYPE, A2UIToolResultProtocol
assert A2UI_MIME_TYPE == "application/json+a2ui"
assert A2UIToolResultProtocol is not None
def test_top_level_lazy_a2ui_export(self):
from praisonaiagents import A2UI
assert A2UI is not None
@@ -129,29 +129,6 @@ class TestStreamEventBridge:
assert agui_events[0].type == "TOOL_CALL_START"
assert agui_events[0].tool_call_name == "search"
def test_tool_call_result_a2ui_emits_custom_event(self):
from praisonaiagents.streaming.events import StreamEvent, StreamEventType
from praisonaiagents.ui.agui.streaming import EventBuffer, stream_event_to_agui_events
from praisonaiagents.ui.protocols import A2UI_MIME_TYPE
buffer = EventBuffer()
a2ui_result = {
"mime_type": A2UI_MIME_TYPE,
"messages": [{"createSurface": {"surfaceId": "panel-1"}}],
}
event = StreamEvent(
type=StreamEventType.TOOL_CALL_RESULT,
tool_call={"id": "tc-a2ui", "name": "send_a2ui_messages", "result": a2ui_result},
)
agui_events = stream_event_to_agui_events(event, "msg-1", buffer)
assert len(agui_events) == 2
assert agui_events[0].type == "TOOL_CALL_RESULT"
assert agui_events[1].type == "CUSTOM"
assert agui_events[1].name == "a2ui"
assert agui_events[1].value["surface_id"] == "panel-1"
assert agui_events[1].value["messages"] == a2ui_result["messages"]
class TestRunLifecycleEvents:
"""Test run lifecycle events."""
@@ -349,161 +349,6 @@ class TestDbSessionAdapter:
assert meta["agent_id"] == "abc"
assert meta["total_input_tokens"] == 42
def test_clear_session_purges_db_history(self):
"""clear_session must not leave stale messages reloadable from the DB."""
from praisonai.integrations.db_session_adapter import DbSessionAdapter
from praisonaiagents.db.protocol import DbMessage
persisted: list = []
mock_db = MagicMock()
def _on_start(agent_name, session_id):
return list(persisted)
def _on_user(sid, content, **kw):
persisted.append(DbMessage(role="user", content=content))
def _on_assistant(sid, content, **kw):
persisted.append(DbMessage(role="assistant", content=content))
mock_db.on_agent_start.side_effect = _on_start
mock_db.on_user_message.side_effect = _on_user
mock_db.on_agent_message.side_effect = _on_assistant
conv = MagicMock()
conv.delete_messages.side_effect = lambda sid, ids: persisted.clear()
mock_db.conversation = conv
adapter = DbSessionAdapter(mock_db)
sid = "bot-trim-session"
adapter.add_message(sid, "user", "old")
adapter.add_message(sid, "assistant", "old-reply")
assert len(persisted) == 2
adapter.clear_session(sid)
conv.delete_messages.assert_called_once_with(sid, None)
assert persisted == []
assert adapter.get_chat_history(sid) == []
adapter.add_message(sid, "user", "new-only")
assert len(adapter.get_chat_history(sid)) == 1
assert len(persisted) == 1
adapter2 = DbSessionAdapter(mock_db)
assert adapter2.get_chat_history(sid) == [{"role": "user", "content": "new-only"}]
def test_metadata_survives_adapter_restart(self):
"""set_metadata must persist to the conversation store for resume after restart."""
from praisonai.integrations.db_session_adapter import DbSessionAdapter
from praisonai.persistence.conversation.base import ConversationSession
sessions: dict = {}
def _get_session(sid):
return sessions.get(sid)
def _create_session(session):
sessions[session.session_id] = session
def _update_session(session):
sessions[session.session_id] = session
conv = MagicMock()
conv.get_session.side_effect = _get_session
conv.create_session.side_effect = _create_session
conv.update_session.side_effect = _update_session
mock_db = MagicMock()
mock_db.on_agent_start.return_value = []
mock_db.conversation = None # Prevent auto-creation of conversation attr
mock_db._conversation_store = conv
sid = "meta-restart-session"
adapter1 = DbSessionAdapter(mock_db)
adapter1.set_metadata(
sid,
{
"agent_id": "agent-xyz",
"total_input_tokens": 99,
"session_history": [{"turn": 1}],
},
)
adapter2 = DbSessionAdapter(mock_db)
meta = adapter2.get_metadata(sid)
assert meta["agent_id"] == "agent-xyz"
assert meta["total_input_tokens"] == 99
assert meta["session_history"] == [{"turn": 1}]
def test_managed_agent_metadata_persistence_e2e(self):
"""Integration test: ManagedAgent metadata survives restart via real adapter."""
import tempfile
import os
from praisonaiagents.managed import LocalManagedAgent
from praisonai.integrations.db_session_adapter import DbSessionAdapter
# Create a real SQLite database for this test
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp:
db_path = tmp.name
try:
# Mock PraisonDB that uses file-based conversation store
mock_db = MagicMock()
mock_db.on_agent_start.return_value = []
# Real conversation store backed by file
from praisonai.persistence.conversation.base import ConversationSession
from praisonai.persistence.conversation.file_store import FileConversationStore
conv_store = FileConversationStore(store_path=db_path + "_conv.json")
mock_db.conversation = None
mock_db._conversation_store = conv_store
# Create and run agent with metadata
adapter = DbSessionAdapter(mock_db)
agent = LocalManagedAgent(
name="test-agent",
instructions="You are a helpful assistant.",
session_store=adapter,
llm="gpt-4o-mini"
)
# Start agent and collect session metadata
session_id = "integration-test-session"
adapter._ensure_session(session_id, agent.name)
# Set some metadata that should persist
test_metadata = {
"agent_id": "test-agent-123",
"total_input_tokens": 150,
"session_history": [{"turn": 1, "action": "started"}],
"compute_instance": "test-instance-1"
}
adapter.set_metadata(session_id, test_metadata)
# Simulate process restart with new adapter and agent
adapter2 = DbSessionAdapter(mock_db)
agent2 = LocalManagedAgent(
name="test-agent",
instructions="You are a helpful assistant.",
session_store=adapter2,
llm="gpt-4o-mini"
)
# Metadata should be restored from persistence
restored_meta = adapter2.get_metadata(session_id)
assert restored_meta["agent_id"] == "test-agent-123"
assert restored_meta["total_input_tokens"] == 150
assert restored_meta["session_history"] == [{"turn": 1, "action": "started"}]
assert restored_meta["compute_instance"] == "test-instance-1"
finally:
# Clean up temp files
for path in [db_path, db_path + "_conv.json"]:
if os.path.exists(path):
os.unlink(path)
# ===========================================================================
# 7. ManagedAgent factory wiring
@@ -1,215 +0,0 @@
"""Test tool schema parity and OpenAI compatibility.
This module tests the fixes for issue #1696 to ensure BaseTool ↔ OpenAI-compatible
tool schema parity in chat/agent dispatch.
"""
import pytest
import json
from typing import Any, Dict, List
from praisonaiagents.tools.base import BaseTool, ToolValidationError, validate_tool_schema_consistency
from praisonaiagents.tools.decorator import tool, FunctionTool
class MockTool(BaseTool):
"""A simple mock tool for testing."""
name = "mock_tool"
description = "A mock tool for testing"
def run(self, query: str, count: int = 1) -> str:
return f"Mock result for {query} (count: {count})"
class TestBasicToolSchemaValidation:
"""Test basic tool schema validation and OpenAI compatibility."""
def test_base_tool_schema_structure(self):
"""Test that BaseTool generates correct OpenAI schema structure."""
tool = MockTool()
schema = tool.get_schema()
# Validate top-level structure
assert isinstance(schema, dict)
assert schema["type"] == "function"
assert "function" in schema
# Validate function structure
func = schema["function"]
assert func["name"] == "mock_tool"
assert func["description"] == "A mock tool for testing"
assert "parameters" in func
# Validate parameters structure
params = func["parameters"]
assert params["type"] == "object"
assert "properties" in params
assert "required" in params
# Validate specific properties
props = params["properties"]
assert "query" in props
assert "count" in props
assert props["query"]["type"] == "string"
assert props["count"]["type"] == "integer"
# Validate required fields
assert "query" in params["required"]
assert "count" not in params["required"] # Has default value
def test_tool_schema_json_serialization(self):
"""Test that tool schemas can be JSON serialized (round-trip test)."""
tool = MockTool()
schema = tool.get_schema()
# Should serialize without errors
serialized = json.dumps(schema)
assert isinstance(serialized, str)
# Should deserialize back to same structure
deserialized = json.loads(serialized)
assert deserialized == schema
def test_tool_validation_passes(self):
"""Test that valid tools pass validation."""
tool = MockTool()
# Basic validation should pass
assert tool.validate() == True
# Round-trip validation should pass
assert tool.validate_schema_roundtrip() == True
def test_function_tool_schema_parity(self):
"""Test that @tool decorated functions have same schema structure."""
@tool
def test_function(query: str, count: int = 1) -> str:
"""A test function for schema validation."""
return f"Result for {query} (count: {count})"
schema = test_function.get_schema()
# Should have same structure as BaseTool
assert schema["type"] == "function"
func = schema["function"]
assert func["name"] == "test_function"
assert "parameters" in func
params = func["parameters"]
assert params["type"] == "object"
assert "properties" in params
assert "required" in params
class TestToolSchemaValidationErrors:
"""Test that validation catches schema errors."""
def test_invalid_tool_missing_properties(self):
"""Test validation fails for tools with invalid parameters."""
class BadTool(BaseTool):
name = "bad_tool"
description = "Bad tool"
parameters = {"type": "object"} # Missing properties
def run(self, query: str) -> str:
return query
tool = BadTool()
with pytest.raises(ToolValidationError, match="must have a 'properties' field"):
tool.validate()
def test_invalid_tool_missing_type(self):
"""Test validation fails for tools with missing type."""
class BadTool(BaseTool):
name = "bad_tool"
description = "Bad tool"
parameters = {"properties": {}} # Missing type
def run(self, query: str) -> str:
return query
tool = BadTool()
with pytest.raises(ToolValidationError, match="must have a 'type' field"):
tool.validate()
class TestToolListValidation:
"""Test validation of tool lists for consistency."""
def test_valid_tool_list(self):
"""Test that valid tool lists pass consistency checks."""
@tool
def tool_one(x: str) -> str:
"""First tool."""
return x
@tool
def tool_two(y: int) -> int:
"""Second tool."""
return y
tool_three = MockTool()
tools = [tool_one, tool_two, tool_three]
# Should pass validation
assert validate_tool_schema_consistency(tools) == True
def test_duplicate_tool_names_error(self):
"""Test that duplicate tool names are caught."""
@tool(name="duplicate_name")
def tool_one(x: str) -> str:
return x
@tool(name="duplicate_name")
def tool_two(y: int) -> int:
return y
tools = [tool_one, tool_two]
with pytest.raises(ToolValidationError, match="Duplicate tool name"):
validate_tool_schema_consistency(tools)
def test_empty_tool_list(self):
"""Test that empty tool lists are valid."""
assert validate_tool_schema_consistency([]) == True
class TestRealAgenticScenario:
"""Test real agentic scenarios to ensure end-to-end compatibility."""
def test_tool_schema_agent_integration(self):
"""Test that tools work with agent in real scenario."""
from praisonaiagents import Agent
@tool
def echo_tool(message: str) -> str:
"""Echo the input message."""
return f"Echo: {message}"
# Create agent with tool - should not raise validation errors
agent = Agent(
name="test_agent",
instructions="You are a helpful assistant",
tools=[echo_tool]
)
# Schema should be valid for LLM use
schema = echo_tool.get_schema()
assert schema["type"] == "function"
# Should be JSON serializable
json.dumps(schema)
# Agent should format tools properly
formatted_tools = agent._format_tools_for_completion(agent.tools)
assert len(formatted_tools) == 1
assert formatted_tools[0]["type"] == "function"
print("✅ Real agentic test passed: Tool schema integration works end-to-end")
if __name__ == "__main__":
pytest.main([__file__, "-v"])
@@ -1,82 +0,0 @@
"""Tests for Agent.__deepcopy__ and Agent.clone_for_channel() (issue #1746).
Validates:
- copy.deepcopy(agent) works without threading.RLock errors
- clone_for_channel() returns an independent instance
- Cloned agent has its own lock objects (not shared)
- Key configuration attributes are preserved in the clone
- Multiple clones are isolated from each other
"""
import copy
import threading
import pytest
class TestAgentDeepCopy:
"""Test Agent.__deepcopy__ and clone_for_channel()."""
def _make_agent(self, **kwargs):
from praisonaiagents.agent.agent import Agent
return Agent(name="TestAgent", instructions="test", **kwargs)
def test_deepcopy_does_not_raise(self):
"""copy.deepcopy(agent) must not raise TypeError for RLock."""
agent = self._make_agent()
cloned = copy.deepcopy(agent)
assert cloned is not agent
def test_clone_for_channel_returns_new_instance(self):
agent = self._make_agent()
cloned = agent.clone_for_channel()
assert cloned is not agent
def test_clone_has_independent_cache_lock(self):
"""__cache_lock must be a new RLock, not shared with original."""
agent = self._make_agent()
cloned = agent.clone_for_channel()
assert cloned._Agent__cache_lock is not agent._Agent__cache_lock
assert isinstance(cloned._Agent__cache_lock, type(threading.RLock()))
def test_clone_has_independent_cost_lock(self):
"""_cost_lock must be a new Lock, not shared with original."""
agent = self._make_agent()
cloned = agent.clone_for_channel()
assert cloned._cost_lock is not agent._cost_lock
def test_clone_preserves_name(self):
agent = self._make_agent(name="MyAgent")
assert agent.clone_for_channel().name == "MyAgent"
def test_clone_preserves_instructions(self):
agent = self._make_agent(instructions="Do the thing")
assert agent.clone_for_channel().instructions == "Do the thing"
def test_clone_preserves_role_goal_backstory(self):
agent = self._make_agent(
role="Analyst",
goal="Analyse",
backstory="Expert analyst",
)
cloned = agent.clone_for_channel()
assert cloned.role == "Analyst"
assert cloned.goal == "Analyse"
assert cloned.backstory == "Expert analyst"
def test_clone_preserves_llm(self):
agent = self._make_agent(llm="gpt-4o-mini")
assert agent.clone_for_channel().llm == "gpt-4o-mini"
def test_multiple_clones_are_isolated(self):
"""Regression: creating two clones (simulating 2nd+ gateway channel) must work."""
agent = self._make_agent()
clone1 = agent.clone_for_channel()
clone2 = agent.clone_for_channel()
assert clone1 is not clone2
assert clone1._Agent__cache_lock is not clone2._Agent__cache_lock
def test_deepcopy_multiple_times(self):
"""Successive deepcopies must all succeed (no cumulative state corruption)."""
agent = self._make_agent()
for _ in range(3):
cloned = copy.deepcopy(agent)
assert cloned is not agent
@@ -1,45 +0,0 @@
from types import SimpleNamespace
import sys
def test_start_verbose_composed_callbacks_forward_all_arguments(monkeypatch):
from praisonaiagents import Agent, AgentTeam
on_start_calls = []
on_complete_calls = []
debug_messages = []
def on_task_start(task, task_id):
on_start_calls.append((task, task_id))
def on_task_complete(task, task_output):
on_complete_calls.append((task, task_output))
team = AgentTeam(
agents=[
Agent(name="Researcher", instructions="Research"),
Agent(name="Writer", instructions="Write"),
],
hooks={
"on_task_start": on_task_start,
"on_task_complete": on_task_complete,
},
)
def fake_run_all_tasks():
task_id, task = next(iter(team.tasks.items()))
task_output = SimpleNamespace(raw="done")
team.on_task_start(task, task_id)
team.on_task_complete(task, task_output)
monkeypatch.setattr(team, "run_all_tasks", fake_run_all_tasks)
monkeypatch.setattr(sys.stdout, "isatty", lambda: True)
monkeypatch.setattr("praisonaiagents.agents.agents.logging.debug", lambda msg: debug_messages.append(str(msg)))
team.start(output="verbose")
assert len(on_start_calls) == 1
assert len(on_complete_calls) == 1
assert on_start_calls[0][1] == next(iter(team.tasks.keys()))
assert on_complete_calls[0][1].raw == "done"
assert all("Error in verbose task" not in message for message in debug_messages)
@@ -5,8 +5,6 @@ Tests Mem0Adapter lazy loading and normalization.
"""
import sys
import importlib
from types import SimpleNamespace
import pytest
from unittest.mock import MagicMock
@@ -176,54 +174,3 @@ class TestProtocolCompliance:
for method in required_methods:
assert hasattr(Mem0Adapter, method), f"Missing method: {method}"
class TestChromaKnowledgeAdapterWhereFilters:
"""Tests for ChromaKnowledgeAdapter where-filter formatting."""
def _make_adapter(self):
from praisonaiagents.knowledge.adapters.factories import ChromaKnowledgeAdapter
adapter = ChromaKnowledgeAdapter.__new__(ChromaKnowledgeAdapter)
adapter.collection = MagicMock()
adapter.collection.query.return_value = {
"ids": [[]],
"documents": [[]],
"metadatas": [[]],
"distances": [[]],
}
return adapter
def test_search_wraps_multiple_where_filters_with_and(self, monkeypatch):
embedding_module = importlib.import_module("praisonaiagents.embedding")
adapter = self._make_adapter()
monkeypatch.setattr(
embedding_module,
"embedding",
lambda *args, **kwargs: SimpleNamespace(embeddings=[[0.1, 0.2]]),
raising=False,
)
adapter.search("q", user_id="u1", agent_id="a1", run_id="r1")
where = adapter.collection.query.call_args.kwargs["where"]
assert where == {
"$and": [{"user_id": "u1"}, {"agent_id": "a1"}, {"run_id": "r1"}]
}
def test_search_keeps_single_where_filter_unwrapped(self, monkeypatch):
embedding_module = importlib.import_module("praisonaiagents.embedding")
adapter = self._make_adapter()
monkeypatch.setattr(
embedding_module,
"embedding",
lambda *args, **kwargs: SimpleNamespace(embeddings=[[0.1, 0.2]]),
raising=False,
)
adapter.search("q", user_id="u1")
where = adapter.collection.query.call_args.kwargs["where"]
assert where == {"user_id": "u1"}
@@ -223,188 +223,3 @@ class TestDefaultContextBuilder:
context = builder.build(results)
assert "\n\n" in context
class TestExtractValue:
"""Tests for the _extract_value helper function."""
def test_extract_value_from_dict(self):
"""Test extracting value from dictionary."""
from praisonaiagents.rag.context import _extract_value
item = {"text": "test content", "metadata": {"source": "test.pdf"}}
assert _extract_value(item, "text") == "test content"
assert _extract_value(item, "metadata")["source"] == "test.pdf"
assert _extract_value(item, "nonexistent", "default") == "default"
def test_extract_value_from_searchresultitem(self):
"""Test extracting value from SearchResultItem object."""
from praisonaiagents.rag.context import _extract_value
from praisonaiagents.knowledge.models import SearchResultItem
item = SearchResultItem(
text="test content",
source="test.pdf",
filename="test.pdf",
metadata={"extra": "data"}
)
assert _extract_value(item, "text") == "test content"
assert _extract_value(item, "source") == "test.pdf"
assert _extract_value(item, "filename") == "test.pdf"
assert _extract_value(item, "metadata")["extra"] == "data"
assert _extract_value(item, "nonexistent", "default") == "default"
class TestExtractMetadataValue:
"""Tests for the _extract_metadata_value helper function."""
def test_extract_metadata_value_from_metadata(self):
"""Test extracting value from metadata dict."""
from praisonaiagents.rag.context import _extract_metadata_value
from praisonaiagents.knowledge.models import SearchResultItem
item = SearchResultItem(text="content", source="fallback.pdf")
metadata = {"source": "metadata.pdf"}
# Should prefer metadata over top-level
result = _extract_metadata_value(item, metadata, "source", "")
assert result == "metadata.pdf"
def test_extract_metadata_value_fallback_to_toplevel(self):
"""Test fallback to top-level when metadata doesn't have key."""
from praisonaiagents.rag.context import _extract_metadata_value
from praisonaiagents.knowledge.models import SearchResultItem
item = SearchResultItem(text="content", source="toplevel.pdf")
metadata = {} # Empty metadata
# Should fall back to top-level source
result = _extract_metadata_value(item, metadata, "source", "")
assert result == "toplevel.pdf"
def test_extract_metadata_value_with_default(self):
"""Test fallback to default when neither metadata nor top-level have key."""
from praisonaiagents.rag.context import _extract_metadata_value
from praisonaiagents.knowledge.models import SearchResultItem
item = SearchResultItem(text="content") # No source
metadata = {} # No source in metadata
# Should use default
result = _extract_metadata_value(item, metadata, "source", "default.pdf")
assert result == "default.pdf"
class TestSearchResultItemSupport:
"""Tests for SearchResultItem object support in context functions."""
def test_deduplicate_chunks_with_searchresultitem(self):
"""Test deduplication with SearchResultItem objects."""
from praisonaiagents.rag.context import deduplicate_chunks
from praisonaiagents.knowledge.models import SearchResultItem
results = [
SearchResultItem(text="Same content", source="a.pdf"),
SearchResultItem(text="Same content", source="a.pdf"), # Duplicate
SearchResultItem(text="Different content", source="b.pdf"),
]
deduped = deduplicate_chunks(results)
assert len(deduped) == 2 # One duplicate should be removed
def test_deduplicate_chunks_mixed_formats(self):
"""Test deduplication with mixed dict and SearchResultItem formats."""
from praisonaiagents.rag.context import deduplicate_chunks
from praisonaiagents.knowledge.models import SearchResultItem
results = [
{"text": "Content A", "metadata": {"source": "a.pdf"}},
SearchResultItem(text="Content A", source="a.pdf"), # Should be deduped
SearchResultItem(text="Content B", source="b.pdf"),
]
deduped = deduplicate_chunks(results)
assert len(deduped) == 2 # One duplicate should be removed
def test_deduplicate_with_top_level_source(self):
"""Test deduplication considers top-level source from SearchResultItem."""
from praisonaiagents.rag.context import deduplicate_chunks
from praisonaiagents.knowledge.models import SearchResultItem
results = [
SearchResultItem(text="Same text", source="source1.pdf"),
SearchResultItem(text="Same text", source="source2.pdf"),
]
deduped = deduplicate_chunks(results)
assert len(deduped) == 2 # Different sources should not be deduped
def test_build_context_with_searchresultitem(self):
"""Test building context with SearchResultItem objects."""
from praisonaiagents.rag.context import build_context
from praisonaiagents.knowledge.models import SearchResultItem
results = [
SearchResultItem(
text="First content",
source="source1.pdf",
filename="file1.pdf"
),
SearchResultItem(
text="Second content",
source="source2.pdf",
filename="file2.pdf"
),
]
context, used = build_context(results)
assert "First content" in context
assert "Second content" in context
assert "file1.pdf" in context # Should use filename
assert "file2.pdf" in context
assert len(used) == 2
def test_build_context_source_fallback(self):
"""Test that build_context falls back to top-level source when metadata is empty."""
from praisonaiagents.rag.context import build_context
from praisonaiagents.knowledge.models import SearchResultItem
# SearchResultItem with source but empty metadata
results = [
SearchResultItem(
text="Content with source",
source="fallback.pdf",
metadata={} # Empty metadata, should fall back to top-level source
)
]
context, used = build_context(results, include_source=True)
assert "Content with source" in context
assert "fallback.pdf" in context # Should use top-level source
assert len(used) == 1
def test_build_context_mixed_formats_with_sources(self):
"""Test building context with mixed dict/SearchResultItem formats."""
from praisonaiagents.rag.context import build_context
from praisonaiagents.knowledge.models import SearchResultItem
results = [
{"text": "Dict content", "metadata": {"filename": "dict.pdf"}},
SearchResultItem(
text="Object content",
filename="object.pdf",
metadata={}
),
]
context, used = build_context(results, include_source=True)
assert "Dict content" in context
assert "Object content" in context
assert "dict.pdf" in context
assert "object.pdf" in context # Should use top-level filename
assert len(used) == 2
@@ -8,7 +8,6 @@ from praisonaiagents.rag.context import (
deduplicate_chunks,
build_context,
)
from praisonaiagents.knowledge.models import SearchResultItem
class TestDeduplicateChunksNormalization:
@@ -123,25 +122,3 @@ class TestEdgeCases:
context, used = build_context(results)
assert len(used) == 1
assert "valid content" in context
class TestSearchResultItemCompatibility:
"""Tests for SearchResultItem object compatibility."""
def test_deduplicate_uses_source_attribute_fallback(self):
"""Different source attributes should produce unique chunks."""
results = [
SearchResultItem(text="same content", source="a.pdf"),
SearchResultItem(text="same content", source="b.pdf"),
]
deduped = deduplicate_chunks(results)
assert len(deduped) == 2
def test_build_context_uses_filename_attribute_fallback(self):
"""Source label should use object filename/source attributes."""
results = [
SearchResultItem(text="chunk", filename="doc.md"),
]
context, used = build_context(results, include_source=True)
assert len(used) == 1
assert "doc.md" in context
@@ -103,35 +103,6 @@ class TestHierarchicalSessionStore:
"""Clean up test fixtures."""
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_add_message_preserves_concurrent_writes(self):
"""Concurrent add_message must not lose messages (same as DefaultSessionStore)."""
with tempfile.TemporaryDirectory() as tmpdir:
writer = HierarchicalSessionStore(session_dir=tmpdir)
reader = HierarchicalSessionStore(session_dir=tmpdir)
writer.add_user_message("session-1", "first")
reader._load_session("session-1")
writer.add_user_message("session-1", "second")
history = writer.get_chat_history("session-1")
assert len(history) == 2
assert history[1]["content"] == "second"
def test_update_session_metadata_preserves_extended_fields(self):
"""Metadata updates must not strip parent_id, snapshots, etc."""
session_id = self.store.create_session(title="Parent")
self.store.add_message(session_id, "user", "hello")
parent_id = self.store.create_session(title="Child", parent_id=session_id)
snapshot_id = self.store.create_snapshot(session_id, label="checkpoint")
assert self.store.update_session_metadata(session_id, model="gpt-4o-mini")
session = self.store.get_extended_session(session_id)
assert session.metadata.get("model") == "gpt-4o-mini"
assert session.title == "Parent"
assert parent_id in session.children_ids
assert any(s.id == snapshot_id for s in session.snapshots)
def test_create_session(self):
"""Test creating a session."""
session_id = self.store.create_session(title="Test Session")
@@ -282,101 +282,6 @@ class TestDefaultSessionStore:
session = temp_store.get_session("session-1")
assert session.agent_name == "TestBot"
assert session.user_id == "user-123"
def test_update_session_metadata_preserves_messages(self, temp_store):
"""Metadata updates must not drop messages added by another store instance."""
with tempfile.TemporaryDirectory() as tmpdir:
writer = DefaultSessionStore(session_dir=tmpdir)
reader = DefaultSessionStore(session_dir=tmpdir)
writer.add_user_message("session-1", "first")
reader._load_session("session-1")
writer.add_user_message("session-1", "second")
assert reader.update_session_metadata(
"session-1", model="gpt-4o-mini", total_tokens=42
)
writer.invalidate_cache("session-1")
history = writer.get_chat_history("session-1")
assert len(history) == 2
assert history[1]["content"] == "second"
session = writer.get_session("session-1")
assert session.metadata.get("model") == "gpt-4o-mini"
assert session.metadata.get("total_tokens") == 42
def test_set_agent_info_preserves_messages(self, temp_store):
"""Agent info updates must not drop messages added by another store instance."""
with tempfile.TemporaryDirectory() as tmpdir:
writer = DefaultSessionStore(session_dir=tmpdir)
reader = DefaultSessionStore(session_dir=tmpdir)
writer.add_user_message("session-1", "first")
reader._load_session("session-1")
writer.add_user_message("session-1", "second")
assert reader.set_agent_info("session-1", agent_name="TestBot", user_id="u-1")
writer.invalidate_cache("session-1")
history = writer.get_chat_history("session-1")
assert len(history) == 2
assert history[1]["content"] == "second"
session = writer.get_session("session-1")
assert session.agent_name == "TestBot"
assert session.user_id == "u-1"
def test_get_chat_history_sees_writes_from_other_store(self, temp_store):
"""Reads must reload from disk, not a stale in-memory cache."""
with tempfile.TemporaryDirectory() as tmpdir:
writer = DefaultSessionStore(session_dir=tmpdir)
reader = DefaultSessionStore(session_dir=tmpdir)
writer.add_user_message("session-1", "first")
reader._load_session("session-1")
writer.add_user_message("session-1", "second")
history = reader.get_chat_history("session-1")
assert len(history) == 2
assert history[1]["content"] == "second"
def test_clear_session_preserves_new_messages(self, temp_store):
"""Clear must reload from disk so concurrent adds are not lost."""
with tempfile.TemporaryDirectory() as tmpdir:
writer = DefaultSessionStore(session_dir=tmpdir)
reader = DefaultSessionStore(session_dir=tmpdir)
writer.add_user_message("session-1", "first")
reader._load_session("session-1")
writer.add_user_message("session-1", "second")
assert reader.clear_session("session-1")
writer.invalidate_cache("session-1")
history = writer.get_chat_history("session-1")
assert len(history) == 0
def test_set_gateway_info_preserves_messages(self, temp_store):
"""Gateway info updates must not drop messages added by another store instance."""
with tempfile.TemporaryDirectory() as tmpdir:
writer = DefaultSessionStore(session_dir=tmpdir)
reader = DefaultSessionStore(session_dir=tmpdir)
writer.add_user_message("session-1", "first")
reader._load_session("session-1")
writer.add_user_message("session-1", "second")
assert reader.set_gateway_info("session-1", gateway_session_id="gw-123", agent_id="agent-456")
writer.invalidate_cache("session-1")
history = writer.get_chat_history("session-1")
assert len(history) == 2
assert history[1]["content"] == "second"
session = writer.get_session("session-1")
assert session.gateway_session_id == "gw-123"
assert session.agent_id == "agent-456"
def test_concurrent_writes(self, temp_store):
"""Test concurrent writes to same session."""
@@ -408,7 +313,7 @@ class TestDefaultSessionStore:
assert history == []
def test_invalidate_cache(self, temp_store):
"""Test that reads always see latest disk state."""
"""Test cache invalidation."""
temp_store.add_user_message("session-1", "Hello")
# Modify file directly
@@ -419,11 +324,11 @@ class TestDefaultSessionStore:
with open(filepath, "w") as f:
json.dump(data, f)
# Reads always reload from disk (no stale cache)
# Without invalidation, cache returns old data
history = temp_store.get_chat_history("session-1")
assert len(history) == 2 # Fresh from disk
assert len(history) == 1 # Cached
# invalidate_cache still works but is now essentially a no-op for reads
# After invalidation, new data is loaded
temp_store.invalidate_cache("session-1")
history = temp_store.get_chat_history("session-1")
assert len(history) == 2
@@ -1,311 +0,0 @@
"""Tests for skill capability validation."""
import pytest
import os
from pathlib import Path
from unittest.mock import Mock, patch
from praisonaiagents.skills.models import SkillProperties, SkillRequirements, SkillState
from praisonaiagents.skills.capability_validator import (
CapabilityValidator,
EnforcementLevel,
ValidationResult
)
class TestSkillRequirements:
"""Test SkillRequirements parsing and normalization."""
def test_from_frontmatter_empty(self):
"""Test parsing empty frontmatter."""
requirements = SkillRequirements.from_frontmatter({})
assert requirements.is_empty()
assert not requirements # Test __bool__
def test_from_frontmatter_tools_list(self):
"""Test parsing tools from list."""
metadata = {"requires_tools": ["web_search", "file_write"]}
requirements = SkillRequirements.from_frontmatter(metadata)
assert requirements.tools == ["web_search", "file_write"]
assert requirements.servers == []
assert not requirements.is_empty()
def test_from_frontmatter_tools_string(self):
"""Test parsing tools from string."""
metadata = {"requires-tools": "web_search file_write"}
requirements = SkillRequirements.from_frontmatter(metadata)
assert requirements.tools == ["web_search", "file_write"]
def test_from_frontmatter_backward_compatibility(self):
"""Test backward compatibility with allowed-tools."""
metadata = {"allowed-tools": ["existing_tool"]}
requirements = SkillRequirements.from_frontmatter(metadata)
assert requirements.tools == ["existing_tool"]
def test_from_frontmatter_servers(self):
"""Test parsing server requirements."""
metadata = {"requires_servers": ["mcp:filesystem", "http:internal"]}
requirements = SkillRequirements.from_frontmatter(metadata)
assert requirements.servers == ["mcp:filesystem", "http:internal"]
def test_from_frontmatter_env_vars(self):
"""Test parsing environment variable requirements."""
metadata = {"requires_env": ["API_KEY", "SECRET_TOKEN"]}
requirements = SkillRequirements.from_frontmatter(metadata)
assert requirements.env_vars == ["API_KEY", "SECRET_TOKEN"]
def test_from_frontmatter_openclaw_hints(self):
"""Test parsing OpenClaw hints."""
metadata = {"openclaw": {"version": "1.0", "compatibility": "hermes"}}
requirements = SkillRequirements.from_frontmatter(metadata)
assert requirements.openclaw_hints == {"version": "1.0", "compatibility": "hermes"}
def test_normalize_list(self):
"""Test list normalization utility."""
assert SkillRequirements._normalize_list("a b c") == ["a", "b", "c"]
assert SkillRequirements._normalize_list("a,b,c") == ["a", "b", "c"]
assert SkillRequirements._normalize_list("a, b c") == ["a", "b", "c"]
assert SkillRequirements._normalize_list(["a", "b", "c"]) == ["a", "b", "c"]
assert SkillRequirements._normalize_list("") == []
assert SkillRequirements._normalize_list([]) == []
assert SkillRequirements._normalize_list(123) == []
class TestCapabilityValidator:
"""Test capability validation logic."""
def test_enforcement_levels(self):
"""Test different enforcement levels."""
assert EnforcementLevel.DISABLED.value == "disabled"
assert EnforcementLevel.WARN.value == "warn"
assert EnforcementLevel.STRICT.value == "strict"
def test_validator_initialization(self):
"""Test validator initialization."""
validator = CapabilityValidator()
assert validator.enforcement_level == EnforcementLevel.WARN
validator = CapabilityValidator(EnforcementLevel.STRICT)
assert validator.enforcement_level == EnforcementLevel.STRICT
def test_validate_skill_no_requirements(self):
"""Test validation of skill with no requirements."""
skill = SkillProperties(
name="test-skill",
description="Test skill",
requirements=None
)
validator = CapabilityValidator()
result = validator.validate_skill(skill)
assert result.skill_name == "test-skill"
assert result.state == SkillState.ACTIVE
assert result.is_fully_satisfied
assert not result.has_critical_missing
def test_validate_skill_empty_requirements(self):
"""Test validation of skill with empty requirements."""
requirements = SkillRequirements()
skill = SkillProperties(
name="test-skill",
description="Test skill",
requirements=requirements
)
validator = CapabilityValidator()
result = validator.validate_skill(skill)
assert result.state == SkillState.ACTIVE
assert result.is_fully_satisfied
@patch.dict(os.environ, {"TEST_VAR": "value"})
def test_validate_skill_satisfied_requirements(self):
"""Test validation with all requirements satisfied."""
requirements = SkillRequirements(
tools=["available_tool"],
servers=["available_server"],
env_vars=["TEST_VAR"]
)
skill = SkillProperties(
name="test-skill",
description="Test skill",
requirements=requirements
)
available_tools = {"available_tool"}
available_servers = {"available_server"}
validator = CapabilityValidator()
result = validator.validate_skill(
skill,
available_tools=available_tools,
available_servers=available_servers
)
assert result.state == SkillState.ACTIVE
assert result.satisfied_tools == ["available_tool"]
assert result.satisfied_servers == ["available_server"]
assert result.satisfied_env_vars == ["TEST_VAR"]
assert result.is_fully_satisfied
def test_validate_skill_missing_tools_warn_mode(self):
"""Test validation with missing tools in warn mode."""
requirements = SkillRequirements(tools=["missing_tool"])
skill = SkillProperties(
name="test-skill",
description="Test skill",
requirements=requirements
)
validator = CapabilityValidator(EnforcementLevel.WARN)
result = validator.validate_skill(
skill,
available_tools=set(),
available_servers=set()
)
assert result.state == SkillState.DEGRADED
assert result.missing_tools == ["missing_tool"]
assert not result.is_fully_satisfied
assert result.has_critical_missing
assert len(result.warnings) == 1
assert "missing_tool" in result.warnings[0]
def test_validate_skill_missing_tools_strict_mode(self):
"""Test validation with missing tools in strict mode."""
requirements = SkillRequirements(tools=["missing_tool"])
skill = SkillProperties(
name="test-skill",
description="Test skill",
requirements=requirements
)
validator = CapabilityValidator(EnforcementLevel.STRICT)
result = validator.validate_skill(
skill,
available_tools=set(),
available_servers=set()
)
assert result.state == SkillState.UNAVAILABLE
assert result.missing_tools == ["missing_tool"]
assert not result.is_fully_satisfied
assert result.has_critical_missing
assert len(result.errors) == 1
assert "missing_tool" in result.errors[0]
def test_validate_skill_missing_env_vars_warn(self):
"""Test validation with missing environment variables in warn mode."""
requirements = SkillRequirements(env_vars=["MISSING_VAR"])
skill = SkillProperties(
name="test-skill",
description="Test skill",
requirements=requirements
)
validator = CapabilityValidator(EnforcementLevel.WARN)
result = validator.validate_skill(skill)
assert result.state == SkillState.DEGRADED
assert result.missing_env_vars == ["MISSING_VAR"]
assert len(result.warnings) == 1
assert len(result.errors) == 0
def test_validate_skill_missing_env_vars_strict(self):
"""Test validation with missing environment variables in strict mode."""
requirements = SkillRequirements(env_vars=["MISSING_VAR"])
skill = SkillProperties(
name="test-skill",
description="Test skill",
requirements=requirements
)
validator = CapabilityValidator(EnforcementLevel.STRICT)
result = validator.validate_skill(skill)
assert result.state == SkillState.UNAVAILABLE
assert result.missing_env_vars == ["MISSING_VAR"]
assert len(result.warnings) == 0
assert len(result.errors) == 1
def test_validation_result_to_dict(self):
"""Test ValidationResult serialization."""
result = ValidationResult(
skill_name="test-skill",
state=SkillState.DEGRADED,
satisfied_tools=["tool1"],
missing_tools=["tool2"],
satisfied_servers=[],
missing_servers=[],
satisfied_env_vars=[],
missing_env_vars=["VAR1"],
warnings=["Missing tool2"],
errors=[]
)
data = result.to_dict()
assert data["skill_name"] == "test-skill"
assert data["state"] == "degraded"
assert data["satisfied_tools"] == ["tool1"]
assert data["missing_tools"] == ["tool2"]
assert data["warnings"] == ["Missing tool2"]
assert data["is_fully_satisfied"] == False
assert data["has_critical_missing"] == True
class TestIntegration:
"""Integration tests for the full capability validation system."""
def test_skill_parsing_with_requirements(self):
"""Test that skills parse requirements correctly."""
from praisonaiagents.skills.parser import read_properties
from tempfile import TemporaryDirectory
with TemporaryDirectory() as temp_dir:
skill_dir = Path(temp_dir) / "test-skill"
skill_dir.mkdir()
skill_md = skill_dir / "SKILL.md"
skill_md.write_text('''---
name: test-skill
description: A test skill
requires_tools:
- web_search
- file_write
requires_servers:
- mcp:filesystem
requires_env:
- API_KEY
openclaw:
version: "1.0"
---
# Test Skill
This is a test skill.
''')
properties = read_properties(skill_dir)
assert properties.requirements is not None
assert properties.requirements.tools == ["web_search", "file_write"]
assert properties.requirements.servers == ["mcp:filesystem"]
assert properties.requirements.env_vars == ["API_KEY"]
assert properties.requirements.openclaw_hints == {"version": "1.0"}
def test_skill_manager_enforcement(self):
"""Test that SkillManager respects enforcement levels."""
from praisonaiagents.skills.manager import SkillManager
# Test that manager initializes with default enforcement
manager = SkillManager()
assert manager._validator.enforcement_level in [
EnforcementLevel.WARN,
EnforcementLevel.DISABLED,
EnforcementLevel.STRICT,
EnforcementLevel.TELEMETRY
]
# Test explicit enforcement level
strict_manager = SkillManager(EnforcementLevel.STRICT)
assert strict_manager._validator.enforcement_level == EnforcementLevel.STRICT
@@ -241,89 +241,3 @@ Body.
loader.activate(skill)
assert skill.is_activated is True
def test_load_utf8_content(self):
"""Test loading UTF-8 encoded skill content.
Regression test for Windows encoding issues (Issue #1695).
Tests UTF-8 handling in SKILL.md body, scripts, and references.
"""
from praisonaiagents.skills.loader import SkillLoader
with tempfile.TemporaryDirectory() as tmpdir:
skill_dir = Path(tmpdir) / "utf8-loader-test"
skill_dir.mkdir()
# Create SKILL.md with UTF-8 characters
skill_content = """---
name: utf8-loader-test
description: Loader test with UTF-8 — em dashes — and more
---
# UTF-8 Instructions
This skill handles UTF-8 content:
- Smart quotes: "hello" and 'world'
- Em dash: —
- Arrows: → ← ↑ ↓
- Accented: café, naïve, résumé
- Emoji: 🔥 📝 ✨
"""
with open(skill_dir / "SKILL.md", 'w', encoding='utf-8') as f:
f.write(skill_content)
# Create scripts directory with UTF-8 script
scripts_dir = skill_dir / "scripts"
scripts_dir.mkdir()
script_content = """#!/bin/bash
# Script with UTF-8: café processing → résumé generation
echo "Processing naïve input with émphasis"
"""
with open(scripts_dir / "process.sh", 'w', encoding='utf-8') as f:
f.write(script_content)
# Create references directory with UTF-8 content
refs_dir = skill_dir / "references"
refs_dir.mkdir()
ref_content = """Reference with UTF-8:
Café automation guide — step by step
Arrow symbols: → ← ↑ ↓
Special chars: "smart quotes" and émphasis
"""
with open(refs_dir / "readme.txt", 'w', encoding='utf-8') as f:
f.write(ref_content)
# Test loading and activation
loader = SkillLoader()
skill = loader.load_metadata(str(skill_dir))
assert skill is not None
assert "—" in skill.properties.description # UTF-8 in metadata
# Test activation (loads SKILL.md body)
success = loader.activate(skill)
assert success is True
assert "—" in skill.instructions # UTF-8 in instructions
assert "café" in skill.instructions
assert "🔥" in skill.instructions # Emoji
# Test script loading
scripts = loader.load_scripts(skill)
assert "process.sh" in scripts
script_text = scripts["process.sh"]
assert "café" in script_text
assert "résumé" in script_text
assert "naïve" in script_text
# Test references loading
refs = loader.load_references(skill)
assert "readme.txt" in refs
ref_text = refs["readme.txt"]
assert "Café" in ref_text
assert "—" in ref_text
assert "émphasis" in ref_text
@@ -286,54 +286,3 @@ Body.
assert False, "Should have raised ValidationError"
except ValidationError as e:
assert "description" in str(e).lower()
def test_read_properties_utf8_encoding(self):
"""Test reading UTF-8 encoded SKILL.md files.
Regression test for Windows encoding issues (Issue #1695).
Verifies that skills with non-ASCII characters load correctly.
"""
from praisonaiagents.skills.parser import read_properties
with tempfile.TemporaryDirectory() as tmpdir:
skill_dir = Path(tmpdir) / "utf8-skill"
skill_dir.mkdir()
# Create SKILL.md with various UTF-8 characters
utf8_content = """---
name: utf8-test
description: UTF-8 encoding test — quality — verification
license: MIT
compatibility: Supports café files → résumé processing
metadata:
author: "José María"
notes: "Handles naïve input with émphasis"
---
# UTF-8 Test Skill
This skill demonstrates UTF-8 character handling:
- Em dash: —
- Smart quotes: "hello" and 'world'
- Arrow: →
- Accented chars: café, naïve, résumé
- Emoji: 🔥 📝 ✨
Works with internationalization!
"""
# Explicitly write as UTF-8 to simulate real-world scenario
with open(skill_dir / "SKILL.md", 'w', encoding='utf-8') as f:
f.write(utf8_content)
# Test that parsing works without UnicodeDecodeError
props = read_properties(skill_dir)
# Verify UTF-8 characters are preserved correctly
assert props.name == "utf8-test"
assert "—" in props.description # em dash
assert "café" in props.compatibility # accented character
assert "résumé" in props.compatibility # accented character
assert "José María" in props.metadata["author"] # accented characters
assert "naïve" in props.metadata["notes"] # accented character
assert "émphasis" in props.metadata["notes"] # accented character
@@ -295,58 +295,3 @@ description: A test skill for validation
assert len(errors) > 0
assert any("directory" in e.lower() for e in errors)
def test_validate_utf8_skill(self):
"""Test validation of UTF-8 encoded skill.
Regression test for Windows encoding issues (Issue #1695).
Ensures validator can handle UTF-8 characters without errors.
"""
from praisonaiagents.skills.validator import validate
with tempfile.TemporaryDirectory() as tmpdir:
skill_dir = Path(tmpdir) / "utf8-validation-test"
skill_dir.mkdir()
# Create SKILL.md with UTF-8 content
utf8_content = """---
name: utf8-validation-test
description: Validation test with UTF-8 — em dashes — and accents
license: MIT
compatibility: Works with café files → résumé output
---
# UTF-8 Validation Test
Testing validation with various UTF-8 characters.
"""
# Write as UTF-8 explicitly
with open(skill_dir / "SKILL.md", 'w', encoding='utf-8') as f:
f.write(utf8_content)
# Validate should not produce errors for UTF-8 content
errors = validate(skill_dir)
assert errors == [], f"Validation should pass for UTF-8 content, got errors: {errors}"
def test_validate_non_utf8_skill_returns_error(self):
"""Test validation returns a structured error for non-UTF-8 SKILL.md."""
from praisonaiagents.skills.validator import validate
with tempfile.TemporaryDirectory() as tmpdir:
skill_dir = Path(tmpdir) / "latin1-validation-test"
skill_dir.mkdir()
latin1_content = """---
name: latin1-validation-test
description: Café skill
---
"""
with open(skill_dir / "SKILL.md", "wb") as f:
f.write(latin1_content.encode("cp1252"))
errors = validate(skill_dir)
assert len(errors) > 0
assert any("utf-8" in e.lower() for e in errors)
@@ -225,19 +225,3 @@ class TestToolCallIdPropagation:
)
assert event.tool_call["id"] == "call_abc123xyz"
class TestToolExecutionMixinStreamEmitterLookup:
def test_mixin_resolves_name_mangled_stream_emitter_without_agent_class_name(self):
from praisonaiagents.agent.tool_execution import ToolExecutionMixin
from praisonaiagents.streaming.events import StreamEventEmitter
class CustomAgent(ToolExecutionMixin):
def __init__(self):
self.__stream_emitter = StreamEventEmitter()
agent = CustomAgent()
emitter = agent._get_existing_stream_emitter()
assert emitter is not None
assert isinstance(emitter, StreamEventEmitter)
@@ -134,37 +134,3 @@ class TestTokenCollectorSinkIntegration:
)
assert sink.records[0]["task_id"] == "task-abc"
assert sink.records[0]["agent_name"] == "researcher"
class TestInMemoryUsageQuery:
"""Test query adapter backed by in-memory sink records."""
def test_get_summary_uses_total_tokens_when_present(self):
from praisonaiagents.telemetry.protocols import InMemoryTokenUsageSink, InMemoryUsageQuery
sink = InMemoryTokenUsageSink()
sink.records.extend([
{
"task_id": "t1",
"agent_name": "a1",
"model": "m1",
"input_tokens": 10,
"output_tokens": 5,
"total_tokens": 50,
},
{
"task_id": "t2",
"agent_name": "a1",
"model": "m2",
"input_tokens": 2,
"output_tokens": 3,
},
])
summary = InMemoryUsageQuery(sink).get_summary()
assert summary["total_input_tokens"] == 12
assert summary["total_output_tokens"] == 8
assert summary["total_tokens"] == 55
assert summary["by_model"]["m1"]["total_tokens"] == 50
assert summary["by_model"]["m2"]["total_tokens"] == 5
@@ -1,293 +0,0 @@
"""
Unit tests for kanban protocols and hook events.
Tests the protocol contracts and hook events without any heavy
implementations like SQLite or LLM calls.
"""
import pytest
import json
from typing import runtime_checkable
from praisonaiagents.kanban.protocols import (
KanbanStoreProtocol,
KanbanTaskProtocol,
KanbanCommentingProtocol,
KanbanLinkingProtocol,
VALID_KANBAN_STATUSES,
)
from praisonaiagents.hooks.types import (
HookEvent,
KanbanHookInput,
)
class TestKanbanProtocol:
"""Test KanbanStoreProtocol runtime checking and constants."""
def test_kanban_store_protocol_is_runtime_checkable(self):
"""Test that KanbanStoreProtocol is @runtime_checkable."""
assert runtime_checkable(KanbanStoreProtocol)
def test_valid_kanban_statuses_includes_ui_columns(self):
"""Test that VALID_KANBAN_STATUSES includes all PraisonAIUI columns."""
expected_statuses = {
"triage",
"todo",
"ready",
"running",
"blocked",
"review",
"done",
"archived"
}
assert VALID_KANBAN_STATUSES == expected_statuses
def test_valid_kanban_statuses_is_frozen(self):
"""Test that VALID_KANBAN_STATUSES is immutable."""
assert isinstance(VALID_KANBAN_STATUSES, frozenset)
def test_mock_store_passes_isinstance_check(self):
"""Test that a mock store implementing the core protocol passes isinstance check."""
class MockKanbanStore:
"""Mock implementation for testing - only core methods required."""
def get_board(self, *, board="default", tenant=None, include_archived=False):
return {"board": board, "tasks": []}
def get_task(self, task_id: str):
return {"id": task_id, "title": "Test Task"}
def create_task(self, data: dict):
return {"id": "task_123", **data}
def update_task(self, task_id: str, data: dict):
return {"id": task_id, **data}
def move_task(self, task_id: str, status: str):
return {"id": task_id, "status": status}
def bulk_update(self, task_ids: list[str], status: str):
return {"updated": len(task_ids), "status": status}
def delete_task(self, task_id: str):
return True
def list_events(self, since: float = 0.0, board: str = "default"):
return []
def health(self):
return {"status": "healthy"}
mock_store = MockKanbanStore()
assert isinstance(mock_store, KanbanStoreProtocol)
def test_extension_protocol_checks(self):
"""Test that extension protocols work independently."""
class CommentingStore:
"""Store that implements commenting functionality."""
def add_comment(self, task_id: str, text: str, author: str | None = None):
return {"id": "comment_1", "text": text, "author": author}
class LinkingStore:
"""Store that implements linking functionality."""
def link_tasks(self, parent_id: str, child_id: str):
return True
def unlink_tasks(self, parent_id: str, child_id: str):
return True
class FullFeaturedStore:
"""Store that implements all protocols."""
# Core protocol
def get_board(self, *, board="default", tenant=None, include_archived=False):
return {"board": board, "tasks": []}
def get_task(self, task_id: str):
return {"id": task_id}
def create_task(self, data: dict):
return {"id": "task_123", **data}
def update_task(self, task_id: str, data: dict):
return {"id": task_id, **data}
def move_task(self, task_id: str, status: str):
return {"id": task_id, "status": status}
def bulk_update(self, task_ids: list[str], status: str):
return {"updated": len(task_ids)}
def delete_task(self, task_id: str):
return True
def list_events(self, since: float = 0.0, board: str = "default"):
return []
def health(self):
return {"status": "healthy"}
# Extension protocols
def add_comment(self, task_id: str, text: str, author: str | None = None):
return {"id": "comment_1", "text": text}
def link_tasks(self, parent_id: str, child_id: str):
return True
def unlink_tasks(self, parent_id: str, child_id: str):
return True
commenting_store = CommentingStore()
linking_store = LinkingStore()
full_store = FullFeaturedStore()
# Test individual extension protocols
assert isinstance(commenting_store, KanbanCommentingProtocol)
assert isinstance(linking_store, KanbanLinkingProtocol)
# Test that extension stores don't implement core protocol
assert not isinstance(commenting_store, KanbanStoreProtocol)
assert not isinstance(linking_store, KanbanStoreProtocol)
# Test full featured store implements all protocols
assert isinstance(full_store, KanbanStoreProtocol)
assert isinstance(full_store, KanbanCommentingProtocol)
assert isinstance(full_store, KanbanLinkingProtocol)
def test_kanban_task_protocol_shape(self):
"""Test KanbanTaskProtocol typed dict structure."""
# This tests that the TypedDict can be used for type hints
# The shape is validated at runtime by mypy/type checkers
task: KanbanTaskProtocol = {
"id": "task_123",
"title": "Test Task",
"body": "Task description",
"status": "todo",
"assignee": "[email protected]",
"priority": "high",
"tenant": "test_tenant",
"board": "default",
"created_at": 1640995200.0,
"updated_at": 1640995200.0,
}
# Basic validation that we can access the fields
assert task["id"] == "task_123"
assert task["title"] == "Test Task"
assert task["status"] == "todo"
class TestKanbanHookEvents:
"""Test kanban hook events and payload types."""
def test_kanban_hook_events_in_enum(self):
"""Test that all kanban hook events are registered in HookEvent."""
expected_events = [
HookEvent.KANBAN_TASK_CREATED,
HookEvent.KANBAN_TASK_CLAIMED,
HookEvent.KANBAN_TASK_MOVED,
HookEvent.KANBAN_TASK_DONE,
HookEvent.KANBAN_TASK_BLOCKED,
HookEvent.KANBAN_TASK_FAILED,
]
for event in expected_events:
assert event in HookEvent
def test_kanban_hook_events_serializable(self):
"""Test that kanban hook events can be serialized."""
events = [
HookEvent.KANBAN_TASK_CREATED,
HookEvent.KANBAN_TASK_CLAIMED,
HookEvent.KANBAN_TASK_MOVED,
HookEvent.KANBAN_TASK_DONE,
HookEvent.KANBAN_TASK_BLOCKED,
HookEvent.KANBAN_TASK_FAILED,
]
for event in events:
# Should be able to serialize to JSON and reconstruct enum value
serialized = json.dumps({"event": event.value})
restored = HookEvent(json.loads(serialized)["event"])
assert restored == event
def test_kanban_hook_input_creation(self):
"""Test KanbanHookInput dataclass creation and fields."""
hook_input = KanbanHookInput(
session_id="session_123",
cwd="/home/user",
event_name="kanban_task_created",
timestamp="2023-01-01T00:00:00Z",
task_id="task_456",
board="project_board",
status="todo",
assignee="[email protected]",
from_status=None,
to_status="running"
)
assert hook_input.session_id == "session_123"
assert hook_input.task_id == "task_456"
assert hook_input.board == "project_board"
assert hook_input.status == "todo"
assert hook_input.assignee == "[email protected]"
assert hook_input.from_status is None
assert hook_input.to_status == "running"
def test_kanban_hook_input_defaults(self):
"""Test KanbanHookInput with default values."""
hook_input = KanbanHookInput(
session_id="session_123",
cwd="/home/user",
event_name="kanban_task_created",
timestamp="2023-01-01T00:00:00Z"
)
# Check defaults
assert hook_input.task_id == ""
assert hook_input.board == "default"
assert hook_input.status == ""
assert hook_input.assignee is None
assert hook_input.from_status is None
assert hook_input.to_status is None
def test_kanban_hook_input_to_dict(self):
"""Test KanbanHookInput serialization to dict."""
hook_input = KanbanHookInput(
session_id="session_123",
cwd="/home/user",
event_name="kanban_task_moved",
timestamp="2023-01-01T00:00:00Z",
task_id="task_456",
board="project_board",
from_status="todo",
to_status="running"
)
result = hook_input.to_dict()
assert result["session_id"] == "session_123"
assert result["event_name"] == "kanban_task_moved"
assert result["task_id"] == "task_456"
assert result["board"] == "project_board"
assert result["from_status"] == "todo"
assert result["to_status"] == "running"
class TestProtocolIntegration:
"""Integration tests for protocol usage patterns."""
def test_protocol_import_structure(self):
"""Test that protocols can be imported from expected paths."""
# Test direct imports
from praisonaiagents.kanban.protocols import KanbanStoreProtocol
from praisonaiagents.hooks.types import KanbanHookInput
assert KanbanStoreProtocol is not None
assert KanbanHookInput is not None
def test_lazy_import_from_package_root(self):
"""Test that kanban protocols can be imported from package root."""
from praisonaiagents.kanban import (
KanbanStoreProtocol,
KanbanTaskProtocol,
KanbanCommentingProtocol,
KanbanLinkingProtocol,
VALID_KANBAN_STATUSES,
)
assert KanbanStoreProtocol is not None
assert KanbanTaskProtocol is not None
assert KanbanCommentingProtocol is not None
assert KanbanLinkingProtocol is not None
assert VALID_KANBAN_STATUSES is not None
+1 -1
View File
@@ -3125,7 +3125,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "1.6.48"
version = "1.6.43"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+7 -15
View File
@@ -1,19 +1,18 @@
{
"version": "1.5.87",
"lastUpdated": "2026-05-30",
"lastUpdated": "2026-05-20",
"generatedBy": "praisonai._dev.parity.generator",
"sourceOfTruth": "Python SDK (praisonaiagents)",
"status": "PARITY_ACHIEVED",
"summary": {
"pythonCoreFeatures": 335,
"pythonCoreFeatures": 334,
"rustFeatures": 667,
"gapCount": 56,
"parityPercentage": 199.1
"gapCount": 55,
"parityPercentage": 199.7
},
"pythonCoreSDK": {
"exports": [
"A2A",
"A2UI",
"AGUI",
"AUTONOMY_PRESETS",
"Agent",
@@ -1105,7 +1104,6 @@
"WebSearchProvider"
],
"config_loader": [
"AutoRagConfig",
"SessionConfig"
],
"context": [
@@ -1741,6 +1739,9 @@
"sandbox": [],
"session": [],
"skills": [],
"specialized": [
"AutoRagConfig"
],
"specialized_agents": [],
"streaming": [],
"task": [],
@@ -2820,15 +2821,6 @@
"status": "DONE",
"category": "ui"
},
{
"feature": "A2UI",
"python": true,
"rust": false,
"priority": "P3",
"effort": "low",
"status": "TODO",
"category": "ui"
},
{
"feature": "AGUI",
"python": true,
+4 -5
View File
@@ -1,16 +1,16 @@
# Rust Feature Parity Tracker
> **Python Features:** 335 | **Rust Features:** 667 | **Parity:** 85.1%
> **Python Features:** 334 | **Rust Features:** 667 | **Parity:** 85.3%
## Summary
| Metric | Count |
|--------|-------|
| Python Core Features | 335 |
| Python Core Features | 334 |
| Rust Features | 667 |
| **Actual Gap Count** | **50** |
| **Actual Gap Count** | **49** |
| Language Limitations (N/A) | 5 |
| **Parity** | **85.1%** |
| **Parity** | **85.3%** |
## Implemented Features
@@ -694,7 +694,6 @@ These Python features cannot be directly implemented in Rust due to reserved key
## Missing Features
- ❌ `A2UI`
- ❌ `AgentMessageEvent`
- ❌ `AsyncLearnProtocol`
- ❌ `AutoApproveBackend`
+4 -15
View File
@@ -1,23 +1,22 @@
{
"version": "1.5.87",
"lastUpdated": "2026-05-30",
"lastUpdated": "2026-05-20",
"generatedBy": "praisonai._dev.parity.generator",
"sourceOfTruth": "Python SDK (praisonaiagents)",
"summary": {
"pythonCoreFeatures": 335,
"pythonCoreFeatures": 334,
"pythonWrapperFeatures": 99,
"typescriptFeatures": 1192,
"gapCount": 43,
"gapCount": 42,
"priorityP0": 1,
"priorityP1": 0,
"priorityP2": 0,
"priorityP3": 42
"priorityP3": 41
},
"pythonCoreSDK": {
"path": "/home/runner/work/PraisonAI/PraisonAI/src/praisonai-agents/praisonaiagents",
"exports": [
"A2A",
"A2UI",
"AGUI",
"AUTONOMY_PRESETS",
"Agent",
@@ -728,7 +727,6 @@
],
"ui": [
"A2A",
"A2UI",
"AGUI"
],
"workflows": [
@@ -3913,15 +3911,6 @@
"status": "DONE",
"category": "ui"
},
{
"feature": "A2UI",
"python": true,
"typescript": false,
"priority": "P3",
"effort": "low",
"status": "TODO",
"category": "ui"
},
{
"feature": "AGUI",
"python": true,
+7 -8
View File
@@ -1,20 +1,20 @@
# Feature Parity Tracker
> **Version:** 1.5.87 | **Last Updated:** 2026-05-30
> **Version:** 1.5.87 | **Last Updated:** 2026-05-20
> **Source of Truth:** Python SDK (praisonaiagents)
## Summary
| Metric | Count |
|--------|-------|
| Python Core Features | 335 |
| Python Core Features | 334 |
| Python Wrapper Features | 99 |
| TypeScript Features | 1192 |
| **Gap Count** | **43** |
| **Gap Count** | **42** |
| P0 (Critical) | 1 |
| P1 (High) | 0 |
| P2 (Medium) | 0 |
| P3 (Low) | 42 |
| P3 (Low) | 41 |
## Gap Matrix
@@ -147,11 +147,10 @@
| `enable\_telemetry` | ✅ | ✅ | low | ✅ DONE |
| `get\_telemetry` | ✅ | ✅ | low | ✅ DONE |
### P3_Advanced (179 done, 42 todo)
### P3_Advanced (179 done, 41 todo)
| Feature | Python | TypeScript | Effort | Status |
|---------|--------|------------|--------|--------|
| `A2UI` | ✅ | ❌ | low | ⏳ TODO |
| `AgentMessageEvent` | ✅ | ❌ | high | ⏳ TODO |
| `AsyncLearnProtocol` | ✅ | ❌ | medium | ⏳ TODO |
| `AutoApproveBackend` | ✅ | ❌ | high | ⏳ TODO |
@@ -630,10 +629,10 @@ from praisonaiagents import ContextEvent, ContextEventType, ContextListSink, Con
</details>
<details>
<summary><strong>ui</strong> (3 exports)</summary>
<summary><strong>ui</strong> (2 exports)</summary>
```python
from praisonaiagents import A2A, A2UI, AGUI
from praisonaiagents import A2A, AGUI
```
</details>
+1 -1
View File
@@ -382,7 +382,7 @@ praisonai flow
Open **http://localhost:7861** — use the **Agent** and **Agent Team** components to create sequential or parallel workflows. Connect Chat Input → Agent Team → Chat Output for instant multi-agent pipelines.
> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/agentflow) — visual agent building, component reference, and deployment
> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/flow) — visual agent building, component reference, and deployment
### 8. PraisonAI UI 🤖 (Clean Chat)
-31
View File
@@ -1,31 +0,0 @@
# Channels gateway integration
Channel bots (Telegram, Discord, Slack, WhatsApp) require the **`praisonai[bot]`** optional extra:
```bash
pip install "praisonai[bot]"
```
## Architecture
- **Pattern B/C host**: `configure_host()` + `AIUIGateway` serve chat UI and REST APIs.
- **Channels feature** (`praisonaiui.features.channels`): starts platform bots via `praisonai.bots`.
- **Gateway WebSocket**: `/ws` on the same process when using Pattern C.
## Environment variables
| Platform | Variables |
|----------|-------------|
| Telegram | `TELEGRAM_BOT_TOKEN` |
| Discord | `DISCORD_BOT_TOKEN` |
| Slack | `SLACK_BOT_TOKEN`, `SLACK_APP_TOKEN` |
| WhatsApp | `WHATSAPP_ACCESS_TOKEN`, `WHATSAPP_PHONE_NUMBER_ID` |
## CLI
```bash
praisonai dashboard --aiui # Pattern B in-process host
PRAISONAI_HOST_LEGACY=1 praisonai ui # Legacy @aiui.reply callbacks only
```
See `praisonai.bots.BotOS` for multi-platform orchestration.
-26
View File
@@ -1,26 +0,0 @@
# Pattern D — Platform API integration (defer)
Pattern D connects PraisonAIUI dashboard pages to **PraisonAI Cloud** via `PlatformClient` `/api/v1`.
## Scope (P3)
| Item | Status |
|------|--------|
| aiui pages → PlatformClient | Documented, optional |
| Platform JWT auth | Optional `auth.platform_jwt` config |
| Issues / kanban pages | Optional `@aiui.page` via platform |
| agent_id linking | See below |
## agent_id linking
Three registries may coexist:
1. **Platform roster** — cloud tenant agents (`/api/v1/agents`)
2. **Gateway registry** — WebSocket `/ws` agent ids
3. **aiui SDKAgentRegistry** — YAML CRUD + local `Agent` instances
Use explicit `agent_id` in session metadata (`source=gateway|platform|ui`) to correlate rows in the workflow-runs and sessions tables.
## Optional platform JWT
When `PRAISONAI_PLATFORM_TOKEN` is set, future aiui auth middleware may validate JWT from the platform issuer. Not required for Patterns B/C self-hosted installs.
Binary file not shown.
+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/v4.6.48.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.48.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.43.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.6.43.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+4 -20
View File
@@ -6,8 +6,6 @@ from .version import __version__
# Define __all__ for lazy loading
__all__ = [
'PraisonAI',
'run',
'arun',
'__version__',
'Deploy',
'DeployConfig',
@@ -32,8 +30,6 @@ __all__ = [
'HostedAgentConfig',
'LocalAgent',
'LocalAgentConfig',
# Integration functions
'run_integrated_gateway',
]
# Telemetry initialization state - thread-safe (threading imported above)
@@ -96,13 +92,7 @@ def __getattr__(name):
# Note: Telemetry initialization moved out of lazy hook to avoid side effects
# It should be called explicitly from cli.PraisonAI.__init__ instead
if name == 'run':
from ._entrypoint import run
return run
elif name == 'arun':
from ._entrypoint import arun
return arun
elif name == 'PraisonAI':
if name == 'PraisonAI':
from .cli import PraisonAI
return PraisonAI
elif name == 'Agent':
@@ -140,15 +130,9 @@ def __getattr__(name):
elif name == 'EmbeddingResult':
from praisonaiagents.embedding import EmbeddingResult
return EmbeddingResult
elif name == 'build_host_app':
from .integration.host_app import build_host_app
return build_host_app
elif name == 'configure_host':
from .integration.host_app import configure_host
return configure_host
elif name == 'run_integrated_gateway':
from .integration.gateway_host import run_integrated_gateway
return run_integrated_gateway
elif name == 'AgentOS':
from .app import AgentOS
return AgentOS
elif name == 'AgentApp':
# Silent alias for AgentOS (backward compatibility)
from .app import AgentOS
+3 -25
View File
@@ -8,8 +8,7 @@ on every call (which is expensive and breaks multi-agent workflows).
import asyncio
import os
import threading
import concurrent.futures
from concurrent.futures import CancelledError as FutureCancelledError, Future
from concurrent.futures import Future
from typing import Awaitable, TypeVar
T = TypeVar("T")
@@ -113,30 +112,9 @@ def run_sync(coro: Awaitable[T], *, timeout: float | None = _DEFAULT_TIMEOUT) ->
with _BG._lock:
loop = _BG.get_unlocked()
fut: Future = asyncio.run_coroutine_threadsafe(coro, loop)
try:
return fut.result(timeout=timeout)
except (TimeoutError, concurrent.futures.TimeoutError):
# Propagate cancellation into the background loop so the underlying
# awaitable (DB query, HTTP call, subprocess wait) actually unwinds.
fut.cancel()
try:
# Give cancellation a short grace period to release resources.
fut.exception(timeout=1.0)
except (
TimeoutError,
concurrent.futures.TimeoutError,
asyncio.CancelledError,
FutureCancelledError,
):
pass
raise
except BaseException:
# Ctrl-C / GeneratorExit / SystemExit must also cancel the bg task.
fut.cancel()
raise
return fut.result(timeout=timeout)
def shutdown() -> None:
"""Public hook for long-running server processes to stop the bridge cleanly."""
_BG.shutdown()
_BG.shutdown()
+1 -1
View File
@@ -119,7 +119,7 @@ class ApiMdGenerator:
'Observability': ['obs', 'FlowDisplay', 'track_workflow'],
'Context': ['FastContext', 'FastContextResult', 'FileMatch', 'LineRange',
'ContextManager', 'ManagerConfig'],
'UI': ['AGUI', 'A2A', 'A2UI'],
'UI': ['AGUI', 'A2A'],
'Config': ['MemoryConfig', 'KnowledgeConfig', 'PlanningConfig', 'ReflectionConfig',
'GuardrailConfig', 'WebConfig', 'OutputConfig', 'ExecutionConfig',
'TemplateConfig', 'CachingConfig', 'HooksConfig', 'SkillsConfig',
@@ -82,7 +82,7 @@ class PythonFeatureExtractor:
'telemetry': ['telemetry'],
'context': ['context.fast', 'context.manager', 'context.models'],
'session': ['session'],
'ui': ['ui.agui', 'ui.a2a', 'ui.a2ui'],
'ui': ['ui.agui', 'ui.a2a'],
'rag': ['rag'],
'planning': ['planning'],
'config': ['config.feature_configs', 'config.param_resolver',
-73
View File
@@ -1,73 +0,0 @@
"""Top-level convenience entry points for the Python API.
Keep this file minimal - it is part of the Minimal API surface and must defer
all heavy imports to call time.
"""
from __future__ import annotations
from typing import Any
def _build_config_list() -> list[dict[str, Any]]:
"""Reuse the same env/keyfile resolution the CLI already performs."""
# Import LLM endpoint resolution from the same path the CLI uses
from praisonai.llm.env import resolve_llm_endpoint
ep = resolve_llm_endpoint()
return [{
'model': ep.model,
'base_url': ep.base_url,
'api_key': ep.api_key,
'api_type': 'openai' # AutoGen expects this field
}]
def run(agent_file: str,
framework: str | None = None,
*,
tools: list | None = None,
agent_yaml: str | None = None,
cli_config: dict | None = None) -> str:
"""One-line Python entry point. Equivalent to `praisonai <agent_file>`."""
from .agents_generator import AgentsGenerator
from ._framework_availability import is_available
if framework is None:
# Mirror the CLI default-resolution order
for candidate in ("crewai", "praisonaiagents", "autogen", "ag2"):
if is_available(candidate):
framework = "praisonai" if candidate == "praisonaiagents" else candidate
break
else:
raise RuntimeError(
"No supported framework installed. "
"Install one of: crewai, praisonaiagents, autogen, ag2."
)
else:
# Validate explicit framework like CLI does
from .framework_adapters.validators import assert_framework_available
assert_framework_available(framework)
gen = AgentsGenerator(
agent_file=agent_file,
framework=framework,
config_list=_build_config_list(),
tools=tools,
agent_yaml=agent_yaml,
cli_config=cli_config,
)
return gen.generate_crew_and_kickoff()
async def arun(agent_file: str,
framework: str | None = None,
*,
tools: list | None = None,
agent_yaml: str | None = None,
cli_config: dict | None = None) -> str:
"""Async equivalent of `run()` for FastAPI / Jupyter async callers."""
import asyncio
return await asyncio.to_thread(
run, agent_file, framework,
tools=tools, agent_yaml=agent_yaml, cli_config=cli_config,
)
@@ -30,11 +30,6 @@ _PROBES: dict[str, Callable[[], bool]] = {
"agentops": lambda: importlib.util.find_spec("agentops") is not None,
"litellm": lambda: importlib.util.find_spec("litellm") is not None,
"openai": lambda: importlib.util.find_spec("openai") is not None,
# Additional probes for Gap 2 consolidation
"acp": lambda: importlib.util.find_spec("acp") is not None,
"rich": lambda: importlib.util.find_spec("rich") is not None,
"gradio": lambda: importlib.util.find_spec("gradio") is not None,
"unsloth": lambda: importlib.util.find_spec("unsloth") is not None,
}
def is_available(name: str) -> bool:
+1 -2
View File
@@ -42,8 +42,7 @@ def __getattr__(name: str):
def _check_acp_available() -> bool:
"""Check if agent-client-protocol package is installed."""
from .._framework_availability import is_available
return is_available("acp")
return importlib.util.find_spec("acp") is not None
def _get_install_instructions() -> str:
File diff suppressed because it is too large. Load diff
+413 -11
View File
@@ -7,17 +7,419 @@ scheduler package for full functionality including async support.
import warnings
warnings.warn(
"praisonai.async_agent_scheduler is deprecated; "
"use 'from praisonai.scheduler import AsyncAgentScheduler' instead.",
DeprecationWarning,
"praisonai.async_agent_scheduler is pending deprecation; it will be moved to "
"praisonai.scheduler.async_agent_scheduler in a future release. Continue "
"importing from praisonai.async_agent_scheduler until then.",
PendingDeprecationWarning,
stacklevel=2,
)
# Re-export from the canonical location
from .scheduler.async_agent_scheduler import ( # noqa: F401
AsyncAgentScheduler,
AsyncPraisonAgentExecutor,
AsyncAgentExecutorInterface,
create_async_agent_scheduler,
logger,
)
# TODO: Once AsyncAgentScheduler is moved to scheduler package, import from there
# For now, re-export the existing implementation to avoid breaking changes
from .scheduler.shared import ScheduleParser, backoff_delay, safe_call
import asyncio
import logging
import threading
from datetime import datetime
from typing import Optional, Dict, Any, Callable, Union
from abc import ABC, abstractmethod
logger = logging.getLogger(__name__)
class AsyncAgentExecutorInterface(ABC):
"""Abstract interface for async agent execution."""
@abstractmethod
async def execute(self, task: str) -> Any:
"""Execute the agent with given task asynchronously."""
pass
class AsyncPraisonAgentExecutor(AsyncAgentExecutorInterface):
"""Async executor for PraisonAI agents."""
def __init__(self, agent):
"""
Initialize executor with a PraisonAI agent.
Args:
agent: PraisonAI Agent instance
"""
self.agent = agent
async def execute(self, task: str) -> Any:
"""
Execute the agent with the given task.
Args:
task: Task description to execute
Returns:
Agent execution result
"""
try:
# Check if agent has async support
if hasattr(self.agent, 'astart'):
result = await self.agent.astart(task)
elif hasattr(self.agent, 'start'):
# Wrap sync call in executor
result = await asyncio.to_thread(self.agent.start, task)
else:
raise AttributeError("Agent must have either 'start' or 'astart' method")
return result
except Exception as e:
logger.error(f"Async agent execution failed: {e}")
raise
class AsyncAgentScheduler:
"""
Async-native scheduler for running PraisonAI agents periodically.
Features:
- Proper async/await execution
- Cancellation support
- No global state pollution
- Native async coordination
Example:
scheduler = AsyncAgentScheduler(agent, task="Check news")
await scheduler.start(schedule_expr="hourly")
await asyncio.sleep(3600) # Let it run
await scheduler.stop()
"""
def __init__(
self,
agent,
task: str,
config: Optional[Dict[str, Any]] = None,
on_success: Optional[Callable] = None,
on_failure: Optional[Callable] = None
):
"""
Initialize async agent scheduler.
Args:
agent: PraisonAI Agent instance
task: Task description to execute
config: Optional configuration dict
on_success: Callback function on successful execution
on_failure: Callback function on failed execution
"""
self.agent = agent
self.task = task
self.config = config or {}
self.on_success = on_success
self.on_failure = on_failure
self.is_running = False
self._task: Optional[asyncio.Task] = None
self._executor = AsyncPraisonAgentExecutor(agent)
self._execution_count = 0
self._success_count = 0
self._failure_count = 0
# Sync lock for async primitives creation and bound loop tracking
self._primitives_lock = threading.Lock()
self._stop_event: Optional[asyncio.Event] = None
self._cancel_event: Optional[asyncio.Event] = None
self._stats_lock: Optional[asyncio.Lock] = None
self._bound_loop: Optional[asyncio.AbstractEventLoop] = None
def _ensure_async_primitives(self) -> None:
"""Create async primitives if they don't exist yet.
Thread-safe and loop-aware: primitives are bound to the current running loop.
If called from a different loop, new primitives are created.
"""
loop = asyncio.get_running_loop() # must be called from a coroutine
with self._primitives_lock:
if self._bound_loop is not loop:
self._stop_event = asyncio.Event()
self._cancel_event = asyncio.Event()
self._stats_lock = asyncio.Lock()
self._bound_loop = loop
async def start(
self,
schedule_expr: str,
max_retries: int = 3,
run_immediately: bool = False
) -> bool:
"""
Start scheduled agent execution.
Args:
schedule_expr: Schedule expression (e.g., "hourly", "*/6h", "3600")
max_retries: Maximum retry attempts on failure
run_immediately: If True, run agent immediately before starting schedule
Returns:
True if scheduler started successfully
"""
if self.is_running:
logger.warning("Async scheduler is already running")
return False
try:
interval = ScheduleParser.parse(schedule_expr)
self.is_running = True
self._ensure_async_primitives() # NEW — bind to the loop start() runs on
self._stop_event.clear()
logger.info(f"Starting async agent scheduler: {getattr(self.agent, 'name', 'Agent')}")
logger.info(f"Task: {self.task}")
logger.info(f"Schedule: {schedule_expr} ({interval}s interval)")
# Run immediately if requested
if run_immediately:
logger.info("Running agent immediately before starting schedule...")
await self._execute_with_retry(max_retries)
# Start background task
self._task = asyncio.create_task(
self._run_schedule(interval, max_retries)
)
logger.info("Async agent scheduler started successfully")
return True
except Exception as e:
logger.error(f"Failed to start async scheduler: {e}")
self.is_running = False
return False
async def stop(self) -> bool:
"""
Stop the scheduler gracefully.
IMPORTANT: This method must be called from the same event loop
that was used to start the scheduler.
Returns:
True if stopped successfully
Raises:
RuntimeError: If called from a different event loop than start()
"""
if not self.is_running:
logger.info("Async scheduler is not running")
return True
logger.info("Stopping async agent scheduler...")
# _stop_event is guaranteed non-None after a successful start()
if self._stop_event is None:
logger.warning("stop() called before start(); nothing to stop.")
self.is_running = False
return True
# Ensure we're on the same loop that was bound during start()
current_loop = asyncio.get_running_loop()
if self._bound_loop is not None and current_loop is not self._bound_loop:
raise RuntimeError(
"stop() must be called from the same event loop as start(). "
f"Expected: {self._bound_loop}, got: {current_loop}"
)
self._stop_event.set()
try:
if self._task:
try:
await asyncio.wait_for(self._task, timeout=10)
except asyncio.TimeoutError:
logger.warning("Scheduler task didn't stop gracefully, cancelling")
self._task.cancel()
try:
await self._task
except asyncio.CancelledError:
pass
except Exception as e:
logger.error(f"Scheduler task raised on cancel: {e}")
except asyncio.CancelledError:
# Task already cancelled; treat as expected during shutdown
pass
except Exception as e:
logger.error(f"Scheduler task raised during stop: {e}")
finally:
self.is_running = False
logger.info("Async agent scheduler stopped")
# Log final stats with consistent snapshot
if self._stats_lock is not None:
async with self._stats_lock:
total = self._execution_count
ok = self._success_count
fail = self._failure_count
else:
total = self._execution_count
ok = self._success_count
fail = self._failure_count
logger.info(f"Execution stats - Total: {total}, Success: {ok}, Failed: {fail}")
return True
async def get_stats(self) -> Dict[str, Any]:
"""
Get current execution statistics (synchronous, best-effort).
Warning: This method provides a best-effort view of stats without
guaranteeing atomicity. For consistent snapshots in async context,
use get_stats_async() instead.
Returns:
Dictionary with execution stats
"""
# Primitives are created lazily in _ensure_async_primitives(); if not yet
# initialized there are no concurrent writers, so a direct read is safe.
if self._stats_lock is not None:
async with self._stats_lock:
total = self._execution_count
ok = self._success_count
fail = self._failure_count
running = self.is_running
else:
total = self._execution_count
ok = self._success_count
fail = self._failure_count
running = self.is_running
return {
"is_running": running,
"total_executions": total,
"successful_executions": ok,
"failed_executions": fail,
"success_rate": (ok / total * 100) if total > 0 else 0
}
async def get_stats_async(self) -> Dict[str, Any]:
"""
Get current execution statistics with atomic snapshot (async).
Returns:
Dictionary with execution stats
"""
if self._stats_lock is None:
# Not yet started: stats are all zero, no lock needed
execs, success, failed = 0, 0, 0
else:
# Take atomic snapshot of all counters
async with self._stats_lock:
execs = self._execution_count
success = self._success_count
failed = self._failure_count
return {
"is_running": self.is_running,
"total_executions": execs,
"successful_executions": success,
"failed_executions": failed,
"success_rate": (success / execs * 100) if execs > 0 else 0
}
def get_stats_sync(self) -> Dict[str, Any]:
"""
Alias for get_stats() for clarity.
Returns:
Dictionary with execution stats
"""
return self.get_stats()
async def _run_schedule(self, interval: int, max_retries: int):
"""Internal method to run scheduled agent executions."""
try:
self._ensure_async_primitives()
while not self._stop_event.is_set():
logger.info(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Starting async scheduled agent execution")
await self._execute_with_retry(max_retries)
# Wait for next scheduled time or stop event
logger.info(f"Next execution in {interval} seconds ({interval/3600:.1f} hours)")
try:
await asyncio.wait_for(self._stop_event.wait(), timeout=interval)
break # Stop event was set
except asyncio.TimeoutError:
continue # Timeout reached, continue with next execution
finally:
self.is_running = False
async def _execute_with_retry(self, max_retries: int):
"""Execute agent with retry logic."""
self._ensure_async_primitives() # guarantees _stats_lock is bound to current loop
async with self._stats_lock:
self._execution_count += 1
last_exc: Optional[Exception] = None
for attempt in range(max_retries):
try:
logger.info(f"Async attempt {attempt + 1}/{max_retries}")
result = await self._executor.execute(self.task)
logger.info(f"Async agent execution successful on attempt {attempt + 1}")
logger.info(f"Result: {result}")
async with self._stats_lock:
self._success_count += 1
safe_call(self.on_success, result)
return
except Exception as e:
last_exc = e
logger.error(f"Async agent execution failed on attempt {attempt + 1}: {e}")
if attempt < max_retries - 1:
wait_time = backoff_delay(attempt)
logger.info(f"Waiting {wait_time}s before async retry...")
await asyncio.sleep(wait_time)
async with self._stats_lock:
self._failure_count += 1
logger.error(f"Async agent execution failed after {max_retries} attempts")
safe_call(
self.on_failure,
last_exc if last_exc is not None
else RuntimeError(f"Failed after {max_retries} attempts")
)
async def execute_once(self) -> Any:
"""
Execute agent immediately (one-time execution).
Returns:
Agent execution result
"""
logger.info("Executing agent once (async)")
try:
result = await self._executor.execute(self.task)
logger.info(f"One-time async execution successful: {result}")
return result
except Exception as e:
logger.error(f"One-time async execution failed: {e}")
raise
def create_async_agent_scheduler(
agent,
task: str,
config: Optional[Dict[str, Any]] = None
) -> AsyncAgentScheduler:
"""
Factory function to create async agent scheduler.
Args:
agent: PraisonAI Agent instance
task: Task description
config: Optional configuration
Returns:
Configured AsyncAgentScheduler instance
"""
return AsyncAgentScheduler(agent, task, config)
+8 -176
View File
@@ -14,7 +14,6 @@ from pydantic import BaseModel
from typing import Any, Dict, List, Optional, Type, TypeVar
import os
import json
import asyncio
import yaml
import threading
from rich import print
@@ -424,13 +423,12 @@ class BaseAutoGenerator:
}
]
self._openai_client = None # lazy, per-instance
self._async_openai_client = None # lazy, per-instance async client
self._client_lock = threading.Lock()
self._openai_client_lock = threading.Lock()
def _get_openai_client(self):
"""Get or create the OpenAI client for this instance."""
if self._openai_client is None:
with self._client_lock:
with self._openai_client_lock:
if self._openai_client is None:
try:
from openai import OpenAI
@@ -444,43 +442,18 @@ class BaseAutoGenerator:
return self._openai_client
def close(self):
"""Close the sync OpenAI client if it exists."""
if not hasattr(self, '_client_lock'):
"""Close the OpenAI client if it exists."""
if not hasattr(self, '_openai_client_lock'):
return # Object was never fully initialized
with self._client_lock:
with self._openai_client_lock:
client = getattr(self, '_openai_client', None)
self._openai_client = None
if client is not None:
client.close()
async def aclose(self):
"""Close both sync and async OpenAI clients if they exist."""
if not hasattr(self, '_client_lock'):
return # Object was never fully initialized
with self._client_lock:
sync_client = getattr(self, '_openai_client', None)
self._openai_client = None
async_client = getattr(self, '_async_openai_client', None)
self._async_openai_client = None
if sync_client is not None:
await asyncio.to_thread(sync_client.close)
if async_client is not None:
await async_client.close()
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
self.close()
return False
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb):
await self.aclose()
return False
def __del__(self):
"""Best-effort cleanup, but the canonical path is explicit close()."""
self.close()
def _structured_completion(self, response_model: Type[T], messages: List[Dict], **kwargs) -> T:
"""
@@ -532,68 +505,6 @@ class BaseAutoGenerator:
"Install with: pip install litellm OR pip install openai"
)
async def _astructured_completion(self, response_model: Type[T], messages: List[Dict], **kwargs) -> T:
"""
Make an async structured LLM completion with provider fallback.
Priority:
1. LiteLLM async (if available) - supports 100+ LLM providers
2. OpenAI AsyncSDK (fallback) - uses beta.chat.completions.parse
Args:
response_model: Pydantic model class for structured output
messages: List of message dicts for the LLM
**kwargs: Additional arguments passed to the LLM
Returns:
Instance of response_model with parsed response
Raises:
ImportError: If neither litellm nor openai is installed
"""
model_name = self.config_list[0]['model']
# Try LiteLLM async first (preferred - supports 100+ providers)
if _check_litellm_available():
litellm = _get_litellm()
response = await litellm.acompletion(
model=model_name,
messages=messages,
response_format=response_model,
**kwargs
)
content = response.choices[0].message.content
return response_model.model_validate_json(content)
# Fallback to OpenAI AsyncSDK (uses beta.chat.completions.parse)
if _check_openai_available():
if self._async_openai_client is None:
with self._client_lock:
if self._async_openai_client is None:
try:
from openai import AsyncOpenAI
except ImportError as e:
raise ImportError("Install with: pip install openai") from e
cfg = self.config_list[0]
self._async_openai_client = AsyncOpenAI(
api_key=cfg.get("api_key") or os.environ.get("OPENAI_API_KEY"),
base_url=cfg.get("base_url"),
)
response = await self._async_openai_client.beta.chat.completions.parse(
model=model_name,
messages=messages,
response_format=response_model,
**kwargs
)
return response.choices[0].message.parsed
# Neither available - raise helpful error
raise ImportError(
"Structured output requires either litellm or openai. "
"Install with: pip install litellm OR pip install openai"
)
@staticmethod
def get_available_tools() -> List[str]:
"""Return list of available tools for agent assignment."""
@@ -797,36 +708,6 @@ class AutoGenerator(BaseAutoGenerator):
self.convert_and_save(json_data, merge=merge)
full_path = os.path.abspath(self.agent_file)
return full_path
async def agenerate(self, merge=False):
"""
Async version of generate() - generates a team structure for the specified topic.
Args:
merge (bool): Whether to merge with existing agents.yaml file instead of overwriting.
Returns:
str: The full path of the YAML file containing the generated team structure.
Raises:
Exception: If the generation process fails.
Usage:
async with AutoGenerator(framework="crewai", topic="Create a movie script about Cat in Mars") as gen:
path = await gen.agenerate()
print(path)
"""
response = await self._astructured_completion(
response_model=TeamStructure,
messages=[
{"role": "system", "content": "You are a helpful assistant designed to output complex team structures."},
{"role": "user", "content": self.get_user_content()}
]
)
json_data = json.loads(response.model_dump_json())
self.convert_and_save(json_data, merge=merge)
full_path = os.path.abspath(self.agent_file)
return full_path
def convert_and_save(self, json_data, merge=False):
"""Converts the provided JSON data into the desired YAML format and saves it to a file.
@@ -1240,32 +1121,6 @@ Respond with:
return self._save_workflow(self.merge_with_existing_workflow(json_data), pattern)
return self._save_workflow(json_data, pattern)
async def agenerate(self, pattern: str = "sequential", merge: bool = False) -> str:
"""
Async version of generate() - Generate a workflow YAML file.
Args:
pattern: Workflow pattern - "sequential", "routing", "parallel", "loop",
"orchestrator-workers", "evaluator-optimizer"
merge: If True, merge with existing workflow file instead of overwriting
Returns:
Path to the generated workflow file
"""
response = await self._astructured_completion(
response_model=WorkflowStructure,
messages=[
{"role": "system", "content": "You are a helpful assistant that designs workflow structures."},
{"role": "user", "content": self._get_prompt(pattern)}
]
)
json_data = json.loads(response.model_dump_json())
if merge and os.path.exists(self.workflow_file):
return self._save_workflow(self.merge_with_existing_workflow(json_data), pattern)
return self._save_workflow(json_data, pattern)
def merge_with_existing_workflow(self, new_data: Dict) -> Dict:
"""
Merge new workflow data with existing workflow file.
@@ -1566,29 +1421,6 @@ class JobWorkflowAutoGenerator(BaseAutoGenerator):
return self._save_workflow(response)
async def agenerate(self, include_judge: bool = True, include_approve: bool = False) -> str:
"""
Async version of generate() - Generate a job workflow YAML file.
Args:
include_judge: Include a judge step for quality gating
include_approve: Include an approve step for human approval
Returns:
Path to the generated workflow file
"""
prompt = self._get_prompt(include_judge, include_approve)
response = await self._astructured_completion(
response_model=JobWorkflowStructure,
messages=[
{"role": "system", "content": "You are a helpful assistant that designs job workflow structures with AI agent steps."},
{"role": "user", "content": prompt}
]
)
return self._save_workflow(response)
def _get_prompt(self, include_judge: bool, include_approve: bool) -> str:
"""Generate the prompt for job workflow generation."""
tools_list = ", ".join(self.get_available_tools())
@@ -1,116 +0,0 @@
"""
Telegram typing indicator renewal utility.
Provides a background task to periodically renew typing indicators
during long-running operations like agent.chat() calls.
"""
import asyncio
import logging
from typing import Any, Callable, Coroutine, Optional, TypeVar
logger = logging.getLogger(__name__)
T = TypeVar("T")
class TypingIndicator:
"""Manages typing indicator renewal during long operations."""
def __init__(self, interval: float = 4.0):
"""
Initialize typing indicator manager.
Args:
interval: Seconds between typing renewals (default 4.0 for Telegram)
"""
self.interval = interval
self._task: Optional[asyncio.Task] = None
self._cancelled = False
async def start(self, typing_func: Callable[[], Any]) -> None:
"""
Start typing indicator renewal task.
Args:
typing_func: Async function that sends typing action
"""
if self._task and not self._task.done():
# Already running
return
self._cancelled = False
self._task = asyncio.create_task(self._typing_loop(typing_func))
def cancel(self) -> None:
"""Cancel the typing indicator renewal."""
self._cancelled = True
if self._task and not self._task.done():
self._task.cancel()
async def _typing_loop(self, typing_func: Callable[[], Any]) -> None:
"""
Background loop that periodically sends typing action.
Args:
typing_func: Function to call for sending typing indicator
"""
try:
while not self._cancelled:
try:
# Send typing indicator
if asyncio.iscoroutinefunction(typing_func):
await typing_func()
else:
typing_func()
# Wait for interval or until cancelled
await asyncio.sleep(self.interval)
except asyncio.CancelledError:
break
except Exception as e:
logger.debug(f"Typing indicator send failed: {e}")
# Continue trying even if one fails
await asyncio.sleep(self.interval)
except asyncio.CancelledError:
pass
finally:
logger.debug("Typing indicator loop ended")
async def with_typing_renewal(
typing_func: Callable[[], Any],
operation_coro: Coroutine[Any, Any, T],
interval: float = 4.0,
) -> T:
"""
Execute an operation with typing indicator renewal.
Args:
typing_func: Function to call for typing indication
operation_coro: Coroutine to execute while showing typing
interval: Seconds between typing renewals
Returns:
Result of the operation
"""
indicator = TypingIndicator(interval=interval)
try:
# Start typing renewal
await indicator.start(typing_func)
# Execute the operation
result = await operation_coro
return result
finally:
# Always cancel typing renewal and await cleanup
indicator.cancel()
if indicator._task is not None:
try:
await indicator._task
except (asyncio.CancelledError, Exception):
pass
+10 -26
View File
@@ -33,7 +33,6 @@ from ._debounce import InboundDebouncer
from ._ack import AckReactor
from ._unknown_user import UnknownUserHandler, BotContext
from ._pairing_ui import PairingUIBuilder, PairingCallbackHandler
from ..gateway.unicode_utils import safe_error_message, safe_log_message, extract_root_cause_from_error
from ..gateway.pairing import PairingStore
logger = logging.getLogger(__name__)
@@ -223,6 +222,9 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
logger.error(f"Message handler error: {e}")
if self._agent and not message.is_command:
if self.config.typing_indicator:
await update.message.chat.send_action("typing")
# Ack reaction
ack_ctx = None
if self._ack.enabled:
@@ -253,28 +255,11 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
) if update.message.from_user else ""
try:
message_text = await self._debouncer.debounce(user_id, message_text)
# Show typing indicator with renewal during long operation
if self.config.typing_indicator:
from ._typing_indicator import with_typing_renewal
async def _typing_action():
await update.message.chat.send_action("typing")
response = await with_typing_renewal(
typing_func=_typing_action,
operation_coro=self._session.chat(
self._agent, user_id, message_text,
chat_id=str(update.message.chat_id) if update.message.chat_id else "",
user_name=user_name,
)
)
else:
response = await self._session.chat(
self._agent, user_id, message_text,
chat_id=str(update.message.chat_id) if update.message.chat_id else "",
user_name=user_name,
)
response = await self._session.chat(
self._agent, user_id, message_text,
chat_id=str(update.message.chat_id) if update.message.chat_id else "",
user_name=user_name,
)
send_result = self.fire_message_sending(
str(update.message.chat_id), str(response),
reply_to=str(update.message.message_id),
@@ -293,9 +278,8 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
if ack_ctx:
await self._ack.done(ack_ctx, react_fn=_tg_react, unreact_fn=_tg_unreact)
except Exception as e:
logger.error(f"Agent error: {safe_log_message(e)}")
user_error = extract_root_cause_from_error(str(e))
await update.message.reply_text(f"Error: {safe_error_message(user_error)}")
logger.error(f"Agent error: {e}")
await update.message.reply_text(f"Error: {str(e)}")
async def handle_voice(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle voice messages."""
+22 -34
View File
@@ -8,9 +8,29 @@ Custom page: Feature Explorer
"""
import os
import praisonaiui as aiui
from praisonai.integration.host_app import configure_host, create_host_app
from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
# ── Set up datastore bridge ─────────────────────────────────
aiui.set_datastore(PraisonAISessionDataStore())
# ── Dashboard style ─────────────────────────────────────────
aiui.set_style("dashboard")
aiui.set_pages([
"chat", # Core AI chat
"channels", # Messaging platforms
"agents", # Agent management
"skills", # Available tools
"memory", # Agent memory
"knowledge", # Knowledge base
"cron", # Scheduled jobs
"guardrails", # Safety rules
"sessions", # Conversation history
"usage", # Token usage & costs
"config", # Runtime configuration
"logs", # Server logs
"debug", # System debug info
])
# ── Agent definitions ───────────────────────────────────────
AGENTS = [
@@ -51,35 +71,6 @@ AGENTS = [
]
def _build_praison_agents():
from praisonaiagents import Agent
return [
Agent(name=a["name"], instructions=a["instructions"], llm=a["model"])
for a in AGENTS
]
configure_host(
pages=[
"chat",
"channels",
"agents",
"skills",
"memory",
"knowledge",
"cron",
"guardrails",
"sessions",
"usage",
"config",
"logs",
"debug",
],
agents=_build_praison_agents(),
modules=["jobs"],
)
# ── Custom page: Feature Explorer ───────────────────────────
@@ -119,10 +110,7 @@ def _register_agents_in_dashboard():
})
print(f" ✓ Agent: {agent_def['icon']} {agent_def['name']}")
try:
_register_agents_in_dashboard()
except Exception as e:
print(f" ⚠️ Agent registration failed (non-fatal): {e}")
app = create_host_app()
@@ -199,24 +199,93 @@ def _auto_start_services(console, host: str) -> bool:
def _run_aiui_dashboard(port: int, host: str, console) -> bool:
"""Run the aiui dashboard in-process via ``build_host_app()`` (Pattern B)."""
console.print("[bold green]🦞 Starting aiui Dashboard (Pattern B host)...[/bold green]")
"""Run the aiui dashboard interface using proper CLI entrypoint."""
console.print("[bold green]🦞 Starting aiui Dashboard...[/bold green]")
try:
import uvicorn
from praisonai.integration.host_app import build_host_app
except ImportError:
console.print("[red]Error: aiui integration dependencies missing.[/red]")
console.print('[yellow]Install with: pip install "praisonai[ui]"[/yellow]')
import tempfile
import os
# Check if aiui is available first
result = subprocess.run([
sys.executable, "-c", "import praisonaiui"
], capture_output=True, text=True)
if result.returncode != 0:
console.print("[red]Error: aiui package not installed.[/red]")
console.print("[yellow]Install with: pip install aiui[/yellow]")
return False
# Create a temporary script for aiui dashboard based on claw pattern
aiui_script = '''import os
import praisonaiui as aiui
from praisonai.ui._aiui_datastore import PraisonAISessionDataStore
# Set up datastore bridge
aiui.set_datastore(PraisonAISessionDataStore())
# Configure dashboard style
aiui.set_style("dashboard")
aiui.set_branding(title="PraisonAI Unified Dashboard", logo="🌟")
# Set up pages for unified dashboard
aiui.set_pages([
"chat", "agents", "memory", "knowledge",
"skills", "sessions", "usage", "config", "logs"
])
# Register a simple welcome message
@aiui.welcome
async def on_welcome():
await aiui.say("🌟 Welcome to PraisonAI Unified Dashboard!")
await aiui.say("Access all your AI services from one interface.")
# Register basic reply handler
@aiui.reply
async def on_reply(message: str, settings: dict | None = None):
await aiui.think("Processing...")
await aiui.say(f"Unified Dashboard received: {message}")
await aiui.say("Use the sidebar to access Flow, Agents, Memory, and more!")
'''
# Create temp file safely
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(aiui_script)
temp_script = f.name
try:
console.print(f"[green]✓ Starting aiui dashboard on {host}:{port}[/green]")
# Use aiui CLI to run the dashboard (NOT create_app)
# Try aiui command first, fallback to python -m
try:
result = subprocess.run([
"aiui", "run", temp_script, "--port", str(port), "--host", host
], check=True)
except (subprocess.CalledProcessError, FileNotFoundError):
# Fallback to python -m praisonaiui
result = subprocess.run([
sys.executable, "-m", "praisonaiui.cli", "run",
temp_script, "--port", str(port), "--host", host
], check=True)
return True
except subprocess.CalledProcessError as e:
console.print(f"[red]aiui dashboard exited with code {e.returncode}[/red]")
return False
finally:
# Clean up temp file
try:
os.unlink(temp_script)
except (OSError, FileNotFoundError):
pass
except subprocess.CalledProcessError as e:
console.print(f"[red]aiui dashboard failed with code {e.returncode}[/red]")
return False
try:
app = build_host_app(pages=["chat", "sessions", "workflows", "hooks", "usage"])
console.print(f"[green]✓ Starting in-process host on {host}:{port}[/green]")
uvicorn.run(app, host=host, port=port, log_level="info")
return True
except Exception as e:
console.print(f"[red]Error running in-process host: {e}[/red]")
console.print(f"[red]Error running aiui dashboard: {e}[/red]")
return False
@@ -543,10 +612,6 @@ def unified(
"""
Launch the PraisonAI Unified Dashboard.
Pattern B (default with --aiui): in-process ``build_host_app()`` host.
Pattern C: ``praisonai serve gateway`` or ``run_integrated_gateway()``.
Legacy: set ``PRAISONAI_HOST_LEGACY=1`` for callback-only ``@aiui.reply`` mode.
Provides a single interface at localhost:3000 to access:
- Flow Visual Builder (Langflow) - port 7860
- Claw Dashboard (Full UI) - port 8082
@@ -237,23 +237,6 @@ def doctor_cleanup(
raise typer.Exit(0 if removed == len(stale_dirs) else 1)
@app.command("skills")
def doctor_skills(
deep: bool = typer.Option(False, "--deep", help="Enable deeper probes"),
json_output: bool = typer.Option(False, "--json", help="Output JSON"),
show_requirements: bool = typer.Option(False, "--requirements", help="Show detailed requirements"),
):
"""Check skills capabilities and requirements."""
args = ["skills"]
if deep:
args.append("--deep")
if json_output:
args.append("--json")
if show_requirements:
args.append("--requirements")
raise typer.Exit(_run_doctor(args))
@app.command("troubleshoot")
def doctor_troubleshoot():
"""
@@ -17,7 +17,7 @@ app = typer.Typer(
@app.command("start")
def gateway_start(
host: str = typer.Option("127.0.0.1", "--host", help="Host to bind to"),
port: Optional[int] = typer.Option(None, "--port", help="Port to listen on"),
port: int = typer.Option(8765, "--port", help="Port to listen on"),
agents: Optional[str] = typer.Option(None, "--agents", help="Path to agent configuration file"),
config: Optional[str] = typer.Option(None, "--config", help="Path to gateway.yaml for multi-bot mode"),
):
@@ -27,54 +27,17 @@ def gateway_start(
praisonai gateway start
praisonai gateway start --config gateway.yaml
praisonai gateway start --agents agents.yaml --port 9000
GATEWAY_PORT=9000 praisonai gateway start
"""
import os
from ..features.gateway import GatewayHandler
# Check for GATEWAY_PORT environment variable if port not specified
if port is None:
try:
port = int(os.environ.get("GATEWAY_PORT", "8765"))
except ValueError:
port = 8765
handler = GatewayHandler()
handler.start(host=host, port=port, agent_file=agents, config_file=config)
@app.command("stop")
def gateway_stop(
host: str = typer.Option("127.0.0.1", "--host", help="Gateway host"),
port: Optional[int] = typer.Option(None, "--port", help="Gateway port"),
force: bool = typer.Option(False, "--force", help="Force stop (kill process)"),
):
"""Stop a running gateway instance.
Examples:
praisonai gateway stop
praisonai gateway stop --port 9000
praisonai gateway stop --force
"""
import os
from ..features.gateway import GatewayHandler
from ..output.console import get_output_controller
# Check for GATEWAY_PORT environment variable if port not specified
if port is None:
try:
port = int(os.environ.get("GATEWAY_PORT", "8765"))
except ValueError:
port = 8765
handler = GatewayHandler()
handler.stop(host=host, port=port, force=force)
@app.command("status")
def gateway_status(
host: str = typer.Option("127.0.0.1", "--host", help="Gateway host"),
port: Optional[int] = typer.Option(None, "--port", help="Gateway port"),
port: int = typer.Option(8765, "--port", help="Gateway port"),
daemon_only: bool = typer.Option(False, "--daemon-only", help="Show only daemon status"),
):
"""Check gateway status and daemon service status.
@@ -84,18 +47,10 @@ def gateway_status(
praisonai gateway status --port 9000
praisonai gateway status --daemon-only
"""
import os
from ..features.gateway import GatewayHandler
from praisonai.daemon import get_daemon_status
from ..output.console import get_output_controller
# Check for GATEWAY_PORT environment variable if port not specified
if port is None:
try:
port = int(os.environ.get("GATEWAY_PORT", "8765"))
except ValueError:
port = 8765
output = get_output_controller()
# Show daemon status
@@ -439,7 +394,6 @@ Manage the gateway server: praisonai gateway <command>
[bold]Commands:[/bold]
[green]start[/green] Start the gateway server
[green]stop[/green] Stop a running gateway instance
[green]status[/green] Check gateway and daemon status
[green]channels[/green] List channels from gateway.yaml
[green]send[/green] Send a test message to a channel
@@ -113,7 +113,6 @@ Start any PraisonAI server with: praisonai serve <type>
[bold]Server Types:[/bold]
[green]agents[/green] HTTP REST API for agents (port 8000)
[green]gateway[/green] WebSocket multi-agent coordination (port 8765)
[green]ui-gateway[/green] Integrated UI-Gateway (Pattern C) (port 8765)
[green]mcp[/green] MCP server for Claude/Cursor (port 8080)
[green]acp[/green] Agent Client Protocol for IDEs (STDIO)
[green]lsp[/green] Language Server Protocol (STDIO)
@@ -370,42 +369,6 @@ def serve_ui(
raise typer.Exit(4)
@app.command("ui-gateway")
def serve_ui_gateway(
host: str = typer.Option("127.0.0.1", "--host", "-h", help="Host to bind to"),
port: int = typer.Option(8765, "--port", "-p", help="Port to bind to"),
title: str = typer.Option("PraisonAI", "--title", "-t", help="Application title"),
style: str = typer.Option("dashboard", "--style", "-s", help="UI style: dashboard, chat"),
agents_file: Optional[str] = typer.Option(None, "--agents", "-a", help="Agents YAML file"),
):
"""Start integrated UI-Gateway (Pattern C) - aiui with backend bridges.
Combines PraisonAI host integration with AIUIGateway for unified dashboard + chat.
Examples:
praisonai serve ui-gateway
praisonai serve ui-gateway --style chat --port 8765
praisonai serve ui-gateway --agents agents.yaml --title "My App"
"""
output = get_output_controller()
try:
from ...integration.gateway_host import run_integrated_gateway
run_integrated_gateway(
host=host,
port=port,
title=title,
style=style
)
except ImportError as e:
output.print_error(f"UI-Gateway module not available: {e}")
output.print("Install with: pip install praisonai[ui]")
raise typer.Exit(4)
except Exception as e:
output.print_error(f"Failed to start UI-Gateway: {e}")
raise typer.Exit(1)
@app.command("rag")
def serve_rag(
host: str = typer.Option("127.0.0.1", "--host", "-h", help="Host to bind to"),
+1 -6
View File
@@ -13,12 +13,7 @@ from typing import Optional
import typer
app = typer.Typer(
help=(
"🤖 PraisonAI Clean Chat UI (Pattern B host via bundled default_app). "
"Pattern C: AIUIGateway. Legacy: PRAISONAI_HOST_LEGACY=1."
)
)
app = typer.Typer(help="🤖 PraisonAI Clean Chat UI")
UI_DIR = Path.home() / ".praisonai" / "ui"
DEFAULT_APP = UI_DIR / "app.py"
Loaded 100 of 174 files, more files were not shown because too many files have changed in this diff. Show more