Compare commits

...
Author SHA1 Message Date
Mervin Praison 0d2c1bd961 Release v4.5.32 2026-03-10 05:42:59 +00:00
Mervin Praison ef6b1a8913 Release v4.5.31 2026-03-09 14:43:23 +00:00
github-actions[bot] dc9c250adb docs: auto-update api.md [skip ci] 2026-03-09 14:06:13 +00:00
Mervin Praison d1fb090a07 Release v4.5.30 2026-03-09 14:05:55 +00:00
Mervin Praison 8479fffa8c feat(paths): refactor directory management for agents and config loading
Updated various parts of the codebase to centralize directory paths
using the new `paths.py` utility. This improves maintainability and
consistency across the project when accessing directory paths for
plugins, snapshots, memory, and configuration.

Added integration and unit tests to ensure correct behavior of path
configurations.
2026-03-09 13:57:35 +00:00
Mervin Praison 8555b5ad69 Release v4.5.29 2026-03-09 06:18:18 +00:00
Mervin Praison 16f9325176 Release v4.5.28 2026-03-08 07:03:40 +00:00
Mervin Praison fc9f8c5e1d Release v4.5.27 2026-03-08 00:06:51 +00:00
Mervin Praison f5ee866e6c feat(gateway, session): add gateway integration and lazy loading
- Implement lazy loading for WebSocketGateway and GatewaySession in the
  gateway module, allowing easier access without needing wrapper imports.
- Enhance SessionData to include gateway session ID and agent ID.
- Introduce methods in DefaultSessionStore for querying sessions by agent
  and linking sessions to gateway sessions.
2026-03-08 00:02:39 +00:00
Mervin Praison af1c26c471 Release v4.5.26 2026-03-07 07:10:51 +00:00
Mervin Praison f20d5d95e4 feat(workflow): add agent-centric and hybrid workflow examples
Introduced new YAML workflow examples demonstrating agent-centric steps
and hybrid workflows. The agent-centric job workflow supports roles
like Technical Writer and includes quality checks, while the hybrid
workflow combines deterministic steps with multi-agent coordination.
Updated version to 1.5.26 for the praisonai-agents package.
2026-03-07 07:07:56 +00:00
Mervin Praison ac89f8a17c Release v4.5.25 2026-03-07 05:51:43 +00:00
Mervin Praison abc7f09046 feat(workflows): add file-based and YAML-defined actions examples
Added examples for file-based actions (line count and system info)
and YAML-defined actions in new workflow files. Updated relevant
files to support these actions, enhancing the capabilities for
custom workflows. Updated package version to 1.5.25.
2026-03-07 05:32:23 +00:00
github-actions[bot] a0287a9919 docs: auto-update api.md [skip ci] 2026-03-06 14:58:00 +00:00
Mervin Praison 5046e9cf88 Release v4.5.24 2026-03-06 14:57:43 +00:00
Mervin Praison 89624c1580 Release v4.5.23 2026-03-06 07:50:12 +00:00
Mervin Praison cc719e5379 Release v4.5.22 2026-03-05 15:18:22 +00:00
Mervin Praison e854a50fd2 Release v4.5.21 2026-03-03 15:46:18 +00:00
Mervin Praison 99fde9b9d9 Delete tmp directory 2026-03-03 13:20:31 +00:00
Mervin Praison 3c73ca95b6 u 2026-03-03 13:12:44 +00:00
Mervin Praison d7c274489e Release v4.5.20 2026-03-03 11:08:28 +00:00
Mervin Praison 23f6c00e21 Release v4.5.19 2026-03-03 09:37:24 +00:00
Mervin Praison a293999e8b Release v4.5.18 2026-03-03 07:22:28 +00:00
Mervin Praison 0b14ae49ef update 2026-03-01 07:12:47 +00:00
Mervin Praison b33c649de5 Release v4.5.17 2026-02-28 08:47:18 +00:00
Mervin Praison 02a1977651 Release v4.5.16 2026-02-27 04:43:28 +00:00
107 changed files with 8904 additions and 574 deletions

No files matched your search

+3 -1
View File
@@ -91,4 +91,6 @@ issue_details.json
target/
Cargo.lock
**/*.rs.bk
*.pdb
*.pdb
.windsurf
.agent
+3
View File
@@ -906,6 +906,7 @@ Methods:
* <code title="cli">praisonai package install <a href="./src/praisonai/praisonai/cli/commands/package.py">--help</a></code>
* <code title="cli">praisonai package list <a href="./src/praisonai/praisonai/cli/commands/package.py">--help</a></code>
* <code title="cli">praisonai package uninstall <a href="./src/praisonai/praisonai/cli/commands/package.py">--help</a></code>
* <code title="cli">praisonai paths show <a href="./src/praisonai/praisonai/cli/commands/paths.py">--help</a></code>
* <code title="cli">praisonai plugins create <a href="./src/praisonai/praisonai/cli/commands/plugins.py">--help</a></code>
* <code title="cli">praisonai plugins disable <a href="./src/praisonai/praisonai/cli/commands/plugins.py">--help</a></code>
* <code title="cli">praisonai plugins discover <a href="./src/praisonai/praisonai/cli/commands/plugins.py">--help</a></code>
@@ -923,6 +924,8 @@ Methods:
* <code title="cli">praisonai profile snapshot <a href="./src/praisonai/praisonai/cli/commands/profile.py">--help</a></code>
* <code title="cli">praisonai profile startup <a href="./src/praisonai/praisonai/cli/commands/profile.py">--help</a></code>
* <code title="cli">praisonai profile suite <a href="./src/praisonai/praisonai/cli/commands/profile.py">--help</a></code>
* <code title="cli">praisonai publish publish-main <a href="./src/praisonai/praisonai/cli/commands/publish.py">--help</a></code>
* <code title="cli">praisonai publish pypi <a href="./src/praisonai/praisonai/cli/commands/publish.py">--help</a></code>
* <code title="cli">praisonai rag chat <a href="./src/praisonai/praisonai/cli/commands/rag.py">--help</a></code>
* <code title="cli">praisonai rag eval <a href="./src/praisonai/praisonai/cli/commands/rag.py">--help</a></code>
* <code title="cli">praisonai rag index <a href="./src/praisonai/praisonai/cli/commands/rag.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.5.15" \
"praisonai>=4.5.32" \
"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.5.15" \
"praisonai>=4.5.32" \
"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.5.15" \
"praisonai>=4.5.32" \
"praisonai[ui]" \
"praisonai[crewai]"
+23
View File
@@ -0,0 +1,23 @@
"""Calculator with Tests - Editor Output Example.
Builds a calculator module, writes unit tests, runs them, generates a
coverage-style report, and reads it back.
"""
from praisonaiagents import Agent
from praisonaiagents.tools import (
write_file, read_file, execute_command
)
agent = Agent(
instructions="You are a helpful coding assistant.",
output="editor",
tools=[write_file, read_file, execute_command],
approval=True,
)
agent.start(
"Build a mini calculator - create /tmp/calculator.py with add, subtract, "
"multiply, divide functions, write unit tests in /tmp/test_calculator.py "
"using unittest, run the tests to make sure they pass, then generate a "
"coverage-style report in /tmp/calc_report.md listing which functions were "
"tested and read it back"
)
+22
View File
@@ -0,0 +1,22 @@
"""Fibonacci Generator - Editor Output Example.
Creates a Fibonacci script, runs it, writes a summary, and reads back
both files to verify everything looks good.
"""
from praisonaiagents import Agent
from praisonaiagents.tools import (
write_file, read_file, execute_command, list_directory
)
agent = Agent(
instructions="You are a helpful coding assistant.",
output="editor",
tools=[write_file, read_file, execute_command, list_directory],
approval=True,
)
agent.start(
"Create a Python file at /tmp/fibonacci.py that generates the first 20 "
"Fibonacci numbers, run it to verify the output is correct, then write "
"a summary to /tmp/fib_summary.md explaining the algorithm you used and "
"the output, finally read back both files to confirm everything looks good"
)
+36
View File
@@ -0,0 +1,36 @@
"""Multi-Step Research - Editor Output Example.
Researches Python frameworks, creates comparison code, runs it, writes a
markdown report, reads it back, lists files, and gets system info.
"""
from praisonaiagents import Agent
from praisonaiagents.tools import (
write_file, read_file, execute_command, list_directory, get_system_info
)
try:
from praisonaiagents.tools import web_search
tools = [web_search, write_file, read_file, execute_command,
list_directory, get_system_info]
except ImportError:
tools = [write_file, read_file, execute_command,
list_directory, get_system_info]
agent = Agent(
instructions="You are a helpful research assistant.",
output="editor",
tools=tools,
approval=True,
)
agent.start(
"Research the top 3 Python web frameworks (Django, FastAPI, Flask), then: "
"1) Search the web for each framework's latest version and key features, "
"2) Create a file called /tmp/framework_comparison.py that contains a "
"Python dictionary with the comparison data, "
"3) Execute the code to verify the dictionary is valid, "
"4) Write a markdown report to /tmp/framework_report.md summarizing your "
"findings in a table format, "
"5) Read back the report file to verify it was written correctly, "
"6) List the files in /tmp to confirm both files exist, "
"7) Get system info to note what OS this report was generated on"
)
+21
View File
@@ -0,0 +1,21 @@
"""System Audit - Editor Output Example.
Runs system commands, gathers info, writes a report, and reads it back.
"""
from praisonaiagents import Agent
from praisonaiagents.tools import (
write_file, read_file, execute_command
)
agent = Agent(
instructions="You are a helpful system administration assistant.",
output="editor",
tools=[write_file, read_file, execute_command],
approval=True,
)
agent.start(
"Investigate my system - get the current date, check disk usage, find out "
"what Python version is installed, list running processes sorted by memory, "
"then write all findings to /tmp/system_audit.md as a neat report and read "
"it back to verify"
)
+28
View File
@@ -0,0 +1,28 @@
"""Weather Search - Editor Output Example.
Searches for weather, creates JSON data, writes a parser script, executes
it, and lists created files.
"""
from praisonaiagents import Agent
from praisonaiagents.tools import (
write_file, execute_command, list_directory
)
try:
from praisonaiagents.tools import web_search
tools = [web_search, write_file, execute_command, list_directory]
except ImportError:
tools = [write_file, execute_command, list_directory]
agent = Agent(
instructions="You are a helpful research assistant.",
output="editor",
tools=tools,
approval=True,
)
agent.start(
"Search the web for the current weather in Tokyo, create a JSON file at "
"/tmp/weather_data.json with the findings, then write a Python script at "
"/tmp/parse_weather.py that reads the JSON and prints a formatted weather "
"report, execute the script, and list all files you created in /tmp"
)
+5
View File
@@ -0,0 +1,5 @@
#!/bin/bash
# Calculator with tests: code + unit tests + test execution + coverage report
# Tests: multiple file creation, test execution, report generation, file reading
praisonai "Build a mini calculator - create /tmp/calculator.py with add, subtract, multiply, divide functions, write unit tests in /tmp/test_calculator.py using unittest, run the tests to make sure they pass, then generate a coverage-style report in /tmp/calc_report.md listing which functions were tested and read it back" --trust
+5
View File
@@ -0,0 +1,5 @@
#!/bin/bash
# Fibonacci generator: code creation, execution, documentation, verification
# Tests: file creation, code execution, markdown writing, file reading
praisonai "Create a Python file at /tmp/fibonacci.py that generates the first 20 Fibonacci numbers, run it to verify the output is correct, then write a summary to /tmp/fib_summary.md explaining the algorithm you used and the output, finally read back both files to confirm everything looks good" --trust
+5
View File
@@ -0,0 +1,5 @@
#!/bin/bash
# Multi-step research: web search, file creation, code execution, reporting
# Tests: web search, file write, code execution, markdown report, file read, directory listing, system info
praisonai "Research the top 3 Python web frameworks (Django, FastAPI, Flask), then: 1) Search the web for each framework's latest version and key features, 2) Create a file called /tmp/framework_comparison.py that contains a Python dictionary with the comparison data, 3) Execute the code to verify the dictionary is valid, 4) Write a markdown report to /tmp/framework_report.md summarizing your findings in a table format, 5) Read back the report file to verify it was written correctly, 6) List the files in /tmp to confirm both files exist, 7) Get system info to note what OS this report was generated on, 8) Analyze the Python code you wrote for any issues, 9) Format the Python code properly, 10) Finally search for any recent news about Python 3.13 features" --trust
+5
View File
@@ -0,0 +1,5 @@
#!/bin/bash
# System audit: multiple command executions, report generation
# Tests: command execution (date, disk, python, processes), file creation, file reading
praisonai "Investigate my system - get the current date, check disk usage, find out what Python version is installed, list running processes sorted by memory, then write all findings to /tmp/system_audit.md as a neat report and read it back to verify" --trust
+5
View File
@@ -0,0 +1,5 @@
#!/bin/bash
# Todo app: multi-file app creation, execution, documentation, verification
# Tests: multiple file creation, code execution, markdown docs, directory listing
praisonai "Write a todo list app at /tmp/todo.py with functions to add, remove, and list tasks stored in /tmp/todos.json, run the script with some sample tasks added, then create documentation at /tmp/todo_docs.md explaining usage with examples, and verify all three files exist" --trust
+5
View File
@@ -0,0 +1,5 @@
#!/bin/bash
# Weather search: web search, JSON creation, Python script, execution
# Tests: web search, JSON file creation, Python scripting, code execution, directory listing
praisonai "Search the web for the current weather in Tokyo, create a JSON file at /tmp/weather_data.json with the findings, then write a Python script at /tmp/parse_weather.py that reads the JSON and prints a formatted weather report, execute the script, and list all files you created in /tmp" --trust
+41
View File
@@ -0,0 +1,41 @@
"""
Example file-based action: line-count
Counts lines in files matching a pattern. Demonstrates how actions
receive step config and flags from the workflow YAML.
Usage in workflow:
- name: Count lines
action: line-count
pattern: "*.py"
directory: src/
"""
import os
from pathlib import Path
def run(step, flags, cwd):
"""Count lines in files matching a glob pattern."""
pattern = step.get("pattern", "*.py")
directory = step.get("directory", ".")
target = Path(cwd) / directory
if not target.exists():
return {"ok": False, "error": f"Directory not found: {target}"}
total_lines = 0
file_count = 0
for filepath in target.rglob(pattern):
if filepath.is_file():
try:
total_lines += len(filepath.read_text().splitlines())
file_count += 1
except (UnicodeDecodeError, PermissionError):
continue
return {
"ok": True,
"output": f"{file_count} files, {total_lines} lines ({pattern} in {directory})",
}
+33
View File
@@ -0,0 +1,33 @@
"""
Example file-based action: system-info
Collects basic system information. Demonstrates the action file contract:
- Must define run(step, flags, cwd) → dict
- Return {"ok": True, "output": "..."} on success
- Return {"ok": False, "error": "..."} on failure
Place in: actions/system_info.py (next to workflow)
or: .praison/actions/system_info.py (project-level)
"""
import platform
import os
def run(step, flags, cwd):
"""Collect system information."""
info_parts = [
f"OS: {platform.system()} {platform.release()}",
f"Python: {platform.python_version()}",
f"Machine: {platform.machine()}",
f"CWD: {cwd}",
]
# Optional: include env vars if requested via step config
if step.get("show_env"):
env_vars = step.get("show_env", "").split(",")
for var in env_vars:
var = var.strip()
info_parts.append(f"{var}: {os.environ.get(var, '(not set)')}")
return {"ok": True, "output": " | ".join(info_parts)}
+33
View File
@@ -0,0 +1,33 @@
# Example: File-Based Actions (Option A)
#
# Actions loaded from .py files in the actions/ directory next to this workflow.
# Each action file exports: run(step, flags, cwd) → dict
#
# Directory structure:
# examples/workflows/
# ├── file-actions-demo.yaml ← this file
# └── actions/
# ├── system_info.py ← action: system-info
# └── line_count.py ← action: line-count
#
# Usage:
# praisonai workflow run file-actions-demo.yaml
# praisonai workflow run file-actions-demo.yaml --dry-run
type: job
name: file-actions-demo
description: Demonstrates file-based custom actions from actions/ directory
steps:
- name: System info
action: system-info
- name: Count Python files
action: line-count
pattern: "*.py"
directory: "."
- name: Count YAML files
action: line-count
pattern: "*.yaml"
directory: "."
+40
View File
@@ -0,0 +1,40 @@
# Example Step Workflow: Publish to PyPI
# Usage:
# praisonai workflow run publish-pypi.yaml --dry-run
# praisonai workflow run publish-pypi.yaml
# praisonai workflow run publish-pypi.yaml --major
# praisonai workflow run publish-pypi.yaml --minor
# praisonai workflow run publish-pypi.yaml --no-bump
type: job
name: publish-pypi
description: Publish package to PyPI with automatic version bump
vars:
token_var:
default: PYPI_TOKEN
description: Environment variable for PyPI token
flags:
major: { description: "Bump major version (X.0.0)" }
minor: { description: "Bump minor version (X.Y.0)" }
no-bump: { description: "Skip version bump" }
steps:
- name: Bump version
action: bump-version
file: pyproject.toml
strategy: patch
if: "{{ not flags.no_bump }}"
- name: Clean dist
run: rm -rf dist
- name: Lock dependencies
run: uv lock
- name: Build package
run: uv build
- name: Publish to PyPI
run: uv publish --token ${{ env.PYPI_TOKEN }}
+40
View File
@@ -0,0 +1,40 @@
# Example: YAML-Defined Actions (Option B)
#
# Actions defined inline in the workflow — self-contained, portable.
# No external files needed.
#
# Usage:
# praisonai workflow run yaml-actions-demo.yaml
# praisonai workflow run yaml-actions-demo.yaml --dry-run
type: job
name: yaml-actions-demo
description: Demonstrates YAML-defined custom actions
# Define reusable actions inline
actions:
check-python:
run: python3 --version
count-files:
script: |
import os
cwd = os.getcwd()
py_files = [f for f in os.listdir(cwd) if f.endswith('.py')]
yaml_files = [f for f in os.listdir(cwd) if f.endswith('.yaml') or f.endswith('.yml')]
result = f"Python: {len(py_files)}, YAML: {len(yaml_files)}"
timestamp:
script: |
from datetime import datetime
result = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
steps:
- name: Check Python version
action: check-python
- name: Count project files
action: count-files
- name: Get timestamp
action: timestamp
@@ -0,0 +1,70 @@
# Agent-Centric Job Workflow Example
#
# This workflow demonstrates the new agent-centric step types:
# - agent: AI agent execution
# - judge: AI quality gate with threshold
# - approve: AI or human approval gate
#
# Run with: praisonai workflow run agent-centric-job-workflow.yaml
# Dry run: praisonai workflow run agent-centric-job-workflow.yaml --dry-run
type: job
name: smart-release
description: Intelligent release workflow with AI-powered steps
flags:
major:
description: Bump major version
minor:
description: Bump minor version
skip-approval:
description: Skip approval gate (auto-approve)
steps:
# Step 1: Deterministic - Bump version
- name: Bump version
action: bump-version
strategy: patch
# Step 2: Agent - Generate changelog
- name: Generate changelog
agent:
role: Technical Writer
instructions: |
You are a technical writer who creates clear, concise changelogs.
Focus on user-facing changes, bug fixes, and new features.
prompt: |
Generate a changelog entry for the latest release.
Include sections for: Features, Bug Fixes, and Breaking Changes.
Keep it concise and professional.
model: gpt-4o-mini
output_file: CHANGELOG_ENTRY.md
# Step 3: Judge - Quality check
- name: Quality check
judge:
input_file: CHANGELOG_ENTRY.md
criteria: |
- Completeness: All sections present (Features, Bug Fixes, Breaking Changes)
- Clarity: Easy to understand for end users
- Formatting: Proper markdown structure
- Conciseness: Not overly verbose
threshold: 7.0
on_fail: warn
model: gpt-4o-mini
# Step 4: Approve - Human approval gate
- name: Approve release
approve:
description: Review and approve the release
risk_level: medium
auto_approve: "{{ flags.skip_approval }}"
# Step 5: Deterministic - Build
- name: Build package
run: echo "Building package..."
# Step 6: Deterministic - Publish (conditional)
- name: Publish to PyPI
run: echo "Publishing to PyPI..."
if: "{{ not flags.skip_approval }}"
+89
View File
@@ -0,0 +1,89 @@
# Hybrid Workflow Example
#
# This workflow demonstrates the `type: hybrid` feature that combines:
# - Deterministic steps (run, action) from job workflows
# - Agent-centric steps (agent, judge, approve) from job workflows
# - Multi-agent workflow steps (workflow, parallel)
#
# Run with: praisonai workflow run hybrid-workflow.yaml
# Dry run: praisonai workflow run hybrid-workflow.yaml --dry-run
type: hybrid
name: hybrid-release-pipeline
description: Hybrid workflow combining shell commands, AI agents, and multi-agent steps
# Define agents for multi-agent workflow steps
agents:
researcher:
name: Researcher
role: Research Analyst
goal: Research and gather information
instructions: Provide concise, factual research findings
model: gpt-4o-mini
writer:
name: Writer
role: Content Writer
goal: Write clear, engaging content
instructions: Write professional documentation
model: gpt-4o-mini
flags:
skip-tests:
description: Skip test execution
auto-approve:
description: Auto-approve deployment
steps:
# Step 1: Deterministic - Check environment
- name: Check environment
run: echo "Checking environment..." && python --version
# Step 2: Agent - Generate release notes
- name: Generate release notes
agent:
role: Technical Writer
instructions: You write clear, concise release notes
prompt: Generate release notes for version 1.2.0 with bug fixes and new features
model: gpt-4o-mini
output_file: RELEASE_NOTES.md
# Step 3: Parallel - Run multiple checks simultaneously
- name: Parallel checks
parallel:
- run: echo "Running lint check..."
- run: echo "Running type check..."
- run: echo "Running security scan..."
# Step 4: Multi-agent workflow step
- name: Research and document
workflow:
agent: researcher
action: Research best practices for release management
# Step 5: Judge - Quality gate
- name: Quality check
judge:
input_file: RELEASE_NOTES.md
criteria: |
- Completeness: All sections present
- Clarity: Easy to understand
- Professional tone
threshold: 7.0
on_fail: warn
# Step 6: Approve - Human approval gate
- name: Approve release
approve:
description: Review and approve the release
risk_level: medium
auto_approve: "{{ flags.auto_approve }}"
# Step 7: Deterministic - Build
- name: Build package
run: echo "Building package..."
if: "{{ not flags.skip_tests }}"
# Step 8: Final notification
- name: Notify completion
run: echo "✅ Hybrid workflow completed successfully!"
+15 -3
View File
@@ -123,6 +123,8 @@ Pattern: Each major module has a protocols.py file
- Protocols define WHAT (interface contract)
- Adapters implement HOW (concrete implementation)
- Naming: XProtocol for interfaces, XAdapter for implementations
(ALWAYS suffix with "Protocol" — never use bare concept names like TraceSink)
- File placement: All protocols live in protocols.py within their module
- Users can implement any protocol for custom behavior
```
@@ -286,9 +288,19 @@ save() / load() # Disk persistence (NOT store)
# Configuration
class XConfig: # Configuration dataclass (MemoryConfig, HooksConfig)
# Protocols
class XProtocol(Protocol): # Abstract interface
class XAdapter: # Implementation
# Protocols — ALWAYS suffix with "Protocol"
class XProtocol(Protocol): # Abstract interface (TraceSinkProtocol, MemoryProtocol)
class XAdapter: # Concrete implementation (ChromaAdapter, FileAdapter)
# ✅ CORRECT protocol naming:
class MemoryProtocol(Protocol): ... # Clear it's a protocol
class SessionStoreProtocol(Protocol): ... # Descriptive + Protocol suffix
class TraceSinkProtocol(Protocol): ... # In protocols.py + suffixed
# ❌ WRONG protocol naming:
class TraceSink(Protocol): ... # Missing Protocol suffix — ambiguous
class IMemory(Protocol): ... # TypeScript convention, not Python
class MemoryBase(Protocol): ... # "Base" implies ABC, not Protocol
# Execution
run() # Synchronous, blocking
@@ -438,6 +438,7 @@ class Agent:
hooks: Optional[Union[List[Any], Any]] = None, # Union[list, HooksConfig]
skills: Optional[Union[List[str], Any]] = None, # Union[list, SkillsConfig]
approval: Optional[Union[bool, Any]] = None, # Union[bool, ApprovalProtocol backend]
tool_timeout: Optional[int] = None, # P8/G11: Timeout in seconds for each tool call
):
"""Initialize an Agent instance.
@@ -713,6 +714,7 @@ class Agent:
json_output = getattr(_output_config, 'json_output', False)
status_trace = getattr(_output_config, 'status_trace', False) # New: clean inline status
simple_output = getattr(_output_config, 'simple_output', False) # status preset: no timestamps
editor_output = getattr(_output_config, 'editor_output', False) # Editor: Step N display
output_file = getattr(_output_config, 'output_file', None) # Auto-save to file
output_template = getattr(_output_config, 'template', None) # Response template
else:
@@ -722,10 +724,20 @@ class Agent:
json_output = False
status_trace = False
simple_output = False
editor_output = False
# Enable trace output mode if configured (takes priority)
# Enable editor output mode if configured (beginner-friendly, takes priority)
# Shows: Step 1: 📄 Creating file: path → ✓ Done
if editor_output:
try:
from ..output.editor import enable_editor_output, is_editor_output_enabled
if not is_editor_output_enabled():
enable_editor_output(use_color=True)
except ImportError:
pass # Editor module not available
# Enable trace output mode if configured
# This provides timestamped inline status with duration
if status_trace:
elif status_trace:
try:
from ..output.trace import enable_trace_output, is_trace_output_enabled
if not is_trace_output_enabled():
@@ -1343,6 +1355,37 @@ class Agent:
self.tools = list(tools)
else:
self.tools = tools or []
# Inject default tools for autonomy mode (after self.tools is initialized)
# ONLY inject if caller didn't provide tools - avoid duplicates with CLI/wrapper tools
if self.autonomy_enabled and hasattr(self, '_autonomy_config_obj'):
config = self._autonomy_config_obj
if config.default_tools is not None:
# User provided custom default tools via AutonomyConfig
self.tools.extend(config.default_tools)
elif not self.tools:
# Only inject AUTONOMY_PROFILE if no tools were provided by caller
# This prevents duplicates when CLI/wrapper already provides tools
try:
from ..tools.profiles import AUTONOMY_PROFILE
from .. import tools as tools_module
resolved_tools = []
for tool_name in AUTONOMY_PROFILE.tools:
try:
tool = getattr(tools_module, tool_name)
if tool is not None:
resolved_tools.append(tool)
except AttributeError:
pass # Tool not available, skip
self.tools.extend(resolved_tools)
except ImportError:
# Fallback to ast-grep tools if profiles not available
try:
from ..tools.ast_grep_tool import get_ast_grep_tools
self.tools.extend(get_ast_grep_tools())
except ImportError:
pass # No default tools available
self.max_iter = max_iter
self.max_rpm = max_rpm
self.max_execution_time = max_execution_time
@@ -1463,6 +1506,9 @@ Your Goal: {self.goal}
# Pending approvals for async (non-blocking) mode
self._pending_approvals = {}
# P8/G11: Tool timeout - prevent slow tools from blocking
self._tool_timeout = tool_timeout
# Cache for system prompts and formatted tools with lazy thread-safe lock
self._system_prompt_cache = {}
self._formatted_tools_cache = {}
@@ -1962,6 +2008,7 @@ Summary:"""
"clear_context": config.clear_context,
"track_changes": config.effective_track_changes,
"snapshot_dir": config.snapshot_dir,
"default_tools": config.default_tools,
}
# Also preserve any extra user-provided keys from dict input
if isinstance(autonomy, dict):
@@ -2270,13 +2317,16 @@ Summary:"""
)
# Take initial snapshot before autonomous execution starts
if self._file_snapshot is not None:
# NOTE: Snapshot tracking is disabled by default for performance (G12 fix)
# FileSnapshot.track() walks entire project and is too slow for large codebases
# Enable with autonomy={"snapshot": True} if needed
if self._file_snapshot is not None and self.autonomy_config.get("snapshot", False):
try:
snap_info = self._file_snapshot.track(message="pre-autonomous")
self._snapshot_stack.append(snap_info.commit_hash)
self._redo_stack.clear()
except Exception as e:
logger.debug(f"Pre-autonomous snapshot failed: {e}")
logging.debug(f"Pre-autonomous snapshot failed: {e}")
start_time = time_module.time()
started_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
@@ -2303,11 +2353,21 @@ Summary:"""
# Reset doom loop tracker for new task
self._reset_doom_loop()
# P3/G2: Import callback helper for autonomy events
from ..main import execute_sync_callback
try:
# Execute the autonomous loop
while iterations < effective_max_iter:
iterations += 1
# P3/G2: Emit autonomy_iteration callback for CLI visibility
execute_sync_callback('autonomy_iteration',
iteration=iterations,
max_iterations=effective_max_iter,
stage=stage
)
# Check timeout
if timeout_seconds and (time_module.time() - start_time) > timeout_seconds:
return AutonomyResult(
@@ -2324,6 +2384,13 @@ Summary:"""
# Check doom loop with graduated recovery (G-RECOVERY-2 fix)
if self._is_doom_loop():
recovery = self._get_doom_recovery()
# P3/G2: Emit doom loop callback for CLI visibility
execute_sync_callback('autonomy_doom_loop',
iteration=iterations,
recovery_action=recovery
)
obs = getattr(self, '_observability_hooks', None)
if obs is not None:
from ..escalation.observability import EventType as _EvtType
@@ -2337,6 +2404,12 @@ Summary:"""
if self._doom_loop_tracker is not None:
self._doom_loop_tracker.clear_actions()
continue
elif recovery == "escalate_model":
# Give the agent one more try with explicit error guidance
prompt = prompt + "\n\n[System: You are stuck in a loop. CRITICAL: Check that all tool argument names exactly match the function signature. Do NOT add '=' to argument names. Use only the documented parameter names.]"
if self._doom_loop_tracker is not None:
self._doom_loop_tracker.clear_actions()
continue
elif recovery == "request_help":
return AutonomyResult(
success=False,
@@ -2380,8 +2453,13 @@ Summary:"""
response_str = str(response)
# Record the action for doom loop tracking (G1 fix: was missing)
# Use iteration number + response hash to avoid false positives
# when same prompt is re-injected (G8 fix: doom loop false positive)
self._record_action(
"chat", {"prompt": prompt}, response_str[:200], True
"chat",
{"iteration": iterations, "response_hash": hash(response_str[:500])},
response_str[:200],
True
)
# Record for action history
@@ -2417,14 +2495,28 @@ Summary:"""
stage_order = ["direct", "heuristic", "planned", "autonomous"]
idx = stage_order.index(stage) if stage in stage_order else 0
if idx < len(stage_order) - 1:
prev_stage = stage
stage = stage_order[idx + 1]
# P3/G2: Emit stage change callback for CLI visibility
execute_sync_callback('autonomy_stage_change',
from_stage=prev_stage,
to_stage=stage
)
if obs is not None:
obs.emit(_EvtType.STAGE_ESCALATE, {"from": stage_order[idx], "to": stage})
obs.emit(_EvtType.STAGE_ESCALATE, {"from": prev_stage, "to": stage})
# Check for completion promise FIRST (structured signal)
if effective_promise:
promise_tag = f"<promise>{effective_promise}</promise>"
if promise_tag in response_str:
# P3/G2: Emit completion callback
execute_sync_callback('autonomy_complete',
completion_reason="promise",
iterations=iterations,
duration_seconds=time_module.time() - start_time
)
return AutonomyResult(
success=True,
output=response_str,
@@ -2438,6 +2530,12 @@ Summary:"""
# Check for keyword-based completion signals (word-boundary, G-COMPLETION-1 fix)
if self._is_completion_signal(response_str):
# P3/G2: Emit completion callback
execute_sync_callback('autonomy_complete',
completion_reason="goal",
iterations=iterations,
duration_seconds=time_module.time() - start_time
)
return AutonomyResult(
success=True,
output=response_str,
@@ -2451,6 +2549,12 @@ Summary:"""
# For DIRECT stage, complete after first response
if stage == "direct":
# P3/G2: Emit completion callback
execute_sync_callback('autonomy_complete',
completion_reason="goal",
iterations=iterations,
duration_seconds=time_module.time() - start_time
)
return AutonomyResult(
success=True,
output=response_str,
@@ -2467,6 +2571,12 @@ Summary:"""
self.clear_history()
# Max iterations reached
# P3/G2: Emit completion callback for max iterations
execute_sync_callback('autonomy_complete',
completion_reason="max_iterations",
iterations=iterations,
duration_seconds=time_module.time() - start_time
)
return AutonomyResult(
success=False,
output="Max iterations reached",
@@ -2620,6 +2730,11 @@ Summary:"""
if self._doom_loop_tracker is not None:
self._doom_loop_tracker.clear_actions()
continue
elif recovery == "escalate_model":
prompt = prompt + "\n\n[System: You are stuck in a loop. CRITICAL: Check that all tool argument names exactly match the function signature. Do NOT add '=' to argument names. Use only the documented parameter names.]"
if self._doom_loop_tracker is not None:
self._doom_loop_tracker.clear_actions()
continue
elif recovery == "request_help":
return AutonomyResult(
success=False,
@@ -2663,8 +2778,13 @@ Summary:"""
response_str = str(response)
# Record the action for doom loop tracking (G2 fix: was missing)
# Use iteration number + response hash to avoid false positives
# when same prompt is re-injected (G8 fix: doom loop false positive)
self._record_action(
"chat", {"prompt": prompt}, response_str[:200], True
"chat",
{"iteration": iterations, "response_hash": hash(response_str[:500])},
response_str[:200],
True
)
# Record for action history
@@ -4274,8 +4394,26 @@ Your Goal: {self.goal}"""
try:
sig = inspect.signature(func)
valid_params = set(sig.parameters.keys()) - {'self'}
casted_args = {}
# Sanitize argument names: strip trailing '=', whitespace, and
# other invalid chars that LLMs sometimes hallucinate in kwarg names
sanitized = {}
for raw_name, arg_value in arguments.items():
clean = raw_name.strip().rstrip('=').strip()
# If the cleaned name matches a valid param, use it;
# otherwise try case-insensitive match
if clean in valid_params:
sanitized[clean] = arg_value
elif clean.lower() in {p.lower() for p in valid_params}:
# Case-insensitive fuzzy match
matched = next(p for p in valid_params if p.lower() == clean.lower())
sanitized[matched] = arg_value
else:
sanitized[clean] = arg_value
arguments = sanitized
for param_name, arg_value in arguments.items():
if param_name in sig.parameters:
param = sig.parameters[param_name]
@@ -4362,7 +4500,19 @@ Your Goal: {self.goal}"""
arguments.update(res.output.modified_data)
with with_injection_context(state):
result = self._execute_tool_impl(function_name, arguments)
# P8/G11: Apply tool timeout if configured
tool_timeout = getattr(self, '_tool_timeout', None)
if tool_timeout and tool_timeout > 0:
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(self._execute_tool_impl, function_name, arguments)
try:
result = future.result(timeout=tool_timeout)
except concurrent.futures.TimeoutError:
logging.warning(f"Tool {function_name} timed out after {tool_timeout}s")
result = {"error": f"Tool timed out after {tool_timeout}s", "timeout": True}
else:
result = self._execute_tool_impl(function_name, arguments)
# Apply tool output truncation to prevent context overflow
# Uses context manager budget if enabled, otherwise applies default limit
@@ -4415,6 +4565,13 @@ Your Goal: {self.goal}"""
)
self._hook_runner.execute_sync(HookEvent.AFTER_TOOL, after_tool_input, target=function_name)
# G10 fix: Mark progress after successful tool execution
# This prevents false doom loop detection when tools succeed
if self._doom_loop_tracker is not None and result is not None:
is_error = isinstance(result, dict) and result.get('error')
if not is_error:
self._doom_loop_tracker.mark_progress(f"tool:{function_name}")
return result
except Exception as e:
# Emit tool call end with error
@@ -4566,6 +4723,9 @@ Your Goal: {self.goal}"""
error_msg = f"Tool execution denied: {decision.reason}"
logging.warning(error_msg)
return {"error": error_msg, "approval_denied": True}
# Bridge: mark tool as approved in global registry context so
# @require_approval decorator wrappers skip their own check.
get_approval_registry().mark_approved(function_name)
if decision.modified_args:
arguments = decision.modified_args
logging.info(f"Using modified arguments: {arguments}")
@@ -5006,7 +5166,9 @@ Your Goal: {self.goal}"""
"display_fn": self._display_generating if self.verbose else None,
"reasoning_steps": reasoning_steps,
"verbose": self.verbose,
"max_iterations": 10
"max_iterations": 10,
"stream_callback": self.stream_emitter.emit,
"emit_events": True,
}
if response_format:
chat_kwargs["response_format"] = response_format
@@ -7046,8 +7208,13 @@ Write the complete compiled report:"""
logging.debug(f"Auto-memory extraction failed: {e}")
def _auto_save_session(self):
"""Auto-save session if auto_save is enabled."""
if not self.auto_save or not self._memory_instance:
"""Auto-save session if auto_save is enabled.
G-1 FIX: Routes to SessionStore instead of Memory.save_session() to
maintain clean separation between conversation history (SessionStore)
and semantic memory (Memory).
"""
if not self.auto_save:
return
try:
@@ -7057,12 +7224,25 @@ Write the complete compiled report:"""
for msg in self.chat_history
]
self._memory_instance.save_session(
name=self.auto_save,
conversation_history=clean_history,
metadata={"agent_name": self.name, "user_id": self.user_id}
)
logging.debug(f"Auto-saved session: {self.auto_save}")
# G-1 FIX: Use SessionStore for conversation history persistence
# This maintains clean separation: Memory = facts, SessionStore = turns
if self._session_store is not None:
# Persist each message to SessionStore
for msg in clean_history:
self._session_store.add_message(
self.auto_save, # Use auto_save name as session_id
role=msg.get("role", "user"),
content=msg.get("content", ""),
)
logging.debug(f"Auto-saved session to SessionStore: {self.auto_save}")
elif self._memory_instance and hasattr(self._memory_instance, 'save_session'):
# Fallback to Memory.save_session() for backward compatibility
self._memory_instance.save_session(
name=self.auto_save,
conversation_history=clean_history,
metadata={"agent_name": self.name, "user_id": self.user_id}
)
logging.debug(f"Auto-saved session to Memory: {self.auto_save}")
except Exception as e:
logging.debug(f"Error auto-saving session: {e}")
@@ -7259,11 +7439,31 @@ Write the complete compiled report:"""
if formatted_tools:
completion_args["tools"] = formatted_tools
# Import StreamEvent types for event emission
from ..streaming.events import StreamEvent, StreamEventType
import time as time_module
# Emit REQUEST_START event
request_start_perf = time_module.perf_counter()
self.stream_emitter.emit(StreamEvent(
type=StreamEventType.REQUEST_START,
timestamp=request_start_perf,
metadata={"model": self.llm, "message_count": len(messages)}
))
completion = self._openai_client.sync_client.chat.completions.create(**completion_args)
# Emit HEADERS_RECEIVED event
self.stream_emitter.emit(StreamEvent(
type=StreamEventType.HEADERS_RECEIVED,
timestamp=time_module.perf_counter()
))
# Stream the response chunks without display
response_text = ""
tool_calls_data = []
first_token_emitted = False
last_content_time = None
for chunk in completion:
delta = chunk.choices[0].delta
@@ -7272,6 +7472,24 @@ Write the complete compiled report:"""
if delta.content is not None:
chunk_content = delta.content
response_text += chunk_content
last_content_time = time_module.perf_counter()
# Emit FIRST_TOKEN on first content
if not first_token_emitted:
self.stream_emitter.emit(StreamEvent(
type=StreamEventType.FIRST_TOKEN,
timestamp=last_content_time,
content=chunk_content
))
first_token_emitted = True
else:
# Emit DELTA_TEXT for subsequent tokens
self.stream_emitter.emit(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=last_content_time,
content=chunk_content
))
yield chunk_content
# Handle tool calls (accumulate but don't yield as chunks)
@@ -7289,6 +7507,18 @@ Write the complete compiled report:"""
if tool_call_delta.function.arguments:
tool_calls_data[tool_call_delta.index]['function']['arguments'] += tool_call_delta.function.arguments
# Emit LAST_TOKEN and STREAM_END events after streaming loop
if last_content_time:
self.stream_emitter.emit(StreamEvent(
type=StreamEventType.LAST_TOKEN,
timestamp=last_content_time
))
self.stream_emitter.emit(StreamEvent(
type=StreamEventType.STREAM_END,
timestamp=time_module.perf_counter(),
metadata={"response_length": len(response_text)}
))
# Handle any tool calls that were accumulated
if tool_calls_data:
# Add assistant message with tool calls to chat history
@@ -7521,6 +7751,10 @@ Write the complete compiled report:"""
error_msg = f"Tool execution denied: {decision.reason}"
logging.warning(error_msg)
return {"error": error_msg, "approval_denied": True}
# Bridge: mark tool as approved in global registry context so
# @require_approval decorator wrappers skip their own check.
from ..approval import get_approval_registry as _get_reg
_get_reg().mark_approved(function_name)
if decision.modified_args:
arguments = decision.modified_args
logging.info(f"Using modified arguments: {arguments}")
@@ -122,6 +122,8 @@ class AutonomyConfig:
track_changes: Optional[bool] = None # None = auto (True for full_auto)
sandbox: Optional[Any] = None # SandboxConfig (lazy import to avoid circular)
snapshot_dir: Optional[str] = None # Defaults to ~/.praisonai/snapshots
# Default tools for autonomy mode (auto-populated with code intelligence tools)
default_tools: Optional[List[Any]] = None # None = auto (uses get_autonomy_default_tools())
def __post_init__(self):
if self.level not in VALID_AUTONOMY_LEVELS:
@@ -134,10 +136,8 @@ class AutonomyConfig:
self.track_changes = (self.level == "full_auto")
# Default snapshot directory
if self.snapshot_dir is None:
import os
self.snapshot_dir = os.path.join(
os.path.expanduser("~"), ".praisonai", "snapshots"
)
from ..paths import get_snapshots_dir
self.snapshot_dir = str(get_snapshots_dir())
@property
def effective_track_changes(self) -> bool:
@@ -166,6 +166,7 @@ class AutonomyConfig:
track_changes=data.get("track_changes"),
sandbox=data.get("sandbox"),
snapshot_dir=data.get("snapshot_dir"),
default_tools=data.get("default_tools"),
)
@@ -162,7 +162,8 @@ class ContextAgent(Agent):
# Enhanced logging and output management
self.debug_mode = os.getenv('LOGLEVEL', '').lower() == 'debug'
self.output_dir = Path(".praisonai/prp") # Save in .praisonai/prp folder
from ..paths import get_project_prp_dir
self.output_dir = get_project_prp_dir() # Save in .praisonai/prp folder
self.setup_output_directories() # Create directories first
self.setup_logging() # Then setup logging
@@ -512,14 +512,16 @@ class AgentTeam:
if not mem_cfg:
mem_cfg = next((t.config.get('memory_config') for t in tasks if hasattr(t, 'config') and t.config), None)
if not mem_cfg:
from ..paths import get_project_data_dir
_pd = str(get_project_data_dir())
mem_cfg = {
"provider": "rag",
"use_embedding": True,
"storage": {
"type": "sqlite",
"path": "./.praisonai/memory.db"
"path": f"{_pd}/memory.db"
},
"rag_db_path": "./.praisonai/chroma_db"
"rag_db_path": f"{_pd}/chroma_db"
}
if _embedder:
if isinstance(_embedder, dict):
@@ -574,6 +574,10 @@ class OutputConfig:
# Shows: [timestamp] Calling LLM..., Executing tool..., Response: ...
status_trace: bool = False
# Editor output mode - beginner-friendly numbered steps
# Shows: Step 1: 📄 Creating file: path → ✓ Done
editor_output: bool = False
# Output file - save agent response to file automatically
# When set, saves the response to the specified file path
output_file: Optional[str] = None
@@ -596,6 +600,7 @@ class OutputConfig:
"simple_output": self.simple_output,
"show_parameters": self.show_parameters,
"status_trace": self.status_trace,
"editor_output": self.editor_output,
"output_file": self.output_file,
"template": self.template,
}
@@ -85,14 +85,32 @@ _config_cache: Optional[Dict[str, Any]] = None
_config_loaded: bool = False
def _default_project_plugins_dir():
"""Return project-local plugins directory via paths.py."""
from ..paths import get_project_data_dir
return get_project_data_dir() / "plugins"
def _default_global_plugins_dir():
"""Return user-global plugins directory via paths.py."""
from ..paths import get_plugins_dir
return get_plugins_dir()
def get_project_data_dir():
"""Re-export for local use in config search."""
from ..paths import get_project_data_dir as _gpd
return _gpd()
@dataclass
class PluginsConfig:
"""Configuration for the plugin system."""
enabled: Union[bool, List[str]] = False # True, False, or list of plugin names
auto_discover: bool = True
directories: List[str] = field(default_factory=lambda: [
"./.praisonai/plugins/",
"~/.praisonai/plugins/"
str(_default_project_plugins_dir()),
str(_default_global_plugins_dir())
])
def to_dict(self) -> Dict[str, Any]:
@@ -174,7 +192,7 @@ def _find_config_file() -> Optional[Path]:
# Project-local locations
cwd = Path.cwd()
local_paths = [
cwd / ".praisonai" / "config.toml",
get_project_data_dir() / "config.toml",
cwd / "praisonai.toml",
]
@@ -183,8 +201,8 @@ def _find_config_file() -> Optional[Path]:
return path
# User global location
home = Path.home()
global_path = home / ".praisonai" / "config.toml"
from ..paths import get_data_dir
global_path = get_data_dir() / "config.toml"
if global_path.exists():
return global_path
@@ -245,8 +263,8 @@ def _dict_to_plugins_config(data: Dict[str, Any]) -> PluginsConfig:
enabled=data.get("enabled", False),
auto_discover=data.get("auto_discover", True),
directories=data.get("directories", [
"./.praisonai/plugins/",
"~/.praisonai/plugins/"
str(_default_project_plugins_dir()),
str(_default_global_plugins_dir())
]),
)
@@ -133,6 +133,17 @@ OUTPUT_PRESETS: Dict[str, Dict[str, Any]] = {
"actions_trace": True,
"json_output": True, # Output as JSONL
},
# Editor preset - beginner-friendly numbered steps
# Shows: Step 1: 📄 Creating file: path → ✓ Done
# Best for non-programmers and interactive scripts
"editor": {
"verbose": False,
"markdown": False,
"stream": False,
"metrics": False,
"reasoning_steps": False,
"editor_output": True, # Enable editor display mode
},
# ==========================================================================
# ALIASES (for backward compatibility)
@@ -6,6 +6,9 @@ implementations that coordinate multi-agent deployments.
This module contains only protocols and lightweight utilities.
Heavy implementations live in the praisonai wrapper package.
Gap S2: WebSocketGateway is re-exported here for convenience but requires
the praisonai wrapper package to be installed.
"""
from .protocols import (
@@ -18,15 +21,59 @@ from .protocols import (
)
from .config import GatewayConfig, SessionConfig, ChannelRouteConfig, MultiChannelGatewayConfig
# Lazy loading cache
_lazy_cache = {}
def __getattr__(name: str):
"""Lazy load heavy gateway implementations from praisonai wrapper.
Gap S2: Re-export WebSocketGateway from praisonai wrapper for convenience.
This allows downstream packages to import from praisonaiagents.gateway
without needing to know about the praisonai wrapper.
"""
if name in _lazy_cache:
return _lazy_cache[name]
if name == "WebSocketGateway":
try:
from praisonai.gateway import WebSocketGateway
_lazy_cache[name] = WebSocketGateway
return WebSocketGateway
except ImportError:
raise ImportError(
"WebSocketGateway requires the praisonai wrapper package. "
"Install with: pip install praisonai"
)
if name == "GatewaySession":
try:
from praisonai.gateway import GatewaySession
_lazy_cache[name] = GatewaySession
return GatewaySession
except ImportError:
raise ImportError(
"GatewaySession requires the praisonai wrapper package. "
"Install with: pip install praisonai"
)
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
__all__ = [
# Protocols (always available)
"GatewayProtocol",
"GatewaySessionProtocol",
"GatewayClientProtocol",
"GatewayEvent",
"GatewayMessage",
"EventType",
# Config (always available)
"GatewayConfig",
"SessionConfig",
"ChannelRouteConfig",
"MultiChannelGatewayConfig",
# Implementations (lazy loaded from praisonai wrapper)
"WebSocketGateway",
"GatewaySession",
]
@@ -416,7 +416,8 @@ class Knowledge:
def config(self):
# Generate unique collection name for each instance (only if not provided in config)
default_collection = f"test_{int(time.time())}_{str(uuid.uuid4())[:8]}"
persist_dir = ".praisonai"
from ..paths import get_project_data_dir
persist_dir = str(get_project_data_dir())
# Create persistent client config
base_config = {
@@ -862,7 +863,8 @@ class Knowledge:
result = IndexResult()
# Initialize file tracker for incremental indexing
state_dir = os.path.join(os.path.dirname(path) if os.path.isfile(path) else path, ".praisonai")
from ..paths import DEFAULT_DIR_NAME
state_dir = os.path.join(os.path.dirname(path) if os.path.isfile(path) else path, DEFAULT_DIR_NAME)
os.makedirs(state_dir, exist_ok=True)
state_file = os.path.join(state_dir, ".index_state.json")
@@ -592,6 +592,18 @@ Respond with ONLY a valid JSON tool call in this format:
f"waiting {retry_delay:.1f}s before retry..."
)
# P5: Emit retry callback for CLI visibility
try:
from ..main import execute_sync_callback
execute_sync_callback('retry',
attempt=attempt + 1,
max_attempts=self._max_retries + 1,
error=error_str[:200],
retry_in_seconds=retry_delay
)
except ImportError:
pass
if self._rate_limiter is not None:
self._rate_limiter.wait_for_retry(retry_delay)
else:
@@ -1156,6 +1156,8 @@ class OpenAIClient:
reasoning_steps: bool = False,
verbose: bool = True,
max_iterations: int = 10,
stream_callback: Optional[Callable] = None,
emit_events: bool = False,
**kwargs
) -> Optional[ChatCompletion]:
"""
@@ -1179,6 +1181,8 @@ class OpenAIClient:
reasoning_steps: Whether to show reasoning
verbose: Whether to show verbose output
max_iterations: Maximum tool calling iterations
stream_callback: Optional callback for StreamEvent emission
emit_events: Whether to emit StreamEvents (requires stream_callback)
**kwargs: Additional API parameters
Returns:
@@ -1208,6 +1212,8 @@ class OpenAIClient:
console=console,
display_fn=display_fn,
reasoning_steps=reasoning_steps,
stream_callback=stream_callback,
emit_events=emit_events,
**kwargs
)
else:
@@ -1377,6 +1383,8 @@ class OpenAIClient:
reasoning_steps: bool = False,
verbose: bool = True,
max_iterations: int = 10,
stream_callback: Optional[Callable] = None,
emit_events: bool = False,
**kwargs
) -> Optional[ChatCompletion]:
"""
@@ -1394,6 +1402,8 @@ class OpenAIClient:
reasoning_steps: Whether to show reasoning
verbose: Whether to show verbose output
max_iterations: Maximum tool calling iterations
stream_callback: Optional callback for StreamEvent emission (can be sync or async)
emit_events: Whether to emit StreamEvents (requires stream_callback)
**kwargs: Additional API parameters
Returns:
@@ -1419,6 +1429,8 @@ class OpenAIClient:
console=console,
display_fn=display_fn,
reasoning_steps=reasoning_steps,
stream_callback=stream_callback,
emit_events=emit_events,
**kwargs
)
else:
@@ -103,6 +103,20 @@ __all__ = [
'display_self_reflection',
]
# P3/G2: Supported callback types documentation
# Standard callbacks: tool_call, interaction, error, llm_start, llm_end
# Autonomy callbacks (PraisonAI unique):
# - autonomy_iteration: {iteration, max_iterations, stage}
# - autonomy_stage_change: {from_stage, to_stage}
# - autonomy_doom_loop: {iteration, recovery_action}
# - autonomy_complete: {completion_reason, iterations, duration_seconds}
# - retry: {attempt, max_attempts, error, retry_in_seconds}
SUPPORTED_CALLBACK_TYPES = [
'tool_call', 'interaction', 'error', 'llm_start', 'llm_end',
'autonomy_iteration', 'autonomy_stage_change', 'autonomy_doom_loop', 'autonomy_complete',
'retry',
]
def register_display_callback(display_type: str, callback_fn, is_async: bool = False):
"""Register a synchronous or asynchronous callback function for a specific display type.
@@ -80,7 +80,8 @@ class DocsManager:
- Priority-based ordering
"""
DOCS_DIR_NAME = ".praisonai/docs"
from ..paths import DEFAULT_DIR_NAME as _DIR_NAME
DOCS_DIR_NAME = f"{_DIR_NAME}/docs"
SUPPORTED_EXTENSIONS = [".md", ".txt", ".rst"]
# Maximum characters per doc (context window management)
@@ -101,7 +102,11 @@ class DocsManager:
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.global_docs_path = Path(global_docs_path) if global_docs_path else Path.home() / ".praisonai" / "docs"
if global_docs_path:
self.global_docs_path = Path(global_docs_path)
else:
from ..paths import get_docs_dir
self.global_docs_path = get_docs_dir()
self.verbose = verbose
self._docs: Dict[str, Doc] = {}
@@ -140,7 +140,8 @@ class FileMemory:
if base_path:
self.base_path = Path(base_path)
else:
self.base_path = Path(".praisonai/memory")
from ..paths import get_project_data_dir
self.base_path = get_project_data_dir() / "memory"
self.user_path = self.base_path / user_id
self.episodic_path = self.user_path / "episodic"
@@ -103,7 +103,8 @@ class HooksManager:
```
"""
CONFIG_FILE = ".praisonai/hooks.json"
from ..paths import DEFAULT_DIR_NAME as _DIR_NAME
CONFIG_FILE = f"{_DIR_NAME}/hooks.json"
def __init__(
self,
@@ -384,7 +385,8 @@ class HooksManager:
timeout: Global timeout in seconds
enabled: Whether hooks are enabled
"""
config_dir = self.workspace_path / ".praisonai"
from ..paths import DEFAULT_DIR_NAME
config_dir = self.workspace_path / DEFAULT_DIR_NAME
config_dir.mkdir(parents=True, exist_ok=True)
config = {
@@ -113,7 +113,8 @@ class MCPConfigManager:
- Easy server management
"""
MCP_DIR_NAME = ".praisonai/mcp"
from ..paths import DEFAULT_DIR_NAME as _DIR_NAME
MCP_DIR_NAME = f"{_DIR_NAME}/mcp"
SUPPORTED_EXTENSIONS = [".json", ".mcpjson"]
def __init__(
@@ -131,7 +132,11 @@ class MCPConfigManager:
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.global_mcp_path = Path(global_mcp_path) if global_mcp_path else Path.home() / ".praisonai" / "mcp"
if global_mcp_path:
self.global_mcp_path = Path(global_mcp_path)
else:
from ..paths import get_mcp_dir
self.global_mcp_path = get_mcp_dir()
self.verbose = verbose
self._configs: Dict[str, MCPConfig] = {}
@@ -226,15 +226,17 @@ class Memory:
self.embedding_dimensions = self._get_embedding_dimensions(self.embedding_model)
self._log_verbose(f"Using embedding dimensions: {self.embedding_dimensions}")
# Create .praisonai directory if it doesn't exist
os.makedirs(".praisonai", exist_ok=True)
# Create project data directory if it doesn't exist
from ..paths import get_project_data_dir
_project_data = str(get_project_data_dir())
os.makedirs(_project_data, exist_ok=True)
# Short-term DB
self.short_db = self.cfg.get("short_db", ".praisonai/short_term.db")
self.short_db = self.cfg.get("short_db", os.path.join(_project_data, "short_term.db"))
self._init_stm()
# Long-term DB
self.long_db = self.cfg.get("long_db", ".praisonai/long_term.db")
self.long_db = self.cfg.get("long_db", os.path.join(_project_data, "long_term.db"))
self._init_ltm()
# Conditionally init Mem0, MongoDB, or local RAG
@@ -122,7 +122,8 @@ class RulesManager:
```
"""
RULES_DIR_NAME = ".praisonai/rules"
from ..paths import DEFAULT_DIR_NAME as _DIR_NAME
RULES_DIR_NAME = f"{_DIR_NAME}/rules"
SUPPORTED_EXTENSIONS = [".md", ".mdc", ".txt"]
# Root-level instruction files (like Claude, Codex, Gemini, Cursor, Windsurf)
@@ -165,7 +166,11 @@ class RulesManager:
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.global_rules_path = Path(global_rules_path) if global_rules_path else Path.home() / ".praisonai" / "rules"
if global_rules_path:
self.global_rules_path = Path(global_rules_path)
else:
from ..paths import get_rules_dir
self.global_rules_path = get_rules_dir()
self.verbose = verbose
self._rules: Dict[str, Rule] = {}
@@ -782,7 +782,8 @@ class WorkflowManager:
```
"""
WORKFLOWS_DIR = ".praisonai/workflows"
from ..paths import DEFAULT_DIR_NAME as _DIR_NAME
WORKFLOWS_DIR = f"{_DIR_NAME}/workflows"
SUPPORTED_EXTENSIONS = [".md", ".yaml", ".yml"]
def __init__(
@@ -1857,7 +1858,8 @@ class WorkflowManager:
def _get_checkpoints_dir(self) -> Path:
"""Get the checkpoints directory path."""
checkpoints_dir = self.workspace_path / ".praisonai" / "checkpoints"
from ..paths import DEFAULT_DIR_NAME
checkpoints_dir = self.workspace_path / DEFAULT_DIR_NAME / "checkpoints"
checkpoints_dir.mkdir(parents=True, exist_ok=True)
return checkpoints_dir
@@ -49,6 +49,15 @@ __all__ = [
"disable_trace_output",
"is_trace_output_enabled",
"get_trace_output",
# Editor output (for editor preset - beginner-friendly numbered steps)
"EditorOutput",
"enable_editor_output",
"disable_editor_output",
"is_editor_output_enabled",
"get_editor_output",
"TOOL_LABELS",
"BlockType",
"DisplayBlock",
]
@@ -112,4 +121,37 @@ def __getattr__(name: str):
from .trace import get_trace_output
return get_trace_output
# Editor output (for editor preset - beginner-friendly numbered steps)
if name == "EditorOutput":
from .editor import EditorOutput
return EditorOutput
if name == "enable_editor_output":
from .editor import enable_editor_output
return enable_editor_output
if name == "disable_editor_output":
from .editor import disable_editor_output
return disable_editor_output
if name == "is_editor_output_enabled":
from .editor import is_editor_output_enabled
return is_editor_output_enabled
if name == "get_editor_output":
from .editor import get_editor_output
return get_editor_output
if name == "TOOL_LABELS":
from .editor import TOOL_LABELS
return TOOL_LABELS
if name == "BlockType":
from .editor import BlockType
return BlockType
if name == "DisplayBlock":
from .editor import DisplayBlock
return DisplayBlock
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -80,6 +80,7 @@ class StatusOutput:
use_color: bool = True,
show_timestamps: bool = True, # NEW: control timestamp display
show_metrics: bool = False, # Enable metrics display for debug mode
show_tool_output: bool = True, # Show tool result output inline
):
self._file = file or sys.stderr # Use stderr to not interfere with agent output
self._format = format
@@ -87,6 +88,7 @@ class StatusOutput:
self._use_color = use_color
self._show_timestamps = show_timestamps
self._show_metrics = show_metrics
self._show_tool_output = show_tool_output
self._console = None
self._tool_start_times: Dict[str, float] = {}
self._lock = threading.Lock() # Per-sink lock for thread safety
@@ -120,12 +122,12 @@ class StatusOutput:
parts = []
for k, v in (redacted or {}).items():
if isinstance(v, str):
v_str = f'"{_truncate(v, 30)}"'
v_str = f'"{_truncate(v, 80)}"'
else:
v_str = _truncate(str(v), 30)
v_str = _truncate(str(v), 80)
parts.append(f"{k}={v_str}")
return _truncate(", ".join(parts), 80)
return _truncate(", ".join(parts), 200)
def agent_start(self, agent_name: str) -> None:
"""Record agent start."""
@@ -185,8 +187,8 @@ class StatusOutput:
"""Record LLM call end with optional metrics for debug mode."""
ts = time.time()
# Use latency_ms if provided, otherwise calculate from start time
if latency_ms is not None:
# Use latency_ms if provided and positive, otherwise calculate from start time
if latency_ms is not None and latency_ms > 0:
duration_ms = latency_ms
elif duration_ms is None and hasattr(self, '_llm_start_time'):
start_ts = self._llm_start_time
@@ -267,9 +269,13 @@ class StatusOutput:
args_str = self._format_args(self._pending_tool_args)
result_str = ""
if result_summary:
result_str = f" → {_truncate(result_summary, 50)}"
if self._show_tool_output:
if result_summary:
result_str = f" → {_truncate(result_summary, 50)}"
elif error_message:
result_str = f" → ✗ {_truncate(error_message, 50)}"
elif error_message:
# Always show errors even when output is hidden
result_str = f" → ✗ {_truncate(error_message, 50)}"
color = "cyan" if status == "ok" else "red"
@@ -285,9 +291,9 @@ class StatusOutput:
separator = "─" * 50
self._emit_text(separator, ts, "dim", show_timestamp=False)
self._emit_text("Final Output:", ts, "bold", show_timestamp=False)
# Print actual content without truncation for final output
# Print actual content to stdout (not stderr) so users see it
with self._lock: # Thread-safe output
print(content, file=self._file)
print(content)
def _emit_text(self, message: str, ts: float, style: str = None, show_timestamp: bool = True) -> None:
"""Emit a text line. Thread-safe for multi-agent execution."""
@@ -320,6 +326,7 @@ def enable_status_output(
use_color: bool = True,
show_timestamps: bool = True,
show_metrics: bool = False, # Enable metrics for debug mode
show_tool_output: bool = True, # Show tool result output inline
) -> StatusOutput:
"""
Enable actions output mode globally.
@@ -334,6 +341,7 @@ def enable_status_output(
use_color: Whether to use colored output (default: True)
show_timestamps: Whether to show timestamps (default: True)
show_metrics: Whether to show token/cost metrics (default: False)
show_tool_output: Whether to show tool result output inline (default: True)
Returns:
StatusOutput instance for programmatic access
@@ -350,6 +358,7 @@ def enable_status_output(
use_color=use_color,
show_timestamps=show_timestamps,
show_metrics=show_metrics,
show_tool_output=show_tool_output,
)
_status_output_enabled = True
@@ -309,6 +309,78 @@ def get_project_data_dir(project_path: Optional[Union[str, Path]] = None) -> Pat
return base / DEFAULT_DIR_NAME
def get_project_sessions_dir(project_path: Optional[Union[str, Path]] = None) -> Path:
"""
Get project-level sessions directory.
Args:
project_path: Project root (defaults to cwd)
Returns:
Path to .praisonai/sessions/ in project
"""
return get_project_data_dir(project_path) / "sessions"
def get_project_knowledge_dir(project_path: Optional[Union[str, Path]] = None) -> Path:
"""
Get project-level knowledge directory.
Args:
project_path: Project root (defaults to cwd)
Returns:
Path to .praisonai/knowledge/ in project
"""
return get_project_data_dir(project_path) / "knowledge"
def get_project_summaries_dir(project_path: Optional[Union[str, Path]] = None) -> Path:
"""
Get project-level summaries directory for RAG.
Args:
project_path: Project root (defaults to cwd)
Returns:
Path to .praisonai/summaries/ in project
"""
return get_project_data_dir(project_path) / "summaries"
def get_project_prp_dir(project_path: Optional[Union[str, Path]] = None) -> Path:
"""
Get project-level PRP output directory.
Args:
project_path: Project root (defaults to cwd)
Returns:
Path to .praisonai/prp/ in project
"""
return get_project_data_dir(project_path) / "prp"
def get_schedules_dir() -> Path:
"""
Get schedules directory.
Returns:
Path to ~/.praisonai/schedules/
"""
return get_data_dir() / "schedules"
def get_storage_path() -> Path:
"""
Get default SQLite storage database path.
Returns:
Path to ~/.praisonai/storage.db
"""
return get_data_dir() / "storage.db"
def ensure_dir(path: Union[str, Path]) -> Path:
"""
Ensure a directory exists, creating it if necessary.
@@ -352,6 +424,8 @@ def get_all_paths() -> Dict[str, Path]:
"rules": get_rules_dir(),
"permissions": get_permissions_dir(),
"storage": get_storage_dir(),
"storage_db": get_storage_path(),
"schedules": get_schedules_dir(),
"checkpoints": get_checkpoints_dir(),
"snapshots": get_snapshots_dir(),
"learn": get_learn_dir(),
@@ -15,8 +15,9 @@ from typing import Optional, List, Any, Dict
logger = logging.getLogger(__name__)
# Default rules file path (follows ~/.praisonai/ convention)
DEFAULT_RULES_DIR = os.path.join(os.path.expanduser("~"), ".praisonai", "rules")
# Default rules file path (uses centralized paths - DRY)
from ..paths import get_rules_dir as _get_rules_dir
DEFAULT_RULES_DIR = str(_get_rules_dir())
DEFAULT_RULES_FILE = os.path.join(DEFAULT_RULES_DIR, "default.rules")
@@ -92,7 +92,7 @@ class RetrievalConfig:
vector_store_provider: str = "chroma"
vector_store_config: Dict[str, Any] = field(default_factory=dict)
collection_name: Optional[str] = None
persist_path: str = ".praisonai"
persist_path: str = None # Resolved to str(get_project_data_dir()) in __post_init__
# Embedding configuration
embedder_provider: Optional[str] = None
@@ -112,6 +112,11 @@ class RetrievalConfig:
{context}
</retrieved_context>"""
system_separation: bool = True # Use system prompt separation for safety
def __post_init__(self):
if self.persist_path is None:
from ..paths import get_project_data_dir
self.persist_path = str(get_project_data_dir())
def get_token_budget(self, model_name: Optional[str] = None):
"""
@@ -233,7 +238,7 @@ class RetrievalConfig:
vector_store_provider=data.get("vector_store_provider", "chroma"),
vector_store_config=data.get("vector_store_config", {}),
collection_name=data.get("collection_name"),
persist_path=data.get("persist_path", ".praisonai"),
persist_path=data.get("persist_path"),
embedder_provider=data.get("embedder_provider"),
embedder_model=data.get("embedder_model"),
auto_keywords=auto_keywords,
@@ -99,7 +99,7 @@ class HierarchicalSummarizer:
def __init__(
self,
llm=None,
persist_path: str = ".praisonai/summaries",
persist_path: str = None,
verbose: bool = False,
):
"""
@@ -111,6 +111,9 @@ class HierarchicalSummarizer:
verbose: Enable verbose logging
"""
self._llm = llm
if persist_path is None:
from ..paths import get_project_summaries_dir
persist_path = str(get_project_summaries_dir())
self._persist_path = persist_path
self._verbose = verbose
self._hierarchy: Dict[str, SummaryNode] = {}
@@ -15,7 +15,8 @@ from .models import ScheduleJob
logger = logging.getLogger(__name__)
_DEFAULT_DIR = os.path.join(os.path.expanduser("~"), ".praisonai", "schedules")
from ..paths import get_schedules_dir as _get_schedules_dir
_DEFAULT_DIR = str(_get_schedules_dir())
_JOBS_FILE = "jobs.json"
+44 -15
View File
@@ -83,7 +83,7 @@ class Session:
default_memory_config = {
"provider": "rag",
"use_embedding": False, # Disable embeddings to avoid OpenAI API key requirement
"rag_db_path": f".praisonai/sessions/{self.session_id}/chroma_db"
"rag_db_path": str(self._get_session_dir() / "chroma_db")
}
if memory_config:
default_memory_config.update(memory_config)
@@ -94,7 +94,7 @@ class Session:
self.knowledge_config = default_knowledge_config
# Create session directory
os.makedirs(f".praisonai/sessions/{self.session_id}", exist_ok=True)
self._get_session_dir().mkdir(parents=True, exist_ok=True)
# Initialize components lazily
self._memory = None
@@ -110,6 +110,12 @@ class Session:
self._agents_instance = None
self._agents = {} # Track agents and their chat histories
def _get_session_dir(self):
"""Return session-specific directory using paths.py."""
from pathlib import Path
from .paths import get_project_sessions_dir
return get_project_sessions_dir() / self.session_id
@property
def memory(self) -> Memory:
"""Lazy-loaded memory instance"""
@@ -323,7 +329,11 @@ class Session:
def _save_agent_chat_histories(self) -> None:
"""
Save all agent chat histories to memory.
Save all agent chat histories.
G-2 FIX: Routes to SessionStore instead of Memory.store_short_term() to
maintain clean separation between conversation history (SessionStore)
and semantic memory (Memory). Falls back to Memory for backward compat.
"""
if self.is_remote:
return
@@ -340,18 +350,37 @@ class Session:
chat_history = agent_data.get("chat_history")
if chat_history is not None:
# Save to memory
history_text = f"Agent chat history for {agent_key}"
self.memory.store_short_term(
text=history_text,
metadata={
"type": "agent_chat_history",
"session_id": self.session_id,
"user_id": self.user_id,
"agent_key": agent_key,
"chat_history": chat_history
}
)
# G-2 FIX: Try SessionStore first for clean separation
session_store = None
try:
from .session import get_default_session_store
session_store = get_default_session_store()
except ImportError:
pass
if session_store is not None:
# Use SessionStore for conversation history
session_id = f"{self.session_id}_{agent_key}"
for msg in chat_history:
if isinstance(msg, dict):
session_store.add_message(
session_id,
role=msg.get("role", "user"),
content=msg.get("content", ""),
)
else:
# Fallback to Memory.store_short_term() for backward compatibility
history_text = f"Agent chat history for {agent_key}"
self.memory.store_short_term(
text=history_text,
metadata={
"type": "agent_chat_history",
"session_id": self.session_id,
"user_id": self.user_id,
"agent_key": agent_key,
"chat_history": chat_history
}
)
def get_state(self, key: str, default: Any = None) -> Any:
"""Get a specific state value"""
@@ -86,7 +86,7 @@ class Session:
default_memory_config = {
"provider": "rag",
"use_embedding": False,
"rag_db_path": f".praisonai/sessions/{self.session_id}/chroma_db"
"rag_db_path": str(self._get_session_dir() / "chroma_db")
}
if memory_config:
default_memory_config.update(memory_config)
@@ -95,7 +95,7 @@ class Session:
default_knowledge_config = knowledge_config or {}
self.knowledge_config = default_knowledge_config
os.makedirs(f".praisonai/sessions/{self.session_id}", exist_ok=True)
self._get_session_dir().mkdir(parents=True, exist_ok=True)
self._memory = None
self._knowledge = None
@@ -109,6 +109,12 @@ class Session:
self._agents_instance = None
self._agents = {}
def _get_session_dir(self):
"""Get session-specific directory using centralized paths."""
from ..paths import get_project_sessions_dir
from pathlib import Path
return get_project_sessions_dir() / self.session_id
@property
def memory(self) -> "Memory":
"""Lazy-loaded memory instance"""
@@ -305,7 +311,12 @@ class Session:
}
def _save_agent_chat_histories(self) -> None:
"""Save all agent chat histories to memory."""
"""Save all agent chat histories.
G-2 FIX: Routes to SessionStore instead of Memory.store_short_term() to
maintain clean separation between conversation history (SessionStore)
and semantic memory (Memory). Falls back to Memory for backward compat.
"""
if self.is_remote:
return
@@ -320,17 +331,37 @@ class Session:
chat_history = agent_data.get("chat_history")
if chat_history is not None:
history_text = f"Agent chat history for {agent_key}"
self.memory.store_short_term(
text=history_text,
metadata={
"type": "agent_chat_history",
"session_id": self.session_id,
"user_id": self.user_id,
"agent_key": agent_key,
"chat_history": chat_history
}
)
# G-2 FIX: Try SessionStore first for clean separation
session_store = None
try:
from . import get_default_session_store
session_store = get_default_session_store()
except ImportError:
pass
if session_store is not None:
# Use SessionStore for conversation history
session_id = f"{self.session_id}_{agent_key}"
for msg in chat_history:
if isinstance(msg, dict):
session_store.add_message(
session_id,
role=msg.get("role", "user"),
content=msg.get("content", ""),
)
else:
# Fallback to Memory.store_short_term() for backward compatibility
history_text = f"Agent chat history for {agent_key}"
self.memory.store_short_term(
text=history_text,
metadata={
"type": "agent_chat_history",
"session_id": self.session_id,
"user_id": self.user_id,
"agent_key": agent_key,
"chat_history": chat_history
}
)
def get_state(self, key: str, default: Any = None) -> Any:
"""Get a specific state value"""
@@ -67,6 +67,9 @@ class SessionData:
agent_name: Optional[str] = None
user_id: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
# Gap S3: Gateway integration - link session to gateway session
gateway_session_id: Optional[str] = None
agent_id: Optional[str] = None # Gateway agent ID (different from agent_name)
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
@@ -78,6 +81,8 @@ class SessionData:
"agent_name": self.agent_name,
"user_id": self.user_id,
"metadata": self.metadata,
"gateway_session_id": self.gateway_session_id,
"agent_id": self.agent_id,
}
@classmethod
@@ -95,6 +100,8 @@ class SessionData:
agent_name=data.get("agent_name"),
user_id=data.get("user_id"),
metadata=data.get("metadata", {}),
gateway_session_id=data.get("gateway_session_id"),
agent_id=data.get("agent_id"),
)
def get_chat_history(self, max_messages: Optional[int] = None) -> List[Dict[str, str]]:
@@ -489,6 +496,177 @@ class DefaultSessionStore:
sessions.sort(key=lambda x: x.get("updated_at", ""), reverse=True)
return sessions[:limit]
# ── Agent-Level Queries (Gap S4) ──────────────────────────────────
def list_sessions_by_agent(self, agent_name: str, limit: int = 50) -> List[str]:
"""List session IDs for a specific agent.
Gap S4: Enables querying sessions by agent instead of just session ID.
Args:
agent_name: The agent name to filter by
limit: Maximum number of session IDs to return
Returns:
List of session IDs belonging to the specified agent
"""
session_ids = []
try:
for filename in os.listdir(self.session_dir):
if filename.endswith(".json"):
filepath = os.path.join(self.session_dir, filename)
try:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
if data.get("agent_name") == agent_name:
session_ids.append(data.get("session_id", filename[:-5]))
except (json.JSONDecodeError, IOError):
continue
except (IOError, OSError):
pass
return session_ids[:limit]
def get_sessions_by_agent(
self,
agent_name: str,
limit: int = 10,
) -> List[SessionData]:
"""Get full session data for a specific agent.
Gap S4: Enables retrieving all sessions for an agent.
Args:
agent_name: The agent name to filter by
limit: Maximum number of sessions to return
Returns:
List of SessionData objects for the specified agent
"""
session_ids = self.list_sessions_by_agent(agent_name, limit)
return [self._load_session(sid) for sid in session_ids]
def get_agent_chat_history(
self,
agent_name: str,
max_messages: Optional[int] = None,
max_sessions: int = 5,
) -> List[Dict[str, Any]]:
"""Get combined chat history across all sessions for an agent.
Gap S4: Enables retrieving conversation history by agent.
Args:
agent_name: The agent name to filter by
max_messages: Maximum messages per session
max_sessions: Maximum number of sessions to include
Returns:
List of messages with session context
"""
sessions = self.get_sessions_by_agent(agent_name, max_sessions)
messages = []
for session in sessions:
history = session.get_chat_history(max_messages)
for msg in history:
messages.append({
**msg,
"session_id": session.session_id,
"agent_name": agent_name,
})
return messages
# ── Gateway Integration (Gap S3) ──────────────────────────────────
def set_gateway_info(
self,
session_id: str,
gateway_session_id: Optional[str] = None,
agent_id: Optional[str] = None,
) -> bool:
"""Link a session to a gateway session.
Gap S3: Enables memory to be scoped to gateway agents/sessions.
Args:
session_id: The local session ID
gateway_session_id: The gateway session ID to link to
agent_id: The gateway agent ID
Returns:
True if saved successfully
"""
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
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.
Gap S3: Enables retrieving memory by gateway session ID.
Args:
gateway_session_id: The gateway session ID to look up
Returns:
SessionData if found, None otherwise
"""
try:
for filename in os.listdir(self.session_dir):
if filename.endswith(".json"):
filepath = os.path.join(self.session_dir, filename)
try:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
if data.get("gateway_session_id") == gateway_session_id:
return SessionData.from_dict(data)
except (json.JSONDecodeError, IOError):
continue
except (IOError, OSError):
pass
return None
def list_sessions_by_gateway_agent(self, agent_id: str, limit: int = 50) -> List[str]:
"""List session IDs linked to a gateway agent.
Gap S3: Enables querying sessions by gateway agent ID.
Args:
agent_id: The gateway agent ID to filter by
limit: Maximum number of session IDs to return
Returns:
List of session IDs linked to the gateway agent
"""
session_ids = []
try:
for filename in os.listdir(self.session_dir):
if filename.endswith(".json"):
filepath = os.path.join(self.session_dir, filename)
try:
with open(filepath, "r", encoding="utf-8") as f:
data = json.load(f)
if data.get("agent_id") == agent_id:
session_ids.append(data.get("session_id", filename[:-5]))
except (json.JSONDecodeError, IOError):
continue
except (IOError, OSError):
pass
return session_ids[:limit]
def session_exists(self, session_id: str) -> bool:
"""Check if a session exists."""
filepath = self._get_session_path(session_id)
@@ -19,7 +19,7 @@ import logging
from pathlib import Path
from typing import Any, Dict, List, Optional
from ..paths import get_storage_dir
from ..paths import get_storage_dir, get_storage_path
logger = logging.getLogger(__name__)
@@ -175,7 +175,7 @@ class SQLiteBackend:
def __init__(
self,
db_path: str = "~/.praisonai/storage.db",
db_path: str = None,
table_name: str = "praison_storage",
auto_create: bool = True,
):
@@ -187,7 +187,7 @@ class SQLiteBackend:
table_name: Name of the storage table
auto_create: Create table if it doesn't exist
"""
self.db_path = os.path.expanduser(db_path)
self.db_path = os.path.expanduser(db_path) if db_path else str(get_storage_path())
self.table_name = table_name
self._local = threading.local()
@@ -155,12 +155,44 @@ TOOL_MAPPINGS = {
'schedule_list': ('.schedule_tools', None),
'schedule_remove': ('.schedule_tools', None),
'schedule_tools': ('.schedule_tools', None),
# AST-Grep Tools (structural code search/rewrite)
'ast_grep_search': ('.ast_grep_tool', None),
'ast_grep_rewrite': ('.ast_grep_tool', None),
'ast_grep_scan': ('.ast_grep_tool', None),
'is_ast_grep_available': ('.ast_grep_tool', None),
'get_ast_grep_tools': ('.ast_grep_tool', None),
'ast_grep_tools': ('.ast_grep_tool', None),
}
_instances = {} # Cache for class instances
# Profile exports (lazy loaded)
_PROFILE_EXPORTS = frozenset({
'ToolProfile', 'AUTONOMY_PROFILE', 'BUILTIN_PROFILES',
'register_profile', 'get_profile', 'resolve_profiles', 'list_profiles'
})
def __getattr__(name: str) -> Any:
"""Smart lazy loading of tools with class method support."""
"""Smart lazy loading of tools and profiles."""
# Handle profile imports first
if name in _PROFILE_EXPORTS:
from .profiles import (
ToolProfile, AUTONOMY_PROFILE, BUILTIN_PROFILES,
register_profile, get_profile, resolve_profiles, list_profiles
)
_profile_map = {
'ToolProfile': ToolProfile,
'AUTONOMY_PROFILE': AUTONOMY_PROFILE,
'BUILTIN_PROFILES': BUILTIN_PROFILES,
'register_profile': register_profile,
'get_profile': get_profile,
'resolve_profiles': resolve_profiles,
'list_profiles': list_profiles,
}
return _profile_map[name]
# Handle tool mappings
if name not in TOOL_MAPPINGS:
raise AttributeError(f"module '{__package__}' has no attribute '{name}'")
@@ -188,10 +220,11 @@ def __getattr__(name: str) -> Any:
'search_web', 'web_search', 'get_available_providers',
'web_crawl', 'crawl_web', 'get_available_crawl_providers',
'run_skill_script', 'read_skill_file', 'list_skill_scripts', 'create_skill_tools',
'schedule_add', 'schedule_list', 'schedule_remove'
'schedule_add', 'schedule_list', 'schedule_remove',
'ast_grep_search', 'ast_grep_rewrite', 'ast_grep_scan', 'is_ast_grep_available', 'get_ast_grep_tools'
]:
return getattr(module, name)
if name in ['file_tools', 'spider_tools', 'python_tools', 'shell_tools', 'cot_tools', 'tavily_tools', 'youdotcom_tools', 'exa_tools', 'crawl4ai_tools', 'skill_tools', 'schedule_tools']:
if name in ['file_tools', 'spider_tools', 'python_tools', 'shell_tools', 'cot_tools', 'tavily_tools', 'youdotcom_tools', 'exa_tools', 'crawl4ai_tools', 'skill_tools', 'schedule_tools', 'ast_grep_tools']:
return module # Returns the callable module
return getattr(module, name)
else:
@@ -214,4 +247,7 @@ __all__ = list(TOOL_MAPPINGS.keys()) + [
# Validation and retry protocols
'ValidationResult', 'ToolValidatorProtocol', 'AsyncToolValidatorProtocol', 'PassthroughValidator',
'RetryPolicy', 'FallbackChain', 'ToolExecutionConfig',
# Tool profiles (DRY tool sets for autonomy/interactive modes)
'ToolProfile', 'AUTONOMY_PROFILE', 'BUILTIN_PROFILES',
'register_profile', 'get_profile', 'resolve_profiles', 'list_profiles',
]
@@ -0,0 +1,338 @@
"""
AST-Grep Tool for PraisonAI Agents.
Provides AST-based structural code search and rewrite capabilities
using the ast-grep (sg) CLI tool. Gracefully handles cases where
ast-grep is not installed.
Features:
- Structural code search using AST patterns
- Structural code rewrite (dry-run by default)
- YAML rule scanning
- Graceful fallback when not installed
- Zero performance impact when not used (lazy loading)
Installation:
pip install ast-grep-cli
Usage:
from praisonaiagents import Agent
from praisonaiagents.tools import ast_grep_search
agent = Agent(
name="coder",
tools=[ast_grep_search],
)
"""
import shutil
import subprocess
import logging
from typing import Optional, List
logger = logging.getLogger(__name__)
# Availability cache for performance (checked once per process)
_availability_cache: Optional[bool] = None
def is_ast_grep_available() -> bool:
"""Check if ast-grep (sg) CLI is available.
Returns:
True if sg binary is found in PATH, False otherwise.
Note:
Result is cached for performance. First call checks PATH,
subsequent calls return cached result.
"""
global _availability_cache
if _availability_cache is not None:
return _availability_cache
# Check if sg binary is in PATH
sg_path = shutil.which('sg')
_availability_cache = sg_path is not None
if _availability_cache:
logger.debug(f"ast-grep found at: {sg_path}")
else:
logger.debug("ast-grep (sg) not found in PATH")
return _availability_cache
def _get_not_installed_message() -> str:
"""Get helpful error message when ast-grep is not installed."""
return (
"ast-grep is not installed or not available in PATH.\n\n"
"To install ast-grep, run:\n"
" pip install ast-grep-cli\n\n"
"Or install via npm:\n"
" npm install -g @ast-grep/cli\n\n"
"For more information, visit: https://ast-grep.github.io/"
)
def ast_grep_search(
pattern: str,
lang: str,
path: str = ".",
json_output: bool = True,
) -> str:
"""Search code using AST patterns.
Performs structural code search using ast-grep's pattern matching.
Unlike regex, this understands code structure and won't match
patterns in comments or strings.
Args:
pattern: AST pattern to search for. Use $VAR for single nodes,
$$$ for multiple nodes. Example: "def $FN($$$)" matches
all Python function definitions.
lang: Programming language (python, javascript, typescript,
rust, go, java, c, cpp, etc.)
path: Directory or file to search in. Defaults to current directory.
json_output: Whether to return JSON format. Defaults to True.
Returns:
Search results as JSON string (if json_output=True) or plain text.
Returns error message if ast-grep is not installed.
Example:
# Find all function definitions
result = ast_grep_search("def $FN($$$)", lang="python", path="./src")
# Find all console.log calls
result = ast_grep_search("console.log($$$)", lang="javascript")
# Find all async functions
result = ast_grep_search("async def $FN($$$)", lang="python")
"""
if not is_ast_grep_available():
return _get_not_installed_message()
if not pattern:
return "Error: Pattern cannot be empty"
try:
cmd = ['sg', '--pattern', pattern, '--lang', lang]
if json_output:
cmd.append('--json')
cmd.append(path)
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=60,
)
if result.returncode != 0 and result.stderr:
return f"Error: {result.stderr}"
output = result.stdout.strip()
if not output:
return "No matches found"
return output
except subprocess.TimeoutExpired:
return "Error: Search timed out after 60 seconds"
except subprocess.SubprocessError as e:
return f"Error executing ast-grep: {e}"
except Exception as e:
logger.exception("Unexpected error in ast_grep_search")
return f"Error: {e}"
def ast_grep_rewrite(
pattern: str,
replacement: str,
lang: str,
path: str = ".",
dry_run: bool = True,
) -> str:
"""Rewrite code using AST patterns.
Performs structural code transformation using ast-grep's pattern
matching and rewriting. By default, runs in dry-run mode to show
what would be changed without modifying files.
Args:
pattern: AST pattern to match. Use $VAR for single nodes,
$$$ for multiple nodes.
replacement: Replacement pattern. Can reference captured
variables from the pattern.
lang: Programming language (python, javascript, typescript,
rust, go, java, c, cpp, etc.)
path: Directory or file to rewrite. Defaults to current directory.
dry_run: If True (default), show changes without modifying files.
Set to False to actually modify files.
Returns:
Rewrite preview/results as string.
Returns error message if ast-grep is not installed.
Example:
# Preview renaming a function (dry run)
result = ast_grep_rewrite(
"def old_name($$$)",
"def new_name($$$)",
lang="python",
path="./src"
)
# Actually apply the rewrite
result = ast_grep_rewrite(
"console.log($MSG)",
"logger.info($MSG)",
lang="javascript",
dry_run=False
)
"""
if not is_ast_grep_available():
return _get_not_installed_message()
if not pattern:
return "Error: Pattern cannot be empty"
if not replacement:
return "Error: Replacement cannot be empty"
try:
cmd = [
'sg', '--pattern', pattern,
'--rewrite', replacement,
'--lang', lang,
]
if not dry_run:
cmd.append('--update-all')
cmd.append(path)
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120,
)
if result.returncode != 0 and result.stderr:
return f"Error: {result.stderr}"
output = result.stdout.strip()
if not output:
if dry_run:
return "No matches found for rewrite"
else:
return "No changes made"
prefix = "[DRY RUN] " if dry_run else ""
return f"{prefix}{output}"
except subprocess.TimeoutExpired:
return "Error: Rewrite timed out after 120 seconds"
except subprocess.SubprocessError as e:
return f"Error executing ast-grep: {e}"
except Exception as e:
logger.exception("Unexpected error in ast_grep_rewrite")
return f"Error: {e}"
def ast_grep_scan(
path: str = ".",
rule_file: Optional[str] = None,
) -> str:
"""Scan code using YAML lint rules.
Runs ast-grep's rule-based scanning to find code patterns
defined in YAML rule files. Useful for enforcing coding
standards and finding anti-patterns.
Args:
path: Directory or file to scan. Defaults to current directory.
rule_file: Optional path to a specific YAML rule file.
If not provided, looks for sgconfig.yml in the path.
Returns:
Scan results as string.
Returns error message if ast-grep is not installed.
Example:
# Scan with default rules
result = ast_grep_scan(path="./src")
# Scan with specific rule file
result = ast_grep_scan(
path="./src",
rule_file="./rules/security.yml"
)
"""
if not is_ast_grep_available():
return _get_not_installed_message()
try:
cmd = ['sg', 'scan']
if rule_file:
cmd.extend(['--rule', rule_file])
cmd.append(path)
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=120,
)
# scan returns non-zero if issues found, which is expected
output = result.stdout.strip()
if result.stderr and not output:
return f"Error: {result.stderr}"
if not output:
return "No issues found"
return output
except subprocess.TimeoutExpired:
return "Error: Scan timed out after 120 seconds"
except subprocess.SubprocessError as e:
return f"Error executing ast-grep: {e}"
except Exception as e:
logger.exception("Unexpected error in ast_grep_scan")
return f"Error: {e}"
# Convenience function to get all ast-grep tools
def get_ast_grep_tools() -> List:
"""Get all ast-grep tools as a list.
Returns:
List of ast-grep tool functions.
Example:
from praisonaiagents import Agent
from praisonaiagents.tools.ast_grep_tool import get_ast_grep_tools
agent = Agent(
name="coder",
tools=get_ast_grep_tools(),
)
"""
return [ast_grep_search, ast_grep_rewrite, ast_grep_scan]
__all__ = [
'is_ast_grep_available',
'ast_grep_search',
'ast_grep_rewrite',
'ast_grep_scan',
'get_ast_grep_tools',
]
@@ -93,6 +93,12 @@ class FileTools:
safe_path = FileTools._validate_path(filepath)
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(safe_path), exist_ok=True)
# Auto-coerce non-string content (LLMs sometimes pass dicts/lists)
if not isinstance(content, str):
if isinstance(content, (dict, list)):
content = json.dumps(content, indent=2, ensure_ascii=False)
else:
content = str(content)
with open(safe_path, 'w', encoding=encoding) as f:
f.write(content)
return True
@@ -0,0 +1,185 @@
"""
ToolProfile system for composable, extensible tool sets.
Provides named profiles of tools that can be combined and extended.
This eliminates DRY violations across interactive/autonomy modes.
Usage:
from praisonaiagents.tools.profiles import resolve_profiles, AUTONOMY_PROFILE
# Get tools for autonomy mode
tools = resolve_profiles("autonomy")
# Combine multiple profiles
tools = resolve_profiles("file_ops", "web", "shell")
# Register custom profile (wrapper/plugins)
from praisonaiagents.tools.profiles import register_profile, ToolProfile
register_profile(ToolProfile(name="custom", tools=["my_tool"]))
"""
from dataclasses import dataclass, field
from typing import Dict, List
@dataclass
class ToolProfile:
"""Named, extensible collection of tool names.
Attributes:
name: Profile identifier (e.g., "file_ops", "shell", "autonomy")
tools: List of tool name strings (resolved via TOOL_MAPPINGS)
description: Human-readable description of the profile
Example:
profile = ToolProfile(
name="file_ops",
tools=["read_file", "write_file", "list_files"],
description="File system operations"
)
"""
name: str
tools: List[str] = field(default_factory=list)
description: str = ""
# Built-in profiles (SDK ships with these)
BUILTIN_PROFILES: Dict[str, ToolProfile] = {
"code_intelligence": ToolProfile(
name="code_intelligence",
tools=["ast_grep_search", "ast_grep_rewrite", "ast_grep_scan"],
description="Structural code search and rewrite using ast-grep",
),
"file_ops": ToolProfile(
name="file_ops",
tools=[
"read_file", "write_file", "list_files",
"copy_file", "move_file", "delete_file", "get_file_info"
],
description="File system operations",
),
"shell": ToolProfile(
name="shell",
tools=["execute_command", "list_processes", "get_system_info"],
description="Shell and system tools",
),
"web": ToolProfile(
name="web",
tools=["internet_search", "search_web", "web_crawl"],
description="Web search and crawling",
),
"code_exec": ToolProfile(
name="code_exec",
tools=["execute_code", "analyze_code", "format_code", "lint_code"],
description="Python code execution and analysis",
),
"schedule": ToolProfile(
name="schedule",
tools=["schedule_add", "schedule_list", "schedule_remove"],
description="Agent-centric scheduling",
),
}
def _build_autonomy_tools() -> List[str]:
"""Build the autonomy profile tools list (deduped)."""
tools = []
for profile_name in ["code_intelligence", "file_ops", "shell", "web"]:
profile = BUILTIN_PROFILES[profile_name]
for tool in profile.tools:
if tool not in tools:
tools.append(tool)
return tools
# Composite autonomy profile - combines multiple profiles
AUTONOMY_PROFILE = ToolProfile(
name="autonomy",
tools=_build_autonomy_tools(),
description="Default tools for autonomous agents (file_ops + shell + web + code_intelligence)",
)
# Extension registry for custom profiles (wrapper/plugins use this)
_custom_profiles: Dict[str, ToolProfile] = {}
def register_profile(profile: ToolProfile) -> None:
"""Register a custom tool profile.
Custom profiles override built-in profiles with the same name.
This allows the wrapper to extend or replace SDK profiles.
Args:
profile: ToolProfile to register
Example:
# Wrapper registers ACP/LSP tools
register_profile(ToolProfile(
name="acp",
tools=["acp_create_file", "acp_edit_file"],
))
"""
_custom_profiles[profile.name] = profile
def get_profile(name: str) -> ToolProfile:
"""Get a tool profile by name.
Custom profiles take precedence over built-in profiles.
Args:
name: Profile name
Returns:
ToolProfile instance
Raises:
KeyError: If profile not found
"""
if name in _custom_profiles:
return _custom_profiles[name]
if name in BUILTIN_PROFILES:
return BUILTIN_PROFILES[name]
if name == "autonomy":
return AUTONOMY_PROFILE
raise KeyError(f"Unknown tool profile: {name}")
def resolve_profiles(*names: str) -> List[str]:
"""Resolve multiple profile names to a flat tool list.
Tools are deduplicated - each tool appears only once.
Args:
*names: Profile names to resolve
Returns:
List of tool name strings (deduped)
Raises:
KeyError: If any profile not found
Example:
# Get tools for autonomy + custom ACP
tools = resolve_profiles("autonomy", "acp")
"""
tools: List[str] = []
for name in names:
profile = get_profile(name)
for tool in profile.tools:
if tool not in tools:
tools.append(tool)
return tools
def list_profiles() -> List[str]:
"""List all available profile names.
Returns:
List of profile names (built-in + custom + autonomy)
"""
names = set(BUILTIN_PROFILES.keys())
names.update(_custom_profiles.keys())
names.add("autonomy")
return sorted(names)
@@ -2955,7 +2955,8 @@ class WorkflowManager:
```
"""
WORKFLOWS_DIR = ".praisonai/workflows"
from ..paths import DEFAULT_DIR_NAME as _DIR_NAME
WORKFLOWS_DIR = f"{_DIR_NAME}/workflows"
SUPPORTED_EXTENSIONS = [".md", ".yaml", ".yml"]
def __init__(
@@ -4246,7 +4247,8 @@ class WorkflowManager:
def _get_checkpoints_dir(self) -> Path:
"""Get the checkpoints directory path."""
checkpoints_dir = self.workspace_path / ".praisonai" / "checkpoints"
from ..paths import DEFAULT_DIR_NAME
checkpoints_dir = self.workspace_path / DEFAULT_DIR_NAME / "checkpoints"
checkpoints_dir.mkdir(parents=True, exist_ok=True)
return checkpoints_dir
+6 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "1.5.15"
version = "1.5.32"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -83,6 +83,11 @@ auth = [
"python-multipart>=0.0.6"
]
# Autonomy mode dependencies (code intelligence tools)
autonomy = [
"ast-grep-cli>=0.30.0"
]
# Combined features
all = [
"praisonaiagents[memory]",
@@ -0,0 +1,200 @@
"""
Integration tests for directory consistency with real API calls.
Tests that PRAISONAI_HOME overrides actually redirect all data,
and that a real agentic call works end-to-end with the wired paths.
"""
import os
import shutil
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
class TestPraisonaiHomeOverride:
"""Tests that PRAISONAI_HOME env var redirects all global paths."""
def test_all_global_paths_honor_override(self):
"""Every global helper should resolve under PRAISONAI_HOME."""
from praisonaiagents.paths import (
_clear_cache,
get_data_dir,
get_sessions_dir,
get_skills_dir,
get_plugins_dir,
get_mcp_dir,
get_docs_dir,
get_rules_dir,
get_permissions_dir,
get_storage_dir,
get_checkpoints_dir,
get_snapshots_dir,
get_learn_dir,
get_cache_dir,
get_mcp_auth_path,
get_memory_dir,
get_workflows_dir,
get_summaries_dir,
get_prp_dir,
get_runs_dir,
get_schedules_dir,
get_storage_path,
)
custom_home = tempfile.mkdtemp(prefix="praisonai_test_")
try:
_clear_cache()
with patch.dict(os.environ, {"PRAISONAI_HOME": custom_home}):
base = Path(custom_home)
assert get_data_dir() == base
assert get_sessions_dir() == base / "sessions"
assert get_skills_dir() == base / "skills"
assert get_plugins_dir() == base / "plugins"
assert get_mcp_dir() == base / "mcp"
assert get_docs_dir() == base / "docs"
assert get_rules_dir() == base / "rules"
assert get_permissions_dir() == base / "permissions"
assert get_storage_dir() == base / "storage"
assert get_checkpoints_dir() == base / "checkpoints"
assert get_snapshots_dir() == base / "snapshots"
assert get_learn_dir() == base / "learn"
assert get_cache_dir() == base / "cache"
assert get_mcp_auth_path() == base / "mcp-auth.json"
assert get_memory_dir() == base / "memory"
assert get_workflows_dir() == base / "workflows"
assert get_summaries_dir() == base / "summaries"
assert get_prp_dir() == base / "prp"
assert get_runs_dir() == base / "runs"
assert get_schedules_dir() == base / "schedules"
assert get_storage_path() == base / "storage.db"
finally:
_clear_cache()
shutil.rmtree(custom_home, ignore_errors=True)
def test_autonomy_config_respects_override(self):
"""AutonomyConfig.snapshot_dir should redirect with PRAISONAI_HOME."""
from praisonaiagents.paths import _clear_cache
custom_home = tempfile.mkdtemp(prefix="praisonai_test_")
try:
_clear_cache()
with patch.dict(os.environ, {"PRAISONAI_HOME": custom_home}):
from praisonaiagents.agent.autonomy import AutonomyConfig
config = AutonomyConfig()
assert config.snapshot_dir == str(Path(custom_home) / "snapshots")
finally:
_clear_cache()
shutil.rmtree(custom_home, ignore_errors=True)
def test_retrieval_config_defaults_to_project_local(self):
"""RetrievalConfig.persist_path should default to project-local .praisonai."""
from praisonaiagents.rag.retrieval_config import RetrievalConfig
from praisonaiagents.paths import get_project_data_dir
rc = RetrievalConfig()
assert rc.persist_path == str(get_project_data_dir())
def test_summarizer_defaults_to_project_local(self):
"""HierarchicalSummarizer._persist_path should default to project-local .praisonai/summaries."""
from praisonaiagents.rag.summarizer import HierarchicalSummarizer
from praisonaiagents.paths import get_project_summaries_dir
hs = HierarchicalSummarizer()
assert hs._persist_path == str(get_project_summaries_dir())
class TestRealAgenticCall:
"""Real agentic test — creates Agent, calls LLM, verifies path consistency."""
@pytest.mark.skipif(
not os.getenv("OPENAI_API_KEY"),
reason="OPENAI_API_KEY not set — skipping real LLM call"
)
def test_agent_start_with_custom_home(self):
"""
MANDATORY REAL AGENTIC TEST.
Creates Agent, calls agent.start(), verifies LLM responds.
"""
from praisonaiagents import Agent
agent = Agent(
name="path-test",
instructions="You are a helpful assistant. Respond in one sentence.",
)
result = agent.start("Say hello in one sentence")
# Agent must produce a non-empty text response from the LLM
assert result is not None
result_str = str(result)
assert len(result_str) > 0, "Agent returned empty response"
print(f"\n✓ Real agentic response: {result_str[:200]}")
@pytest.mark.skipif(
not os.getenv("OPENAI_API_KEY"),
reason="OPENAI_API_KEY not set — skipping real LLM call"
)
def test_agent_with_praisonai_home_override(self):
"""
Run agent with PRAISONAI_HOME override and verify data paths work.
"""
from praisonaiagents.paths import _clear_cache
custom_home = tempfile.mkdtemp(prefix="praisonai_agent_test_")
try:
_clear_cache()
os.environ["PRAISONAI_HOME"] = custom_home
from praisonaiagents import Agent
agent = Agent(
name="home-test",
instructions="You are a helpful assistant.",
)
result = agent.start("Say 'path test passed' in one sentence")
assert result is not None
print(f"\n✓ Agent with custom PRAISONAI_HOME: {str(result)[:200]}")
finally:
os.environ.pop("PRAISONAI_HOME", None)
_clear_cache()
shutil.rmtree(custom_home, ignore_errors=True)
class TestNoHardcodedPaths:
"""Verify no remaining hardcoded paths in the fixed modules."""
def test_no_expanduser_in_policy_config(self):
"""policy/config.py should not expanduser for rules."""
import inspect
from praisonaiagents.policy import config
source = inspect.getsource(config)
assert 'expanduser("~"), ".praisonai", "rules"' not in source
def test_no_expanduser_in_autonomy(self):
"""autonomy.py should not expanduser for snapshots."""
import inspect
from praisonaiagents.agent import autonomy
source = inspect.getsource(autonomy.AutonomyConfig.__post_init__)
assert 'expanduser("~"), ".praisonai", "snapshots"' not in source
def test_no_expanduser_in_scheduler_store(self):
"""scheduler/store.py should not expanduser for schedules."""
import inspect
from praisonaiagents.scheduler import store
source = inspect.getsource(store)
assert 'expanduser("~"), ".praisonai", "schedules"' not in source
def test_no_hardcoded_string_in_backends(self):
"""storage/backends.py should not have hardcoded ~/.praisonai/storage.db default."""
import inspect
from praisonaiagents.storage.backends import SQLiteBackend
source = inspect.getsource(SQLiteBackend.__init__)
assert '"~/.praisonai/storage.db"' not in source
@@ -16,6 +16,17 @@ requires_knowledge = pytest.mark.skipif(
)
def _make_knowledge_or_skip():
"""Create a Knowledge() instance, skipping if OpenAI API key is unavailable."""
from praisonaiagents import Knowledge
try:
return Knowledge()
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required for embeddings: {e}")
raise
@requires_knowledge
class TestDirectoryIngestion:
"""Test that directories are properly processed and file contents stored."""
@@ -42,10 +53,14 @@ class TestDirectoryIngestion:
def test_directory_ingestion_stores_text_not_path(self, temp_knowledge_dir):
"""Test that ingesting a directory stores file contents, not the directory path."""
from praisonaiagents import Knowledge
knowledge = _make_knowledge_or_skip()
knowledge = Knowledge()
result = knowledge.add(temp_knowledge_dir, user_id='test_user')
try:
result = knowledge.add(temp_knowledge_dir, user_id='test_user')
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required: {e}")
raise
# Should have results from files
assert 'results' in result
@@ -65,10 +80,14 @@ class TestDirectoryIngestion:
def test_directory_ingestion_processes_multiple_files(self, temp_knowledge_dir):
"""Test that all files in a directory are processed."""
from praisonaiagents import Knowledge
knowledge = _make_knowledge_or_skip()
knowledge = Knowledge()
result = knowledge.add(temp_knowledge_dir, user_id='test_user')
try:
knowledge.add(temp_knowledge_dir, user_id='test_user')
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required: {e}")
raise
# Search for content from both files
search1 = knowledge.search('ZEBRA-71', user_id='test_user')
@@ -80,10 +99,14 @@ class TestDirectoryIngestion:
def test_directory_ingestion_sets_metadata(self, temp_knowledge_dir):
"""Test that file metadata is properly set."""
from praisonaiagents import Knowledge
knowledge = _make_knowledge_or_skip()
knowledge = Knowledge()
knowledge.add(temp_knowledge_dir, user_id='test_user')
try:
knowledge.add(temp_knowledge_dir, user_id='test_user')
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required: {e}")
raise
search_result = knowledge.search('ZEBRA-71', user_id='test_user')
@@ -114,15 +137,20 @@ class TestContextBuilderUsesText:
"""Test that Agent._get_knowledge_context returns text, not path."""
from praisonaiagents import Agent
agent = Agent(
name='TestAgent',
instructions='Answer based on knowledge.',
knowledge=[temp_knowledge_dir],
output='silent',
)
# Ensure knowledge is processed
agent._ensure_knowledge_processed()
try:
agent = Agent(
name='TestAgent',
instructions='Answer based on knowledge.',
knowledge=[temp_knowledge_dir],
output='silent',
)
# Ensure knowledge is processed
agent._ensure_knowledge_processed()
except Exception as e:
if "api_key" in str(e).lower() or "openai" in str(e).lower():
pytest.skip(f"Skipping: OpenAI API key required: {e}")
raise
# Get context
context, _ = agent._get_knowledge_context('What is the password?', use_rag=True)
@@ -23,14 +23,18 @@ from unittest.mock import patch, MagicMock, PropertyMock
class TestDoomLoopFiring:
"""G1/G2: Doom loop detection must actually fire in run_autonomous()."""
def test_doom_loop_fires_when_agent_repeats_same_response(self):
@patch("praisonaiagents.agent.agent.Agent.get_recommended_stage", return_value="heuristic")
def test_doom_loop_fires_when_agent_repeats_same_response(self, mock_stage):
"""run_autonomous() must call _record_action() so DoomLoopTracker gets data."""
from praisonaiagents import Agent
agent = Agent(
name="test_doom",
instructions="You are a test agent",
autonomy={"max_iterations": 10, "doom_loop_threshold": 3},
# Need enough iterations for graduated recovery to exhaust all 4 stages:
# threshold=3 means 3 identical actions per doom-loop cycle, 4 recovery
# stages, plus check iterations = ~16 iterations minimum.
autonomy={"max_iterations": 25, "doom_loop_threshold": 3},
)
# Mock chat to return the same response every time
@@ -43,16 +47,16 @@ class TestDoomLoopFiring:
agent.chat = mock_chat
# Use a prompt that triggers autonomous stage (multi-step)
# so the loop runs multiple iterations instead of exiting after 1 (direct)
result = agent.run_autonomous("First refactor the auth module, then test it step by step")
# Doom loop should have been detected before max_iterations
assert result.completion_reason == "doom_loop", (
f"Expected doom_loop but got {result.completion_reason} after {result.iterations} iterations"
# Graduated recovery produces needs_help (stage 3) or doom_loop (stage 4).
# Both are valid doom-loop-family abort outcomes.
assert result.completion_reason in ("doom_loop", "needs_help"), (
f"Expected doom_loop or needs_help but got {result.completion_reason} "
f"after {result.iterations} iterations"
)
assert result.success is False
assert result.iterations < 10 # Should stop well before max
assert result.iterations < 25 # Should stop well before max
def test_doom_loop_does_not_fire_with_varied_responses(self):
"""Varied responses should NOT trigger doom loop."""
@@ -157,7 +157,11 @@ class TestDoomLoopRecovery:
agent = self._make_agent(threshold=2)
result = agent.run_autonomous("Do something")
assert result.success is False
assert result.completion_reason == "doom_loop"
# Graduated recovery produces needs_help (stage 3) or doom_loop (stage 4).
# Both are valid doom-loop-family abort outcomes.
assert result.completion_reason in ("doom_loop", "needs_help"), (
f"Expected doom_loop or needs_help but got {result.completion_reason}"
)
@patch("praisonaiagents.agent.agent.Agent.chat")
@patch("praisonaiagents.agent.agent.Agent.get_recommended_stage", return_value="heuristic")
@@ -0,0 +1,362 @@
"""
Tests for bot framework fixes.
Tests cover:
1. WhatsApp health() uses correct session manager attribute
2. asyncio.get_running_loop() usage (not deprecated get_event_loop)
3. Rate limiter functionality
4. Background task tracking
5. Deprecation warning for legacy approval
"""
import pytest
import time
class TestRateLimiter:
"""Tests for the RateLimiter utility."""
def test_rate_limiter_import(self):
"""Test that RateLimiter can be imported."""
from praisonai.bots._rate_limit import RateLimiter, RateLimitConfig
assert RateLimiter is not None
assert RateLimitConfig is not None
def test_rate_limiter_for_platform(self):
"""Test platform-specific rate limiter creation."""
from praisonai.bots._rate_limit import RateLimiter, PLATFORM_LIMITS
for platform in ["telegram", "discord", "slack", "whatsapp"]:
limiter = RateLimiter.for_platform(platform)
assert limiter is not None
assert limiter._config == PLATFORM_LIMITS[platform]
def test_rate_limiter_unknown_platform_uses_defaults(self):
"""Test that unknown platform uses default config."""
from praisonai.bots._rate_limit import RateLimiter, RateLimitConfig
limiter = RateLimiter.for_platform("unknown_platform")
default = RateLimitConfig()
assert limiter._config.messages_per_second == default.messages_per_second
@pytest.mark.asyncio
async def test_rate_limiter_acquire_no_wait_with_tokens(self):
"""Test that acquire doesn't wait when tokens available."""
from praisonai.bots._rate_limit import RateLimiter, RateLimitConfig
config = RateLimitConfig(messages_per_second=100, burst_size=10)
limiter = RateLimiter(config)
start = time.monotonic()
await limiter.acquire()
elapsed = time.monotonic() - start
# Should be nearly instant
assert elapsed < 0.1
@pytest.mark.asyncio
async def test_rate_limiter_acquire_waits_when_exhausted(self):
"""Test that acquire waits when tokens exhausted."""
from praisonai.bots._rate_limit import RateLimiter, RateLimitConfig
config = RateLimitConfig(messages_per_second=10, burst_size=1)
limiter = RateLimiter(config)
# Exhaust the single token
await limiter.acquire()
start = time.monotonic()
await limiter.acquire()
elapsed = time.monotonic() - start
# Should wait ~0.1s for token refill (1/10 = 0.1s)
assert elapsed >= 0.05 # Allow some tolerance
@pytest.mark.asyncio
async def test_rate_limiter_per_channel_delay(self):
"""Test per-channel delay enforcement."""
from praisonai.bots._rate_limit import RateLimiter, RateLimitConfig
config = RateLimitConfig(
messages_per_second=100,
per_channel_delay=0.1,
burst_size=10
)
limiter = RateLimiter(config)
channel = "test_channel"
# First message - no delay
await limiter.acquire(channel)
# Second message to same channel - should delay
start = time.monotonic()
await limiter.acquire(channel)
elapsed = time.monotonic() - start
assert elapsed >= 0.05 # Allow tolerance
def test_rate_limiter_reset(self):
"""Test rate limiter reset."""
from praisonai.bots._rate_limit import RateLimiter, RateLimitConfig
config = RateLimitConfig(burst_size=5)
limiter = RateLimiter(config)
limiter._tokens = 0
limiter._channel_last_send["test"] = time.monotonic()
limiter.reset()
assert limiter._tokens == 5.0
assert len(limiter._channel_last_send) == 0
class TestAsyncioGetRunningLoop:
"""Tests to verify get_running_loop is used instead of deprecated get_event_loop."""
def test_session_manager_uses_get_running_loop(self):
"""Verify _session.py uses get_running_loop."""
import inspect
from praisonai.bots._session import BotSessionManager
source = inspect.getsource(BotSessionManager)
assert "get_running_loop" in source
assert "get_event_loop()" not in source
def test_approval_uses_get_running_loop(self):
"""Verify _approval.py uses get_running_loop."""
# Import will trigger deprecation warning, that's expected
import warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
from praisonai.bots._approval import BotApprovalBackend
import inspect
source = inspect.getsource(BotApprovalBackend)
assert "get_running_loop" in source
class TestWhatsAppSessionManagerAttribute:
"""Tests for WhatsApp bot session manager attribute fix."""
def test_whatsapp_health_uses_session_mgr(self):
"""Verify WhatsApp health() uses _session_mgr not _session."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.health)
# Should use _session_mgr
assert "_session_mgr" in source
# Should NOT use bare _session (except as part of _session_mgr)
# Count occurrences
session_mgr_count = source.count("_session_mgr")
bare_session_count = source.count("self._session") - source.count("self._session_mgr")
assert bare_session_count == 0, "health() should not use self._session"
assert session_mgr_count > 0, "health() should use self._session_mgr"
class TestWhatsAppHttpSession:
"""Tests for WhatsApp shared HTTP session."""
def test_whatsapp_has_http_session_attribute(self):
"""Verify WhatsApp bot has _http_session attribute."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.__init__)
assert "_http_session" in source
def test_whatsapp_creates_session_on_start(self):
"""Verify WhatsApp creates ClientSession on start."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot._start_cloud_mode)
assert "self._http_session = aiohttp.ClientSession" in source
def test_whatsapp_closes_session_on_stop(self):
"""Verify WhatsApp closes ClientSession on stop."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.stop)
assert "_http_session.close()" in source
class TestWhatsAppBackgroundTasks:
"""Tests for WhatsApp background task tracking."""
def test_whatsapp_has_background_tasks_set(self):
"""Verify WhatsApp bot has _background_tasks set."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.__init__)
assert "_background_tasks" in source
def test_whatsapp_tracks_webhook_tasks(self):
"""Verify webhook processing tasks are tracked."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot._handle_webhook)
assert "self._background_tasks.add(task)" in source
assert "add_done_callback" in source
def test_whatsapp_cancels_tasks_on_stop(self):
"""Verify background tasks are cancelled on stop."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.stop)
assert "task.cancel()" in source or "for task in self._background_tasks" in source
class TestLegacyApprovalDeprecation:
"""Tests for legacy approval deprecation warning."""
def test_legacy_approval_emits_deprecation_warning(self):
"""Verify importing _approval.py emits DeprecationWarning."""
import warnings
import sys
import importlib
# Remove from cache to force re-import
if 'praisonai.bots._approval' in sys.modules:
del sys.modules['praisonai.bots._approval']
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
# Use importlib to avoid lint warning about unused import
importlib.import_module('praisonai.bots._approval')
# Check that a DeprecationWarning was raised
deprecation_warnings = [x for x in w if issubclass(x.category, DeprecationWarning)]
assert len(deprecation_warnings) >= 1
assert "deprecated" in str(deprecation_warnings[0].message).lower()
class TestAckReactorIntegration:
"""Tests for AckReactor integration across adapters."""
def test_discord_has_ack_reactor(self):
"""Verify Discord adapter has AckReactor."""
import inspect
from praisonai.bots.discord import DiscordBot
source = inspect.getsource(DiscordBot.__init__)
assert "_ack" in source
assert "AckReactor" in source
def test_discord_uses_ack_in_message_handling(self):
"""Verify Discord uses AckReactor in message handling."""
import inspect
from praisonai.bots.discord import DiscordBot
# Get the start method source which contains on_message
source = inspect.getsource(DiscordBot.start)
assert "self._ack.enabled" in source
assert "add_reaction" in source
assert "self._ack.done" in source
def test_slack_has_ack_reactor(self):
"""Verify Slack adapter has AckReactor."""
import inspect
from praisonai.bots.slack import SlackBot
source = inspect.getsource(SlackBot.__init__)
assert "_ack" in source
assert "AckReactor" in source
def test_slack_uses_ack_in_message_handling(self):
"""Verify Slack uses AckReactor in message handling."""
import inspect
from praisonai.bots.slack import SlackBot
source = inspect.getsource(SlackBot.start)
assert "self._ack.enabled" in source
assert "reactions_add" in source
assert "self._ack.done" in source
def test_whatsapp_has_ack_reactor(self):
"""Verify WhatsApp adapter has AckReactor."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.__init__)
assert "_ack" in source
assert "AckReactor" in source
def test_whatsapp_uses_ack_in_message_handling(self):
"""Verify WhatsApp uses AckReactor in message handling."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot._handle_incoming_message)
assert "self._ack.enabled" in source
assert "send_reaction" in source
assert "self._ack.done" in source
class TestWhatsAppRateLimiterIntegration:
"""Tests for WhatsApp rate limiter integration."""
def test_whatsapp_has_rate_limiter(self):
"""Verify WhatsApp adapter has rate limiter."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.__init__)
assert "_rate_limiter" in source
assert "RateLimiter.for_platform" in source
def test_whatsapp_send_message_uses_rate_limiter(self):
"""Verify WhatsApp send_message uses rate limiter."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.send_message)
assert "self._rate_limiter.acquire" in source
def test_whatsapp_send_template_uses_rate_limiter(self):
"""Verify WhatsApp send_template uses rate limiter."""
import inspect
from praisonai.bots.whatsapp import WhatsAppBot
source = inspect.getsource(WhatsAppBot.send_template)
assert "self._rate_limiter.acquire" in source
class TestPlatformLimits:
"""Tests for platform rate limit configurations."""
def test_telegram_limits(self):
"""Test Telegram rate limits are configured correctly."""
from praisonai.bots._rate_limit import PLATFORM_LIMITS
telegram = PLATFORM_LIMITS["telegram"]
assert telegram.messages_per_second >= 20 # ~30 per Telegram docs
assert telegram.burst_size >= 20
def test_discord_limits(self):
"""Test Discord rate limits are configured correctly."""
from praisonai.bots._rate_limit import PLATFORM_LIMITS
discord = PLATFORM_LIMITS["discord"]
assert discord.messages_per_second <= 2 # 5 per 5 seconds = 1/sec
assert discord.per_channel_delay >= 0.5
def test_slack_limits(self):
"""Test Slack rate limits are configured correctly."""
from praisonai.bots._rate_limit import PLATFORM_LIMITS
slack = PLATFORM_LIMITS["slack"]
assert slack.messages_per_second <= 2 # 1 per second per channel
assert slack.per_channel_delay >= 0.5
def test_whatsapp_limits(self):
"""Test WhatsApp rate limits are configured correctly."""
from praisonai.bots._rate_limit import PLATFORM_LIMITS
whatsapp = PLATFORM_LIMITS["whatsapp"]
assert whatsapp.messages_per_second >= 30 # ~80 per Cloud API docs
@@ -0,0 +1,322 @@
"""
Unit tests for EditorOutput module.
Tests the user-friendly editor-style output mode with:
- Step numbering
- Human-readable tool labels
- Multi-agent prefix support
- Markdown export
- Thread safety
"""
import pytest
import threading
from unittest.mock import MagicMock, patch
class TestEditorOutputImports:
"""Test that EditorOutput can be imported correctly."""
def test_import_from_output_module(self):
"""Test lazy import from output module."""
from praisonaiagents.output import EditorOutput
assert EditorOutput is not None
def test_import_enable_disable(self):
"""Test enable/disable functions can be imported."""
from praisonaiagents.output import enable_editor_output, disable_editor_output
assert callable(enable_editor_output)
assert callable(disable_editor_output)
def test_import_tool_labels(self):
"""Test TOOL_LABELS can be imported."""
from praisonaiagents.output import TOOL_LABELS
assert isinstance(TOOL_LABELS, dict)
assert 'internet_search' in TOOL_LABELS
def test_import_block_types(self):
"""Test BlockType and DisplayBlock can be imported."""
from praisonaiagents.output import BlockType, DisplayBlock
assert BlockType.NARRATIVE.value == "narrative"
assert BlockType.COMMAND.value == "command"
class TestEditorOutputBasic:
"""Test basic EditorOutput functionality."""
def test_init_default(self):
"""Test default initialization."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
assert editor._step_count == 0
assert editor._blocks == []
assert editor._multi_agent_mode is False
def test_init_with_agent_name(self):
"""Test initialization with agent name."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False, agent_name="TestAgent")
assert editor._agent_name == "TestAgent"
def test_elapsed_time(self):
"""Test elapsed time tracking."""
from praisonaiagents.output.editor import EditorOutput
import time
editor = EditorOutput(use_rich=False)
time.sleep(0.1)
elapsed = editor.elapsed_time()
assert elapsed >= 0.1
class TestToolCall:
"""Test tool_call method."""
def test_tool_call_increments_step(self):
"""Test that tool_call increments step counter."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
editor.tool_call("test_tool")
assert editor._step_count == 1
editor.tool_call("test_tool")
assert editor._step_count == 2
def test_tool_call_with_known_tool(self):
"""Test tool_call with a known tool label."""
from praisonaiagents.output.editor import EditorOutput, TOOL_LABELS
editor = EditorOutput(use_rich=False)
# internet_search should have a label
assert 'internet_search' in TOOL_LABELS
editor.tool_call("internet_search", args={"query": "test"})
blocks = editor.get_blocks()
assert len(blocks) == 1
assert "Searching the web" in blocks[0].content
def test_tool_call_with_unknown_tool(self):
"""Test tool_call with an unknown tool."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
editor.tool_call("my_custom_tool")
blocks = editor.get_blocks()
assert len(blocks) == 1
assert "my_custom_tool" in blocks[0].content
def test_tool_call_result_formatting(self):
"""Test that JSON results are formatted nicely."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
# Test list result
editor.tool_call("test", result='[{"a": 1}, {"b": 2}]')
blocks = editor.get_blocks()
assert "Found 2 items" in blocks[0].output
def test_tool_call_success_result(self):
"""Test success result formatting."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
editor.tool_call("test", result='{"success": true}')
blocks = editor.get_blocks()
assert "Success" in blocks[0].output
class TestMultiAgentPrefix:
"""Test multi-agent prefix support."""
def test_no_prefix_by_default(self):
"""Test that no prefix is shown by default."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False, agent_name="Agent1")
prefix = editor._get_prefix()
assert prefix == ""
def test_prefix_when_multi_agent_enabled(self):
"""Test prefix is shown when multi-agent mode is enabled."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False, agent_name="Agent1")
editor.enable_multi_agent_mode()
prefix = editor._get_prefix()
assert prefix == "[Agent1] "
def test_set_agent_name(self):
"""Test setting agent name."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
editor.set_agent_name("NewAgent")
editor.enable_multi_agent_mode()
prefix = editor._get_prefix()
assert prefix == "[NewAgent] "
def test_prefix_override(self):
"""Test prefix can be overridden per-call."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False, agent_name="Agent1")
editor.enable_multi_agent_mode()
prefix = editor._get_prefix("Agent2")
assert prefix == "[Agent2] "
class TestAdditionalMethods:
"""Test additional display methods."""
def test_narrative(self):
"""Test narrative method."""
from praisonaiagents.output.editor import EditorOutput, BlockType
editor = EditorOutput(use_rich=False)
editor.narrative("This is a narrative block")
blocks = editor.get_blocks()
assert len(blocks) == 1
assert blocks[0].type == BlockType.NARRATIVE
assert blocks[0].content == "This is a narrative block"
def test_code(self):
"""Test code method."""
from praisonaiagents.output.editor import EditorOutput, BlockType
editor = EditorOutput(use_rich=False)
editor.code("print('hello')", language="python")
blocks = editor.get_blocks()
assert len(blocks) == 1
assert blocks[0].type == BlockType.CODE
assert blocks[0].metadata["language"] == "python"
def test_action(self):
"""Test action method."""
from praisonaiagents.output.editor import EditorOutput, BlockType
editor = EditorOutput(use_rich=False)
editor.action("File created", details=["path: /tmp/test.py"])
blocks = editor.get_blocks()
assert len(blocks) == 1
assert blocks[0].type == BlockType.ACTION
assert blocks[0].title == "File created"
def test_list_items(self):
"""Test list_items method."""
from praisonaiagents.output.editor import EditorOutput, BlockType
editor = EditorOutput(use_rich=False)
editor.list_items(["Item 1", "Item 2"], title="My List")
blocks = editor.get_blocks()
assert len(blocks) == 1
assert blocks[0].type == BlockType.LIST
assert blocks[0].items == ["Item 1", "Item 2"]
def test_error(self):
"""Test error method."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
# Just verify it doesn't crash
editor.error("Something went wrong")
def test_summary(self):
"""Test summary method."""
from praisonaiagents.output.editor import EditorOutput, BlockType
editor = EditorOutput(use_rich=False)
editor.summary("Completed", items=["Duration: 5s", "Steps: 3"])
blocks = editor.get_blocks()
assert len(blocks) == 1
assert blocks[0].type == BlockType.SUMMARY
class TestToMarkdown:
"""Test to_markdown export."""
def test_to_markdown_empty(self):
"""Test to_markdown with no blocks."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
md = editor.to_markdown()
assert md == ""
def test_to_markdown_with_blocks(self):
"""Test to_markdown with various blocks."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
editor.narrative("Hello world")
editor.tool_call("test_tool", result="Done")
editor.summary("Complete", items=["Step 1"])
md = editor.to_markdown()
assert "Hello world" in md
assert "Complete" in md
assert "Step 1" in md
class TestThreadSafety:
"""Test thread safety of EditorOutput."""
def test_concurrent_tool_calls(self):
"""Test that concurrent tool_call is thread-safe."""
from praisonaiagents.output.editor import EditorOutput
editor = EditorOutput(use_rich=False)
def call_tool(n):
for _ in range(10):
editor.tool_call(f"tool_{n}")
threads = [threading.Thread(target=call_tool, args=(i,)) for i in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
# Should have 50 steps total (5 threads * 10 calls)
assert editor._step_count == 50
assert len(editor.get_blocks()) == 50
class TestEnableDisable:
"""Test enable/disable functions."""
def test_enable_returns_editor(self):
"""Test that enable_editor_output returns an EditorOutput instance."""
from praisonaiagents.output.editor import enable_editor_output, disable_editor_output, EditorOutput
editor = enable_editor_output(use_color=False)
assert isinstance(editor, EditorOutput)
disable_editor_output()
def test_is_enabled(self):
"""Test is_editor_output_enabled function."""
from praisonaiagents.output.editor import (
enable_editor_output, disable_editor_output, is_editor_output_enabled
)
disable_editor_output()
assert is_editor_output_enabled() is False
enable_editor_output(use_color=False)
assert is_editor_output_enabled() is True
disable_editor_output()
assert is_editor_output_enabled() is False
def test_get_editor_output(self):
"""Test get_editor_output function."""
from praisonaiagents.output.editor import (
enable_editor_output, disable_editor_output, get_editor_output
)
disable_editor_output()
assert get_editor_output() is None
editor = enable_editor_output(use_color=False)
assert get_editor_output() is editor
disable_editor_output()
@@ -232,7 +232,10 @@ class TestMultiChannelGatewayConfig:
"""Test MultiChannelGatewayConfig with defaults."""
from praisonaiagents.gateway import MultiChannelGatewayConfig
config = MultiChannelGatewayConfig()
assert isinstance(config.gateway, GatewayConfig)
# Use qualname check to be robust against module reimport identity issues
assert type(config.gateway).__qualname__ == "GatewayConfig", (
f"Expected GatewayConfig, got {type(config.gateway).__qualname__}"
)
assert config.agents == {}
assert config.channels == {}
@@ -0,0 +1,186 @@
"""TDD Tests for Memory vs History Separation Fixes.
Tests G-1 (auto_save_session routes to SessionStore) and
G-2 (Session._save_agent_chat_histories routes to SessionStore).
These tests verify the clean separation between:
- Memory = semantic facts/entities (injected into system prompt)
- History = conversation turns (injected into message array)
"""
import pytest
from unittest.mock import MagicMock, patch
class TestG1AutoSaveSessionRouting:
"""G-1: _auto_save_session() should route to SessionStore first."""
def test_auto_save_uses_session_store_when_available(self):
"""When SessionStore is available, use it instead of Memory.save_session()."""
from praisonaiagents import Agent
# Create agent with auto_save
agent = Agent(
name="test",
instructions="Test agent",
auto_save="test_session",
)
# Mock session store
mock_session_store = MagicMock()
agent._session_store = mock_session_store
# Add some chat history
agent.chat_history = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
]
# Call _auto_save_session
agent._auto_save_session()
# Verify SessionStore.add_message was called
assert mock_session_store.add_message.call_count == 2
def test_auto_save_falls_back_to_memory_when_no_session_store(self):
"""When SessionStore is not available, fall back to Memory.save_session()."""
from praisonaiagents import Agent
# Create agent with auto_save and memory
agent = Agent(
name="test",
instructions="Test agent",
auto_save="test_session",
)
# No session store
agent._session_store = None
# Mock memory instance
mock_memory = MagicMock()
mock_memory.save_session = MagicMock()
agent._memory_instance = mock_memory
# Add some chat history
agent.chat_history = [
{"role": "user", "content": "Hello"},
]
# Call _auto_save_session
agent._auto_save_session()
# Verify Memory.save_session was called as fallback
mock_memory.save_session.assert_called_once()
def test_auto_save_does_nothing_when_disabled(self):
"""When auto_save is not set, do nothing."""
from praisonaiagents import Agent
agent = Agent(
name="test",
instructions="Test agent",
)
# No auto_save set
agent.auto_save = None
# Mock session store
mock_session_store = MagicMock()
agent._session_store = mock_session_store
# Call _auto_save_session
agent._auto_save_session()
# Verify nothing was called
mock_session_store.add_message.assert_not_called()
class TestG2SessionSaveAgentChatHistories:
"""G-2: Session._save_agent_chat_histories() should route to SessionStore first."""
def test_save_agent_histories_routes_to_session_store(self):
"""Verify _save_agent_chat_histories has SessionStore routing logic."""
from praisonaiagents.session import Session
import inspect
# Verify the method exists and has the G-2 fix comment
source = inspect.getsource(Session._save_agent_chat_histories)
# Check that the G-2 fix is present
assert "G-2 FIX" in source, "G-2 fix should be documented in the method"
assert "SessionStore" in source, "Method should reference SessionStore"
assert "get_default_session_store" in source, "Method should try to get SessionStore"
class TestMemoryHistorySeparation:
"""Test that Memory and History are properly separated."""
def test_memory_context_does_not_include_conversation_turns(self):
"""Memory.get_context() should return facts/entities, not conversation turns."""
from praisonaiagents.memory import FileMemory
memory = FileMemory(user_id="test_user")
# Store some facts using correct API
memory.add_long_term("User prefers Python programming")
memory.add_entity("Alice", "person", {"role": "developer"})
# Get context
context = memory.get_context()
# Should contain facts and entities
assert "Important Facts" in context or "Known Entities" in context or context == ""
# Should NOT contain conversation turn markers
assert "role" not in context.lower() or "user:" not in context.lower()
def test_session_store_stores_conversation_turns(self):
"""SessionStore should store conversation turns with role/content."""
from praisonaiagents.session import get_default_session_store
store = get_default_session_store()
# Add messages
store.add_message("test_session", role="user", content="Hello")
store.add_message("test_session", role="assistant", content="Hi!")
# Get history
history = store.get_chat_history("test_session")
# Should have conversation turns
assert len(history) >= 2
assert any(msg.get("role") == "user" for msg in history)
assert any(msg.get("role") == "assistant" for msg in history)
class TestG3PerTurnPersist:
"""G-3: Verify per-turn persist exists via _persist_message()."""
def test_persist_message_exists(self):
"""Agent should have _persist_message method for per-turn persistence."""
from praisonaiagents import Agent
agent = Agent(name="test", instructions="Test")
# Verify method exists
assert hasattr(agent, '_persist_message')
assert callable(agent._persist_message)
def test_persist_message_uses_session_store(self):
"""_persist_message should use SessionStore when available."""
from praisonaiagents import Agent
# Create agent without session_id (it's set via memory config)
agent = Agent(name="test", instructions="Test")
# Mock session store
mock_store = MagicMock()
agent._session_store = mock_store
agent._session_id = "test_session"
agent._db = None # No DB adapter
# Call _persist_message
agent._persist_message("user", "Hello")
# Verify SessionStore was used
mock_store.add_user_message.assert_called_once_with("test_session", "Hello")
@@ -322,3 +322,146 @@ class TestImportPerformance:
# Should import in under 10ms (no heavy deps)
assert elapsed < 0.01, f"Import took {elapsed*1000:.2f}ms, expected < 10ms"
class TestProjectLocalHelpers:
"""Tests for project-local path helpers (get_project_*_dir)."""
def test_project_sessions_dir(self):
"""get_project_sessions_dir returns .praisonai/sessions/ in cwd."""
from praisonaiagents.paths import get_project_sessions_dir
result = get_project_sessions_dir()
assert result == Path.cwd() / ".praisonai" / "sessions"
def test_project_sessions_dir_custom_path(self, tmp_path):
"""get_project_sessions_dir accepts custom project path."""
from praisonaiagents.paths import get_project_sessions_dir
result = get_project_sessions_dir(tmp_path)
assert result == tmp_path / ".praisonai" / "sessions"
def test_project_knowledge_dir(self):
"""get_project_knowledge_dir returns .praisonai/knowledge/ in cwd."""
from praisonaiagents.paths import get_project_knowledge_dir
result = get_project_knowledge_dir()
assert result == Path.cwd() / ".praisonai" / "knowledge"
def test_project_knowledge_dir_custom_path(self, tmp_path):
"""get_project_knowledge_dir accepts custom project path."""
from praisonaiagents.paths import get_project_knowledge_dir
result = get_project_knowledge_dir(tmp_path)
assert result == tmp_path / ".praisonai" / "knowledge"
def test_project_summaries_dir(self):
"""get_project_summaries_dir returns .praisonai/summaries/ in cwd."""
from praisonaiagents.paths import get_project_summaries_dir
result = get_project_summaries_dir()
assert result == Path.cwd() / ".praisonai" / "summaries"
def test_project_prp_dir(self):
"""get_project_prp_dir returns .praisonai/prp/ in cwd."""
from praisonaiagents.paths import get_project_prp_dir
result = get_project_prp_dir()
assert result == Path.cwd() / ".praisonai" / "prp"
class TestGetSchedulesDir:
"""Tests for get_schedules_dir() function."""
def test_returns_schedules_subdir(self):
"""Should return schedules subdirectory of data dir."""
from praisonaiagents.paths import get_schedules_dir
result = get_schedules_dir()
assert result == Path.home() / ".praisonai" / "schedules"
def test_respects_env_override(self):
"""Should respect PRAISONAI_HOME override."""
from praisonaiagents.paths import get_schedules_dir
with patch.dict(os.environ, {"PRAISONAI_HOME": "/custom"}):
result = get_schedules_dir()
assert result == Path("/custom/schedules")
class TestGetStoragePath:
"""Tests for get_storage_path() function."""
def test_returns_storage_db(self):
"""Should return storage.db in data dir."""
from praisonaiagents.paths import get_storage_path
result = get_storage_path()
assert result == Path.home() / ".praisonai" / "storage.db"
def test_respects_env_override(self):
"""Should respect PRAISONAI_HOME override."""
from praisonaiagents.paths import get_storage_path
with patch.dict(os.environ, {"PRAISONAI_HOME": "/custom"}):
result = get_storage_path()
assert result == Path("/custom/storage.db")
class TestModuleWiring:
"""Tests verifying that modules use paths.py instead of hardcoded paths."""
def test_policy_config_uses_paths(self):
"""policy/config.py should resolve rules dir via paths.py."""
from praisonaiagents.policy.config import DEFAULT_RULES_DIR
from praisonaiagents.paths import get_rules_dir
assert DEFAULT_RULES_DIR == str(get_rules_dir())
def test_policy_config_no_hardcoded_expanduser(self):
"""policy/config.py should not have hardcoded expanduser for rules."""
import inspect
from praisonaiagents.policy import config
source = inspect.getsource(config)
# The module should not contain expanduser + .praisonai for rules
assert 'expanduser("~"), ".praisonai", "rules"' not in source
def test_autonomy_snapshot_dir_defaults_via_paths(self):
"""autonomy.py AutonomyConfig should default snapshot_dir via paths.py."""
from praisonaiagents.agent.autonomy import AutonomyConfig
from praisonaiagents.paths import get_snapshots_dir
config = AutonomyConfig()
assert config.snapshot_dir == str(get_snapshots_dir())
def test_autonomy_snapshot_dir_respects_env(self):
"""AutonomyConfig snapshot_dir should respect PRAISONAI_HOME."""
from praisonaiagents.paths import _clear_cache
_clear_cache()
with patch.dict(os.environ, {"PRAISONAI_HOME": "/custom"}):
from praisonaiagents.agent.autonomy import AutonomyConfig
config = AutonomyConfig()
assert config.snapshot_dir == str(Path("/custom/snapshots"))
_clear_cache()
def test_scheduler_store_uses_paths(self):
"""scheduler/store.py should resolve default dir via paths.py."""
from praisonaiagents.scheduler.store import _DEFAULT_DIR
from praisonaiagents.paths import get_schedules_dir
assert _DEFAULT_DIR == str(get_schedules_dir())
def test_sqlite_backend_default_uses_paths(self):
"""SQLiteBackend default db_path should use paths.py."""
from praisonaiagents.storage.backends import SQLiteBackend
from praisonaiagents.paths import get_storage_path
# We can't construct SQLiteBackend without it trying to create the DB,
# so we verify the import is present
import inspect
source = inspect.getsource(SQLiteBackend.__init__)
assert "get_storage_path" in source
@@ -0,0 +1,456 @@
"""
Unit tests verifying that all modules use paths.py instead of hardcoded paths.
Tests every module fixed in Phases 2, 3, and 3.5 of the path standardization effort.
Validates both source-level absence of hardcoded strings AND runtime behavior.
"""
import inspect
import os
from pathlib import Path
from unittest.mock import patch
import pytest
# ---------------------------------------------------------------------------
# Phase 2 Modules (SDK core)
# ---------------------------------------------------------------------------
class TestPolicyConfigWiring:
"""Verify policy/config.py routes through paths.py."""
def test_default_rules_dir_matches_paths(self):
from praisonaiagents.policy.config import DEFAULT_RULES_DIR
from praisonaiagents.paths import get_rules_dir
assert DEFAULT_RULES_DIR == str(get_rules_dir())
def test_no_hardcoded_expanduser(self):
from praisonaiagents.policy import config
source = inspect.getsource(config)
assert 'expanduser("~"), ".praisonai", "rules"' not in source
assert 'os.path.join(os.path.expanduser("~")' not in source
class TestAutonomyWiring:
"""Verify agent/autonomy.py routes through paths.py."""
def test_default_snapshot_dir_matches_paths(self):
from praisonaiagents.agent.autonomy import AutonomyConfig
from praisonaiagents.paths import get_snapshots_dir
config = AutonomyConfig()
assert config.snapshot_dir == str(get_snapshots_dir())
def test_snapshot_dir_respects_env(self):
from praisonaiagents.paths import _clear_cache
_clear_cache()
with patch.dict(os.environ, {"PRAISONAI_HOME": "/custom"}):
from praisonaiagents.agent.autonomy import AutonomyConfig
config = AutonomyConfig()
assert config.snapshot_dir == str(Path("/custom/snapshots"))
_clear_cache()
def test_no_hardcoded_expanduser(self):
from praisonaiagents.agent import autonomy
source = inspect.getsource(autonomy.AutonomyConfig.__post_init__)
assert 'expanduser' not in source
class TestSchedulerStoreWiring:
"""Verify scheduler/store.py routes through paths.py."""
def test_default_dir_matches_paths(self):
from praisonaiagents.scheduler.store import _DEFAULT_DIR
from praisonaiagents.paths import get_schedules_dir
assert _DEFAULT_DIR == str(get_schedules_dir())
def test_no_hardcoded_expanduser(self):
from praisonaiagents.scheduler import store
source = inspect.getsource(store)
assert 'expanduser("~"), ".praisonai"' not in source
class TestStorageBackendsWiring:
"""Verify storage/backends.py routes through paths.py."""
def test_sqlite_init_uses_get_storage_path(self):
from praisonaiagents.storage.backends import SQLiteBackend
source = inspect.getsource(SQLiteBackend.__init__)
assert "get_storage_path" in source
assert '"~/.praisonai/storage.db"' not in source
# ---------------------------------------------------------------------------
# Phase 3 Modules (project-local)
# ---------------------------------------------------------------------------
class TestSessionApiWiring:
"""Verify session/api.py routes through paths.py."""
def test_no_hardcoded_praisonai_sessions(self):
from praisonaiagents.session import api
source = inspect.getsource(api)
assert '".praisonai/sessions/' not in source
assert "f'.praisonai/sessions" not in source
def test_has_get_session_dir_method(self):
from praisonaiagents.session.api import Session
assert hasattr(Session, '_get_session_dir')
def test_get_session_dir_uses_paths(self):
from praisonaiagents.session.api import Session
source = inspect.getsource(Session._get_session_dir)
assert "get_project_sessions_dir" in source
class TestKnowledgeWiring:
"""Verify knowledge/knowledge.py routes through paths.py."""
def test_no_hardcoded_praisonai_literal(self):
from praisonaiagents.knowledge import knowledge
source = inspect.getsource(knowledge)
# Should not have hardcoded path construction with ".praisonai"
# But it's OK in docstrings/comments
lines = source.split('\n')
code_lines = [l for l in lines
if l.strip() and not l.strip().startswith('#')
and not l.strip().startswith('"""')
and not l.strip().startswith("'''")
and 'docstring' not in l.lower()]
code_text = '\n'.join(code_lines)
assert 'Path(".praisonai"' not in code_text
assert '".praisonai"' not in code_text or 'DEFAULT_DIR_NAME' in source
class TestContextAgentWiring:
"""Verify agent/context_agent.py routes through paths.py."""
def test_prp_dir_uses_paths(self):
from praisonaiagents.agent import context_agent
source = inspect.getsource(context_agent)
assert "get_project_prp_dir" in source
def test_no_hardcoded_prp_path(self):
from praisonaiagents.agent import context_agent
source = inspect.getsource(context_agent)
assert 'Path(".praisonai/prp")' not in source
class TestSummarizerWiring:
"""Verify rag/summarizer.py routes through paths.py."""
def test_default_persist_path_matches_paths(self):
from praisonaiagents.rag.summarizer import HierarchicalSummarizer
from praisonaiagents.paths import get_project_summaries_dir
hs = HierarchicalSummarizer()
assert hs._persist_path == str(get_project_summaries_dir())
def test_no_hardcoded_summaries_path(self):
from praisonaiagents.rag import summarizer
source = inspect.getsource(summarizer.HierarchicalSummarizer.__init__)
assert '".praisonai/summaries"' not in source
class TestRetrievalConfigWiring:
"""Verify rag/retrieval_config.py routes through paths.py."""
def test_default_persist_path_matches_paths(self):
from praisonaiagents.rag.retrieval_config import RetrievalConfig
from praisonaiagents.paths import get_project_data_dir
rc = RetrievalConfig()
assert rc.persist_path == str(get_project_data_dir())
def test_no_hardcoded_persist_path(self):
from praisonaiagents.rag import retrieval_config
source = inspect.getsource(retrieval_config)
assert 'persist_path: str = ".praisonai"' not in source
# ---------------------------------------------------------------------------
# Phase 3.5 Modules (batch fix)
# ---------------------------------------------------------------------------
class TestMemoryMemoryWiring:
"""Verify memory/memory.py routes through paths.py."""
def test_no_hardcoded_makedirs(self):
from praisonaiagents.memory import memory
source = inspect.getsource(memory.Memory.__init__)
assert 'makedirs(".praisonai"' not in source
def test_no_hardcoded_short_db(self):
from praisonaiagents.memory import memory
source = inspect.getsource(memory.Memory.__init__)
assert '".praisonai/short_term.db"' not in source
def test_no_hardcoded_long_db(self):
from praisonaiagents.memory import memory
source = inspect.getsource(memory.Memory.__init__)
assert '".praisonai/long_term.db"' not in source
def test_uses_get_project_data_dir(self):
from praisonaiagents.memory import memory
source = inspect.getsource(memory.Memory.__init__)
assert "get_project_data_dir" in source
class TestMemoryHooksWiring:
"""Verify memory/hooks.py routes through paths.py."""
def test_config_file_uses_default_dir_name(self):
from praisonaiagents.memory import hooks
source = inspect.getsource(hooks)
# Should use DEFAULT_DIR_NAME for CONFIG_FILE
assert "DEFAULT_DIR_NAME" in source or "_DIR_NAME" in source
def test_no_hardcoded_config_file(self):
from praisonaiagents.memory import hooks
# Check the HooksManager class body for hardcoded CONFIG_FILE
source = inspect.getsource(hooks.HooksManager)
assert 'CONFIG_FILE = ".praisonai/hooks.json"' not in source
def test_workspace_dir_uses_default_dir_name(self):
from praisonaiagents.memory import hooks
source = inspect.getsource(hooks.HooksManager.__init__)
assert 'workspace_path / ".praisonai"' not in source
class TestDocsManagerWiring:
"""Verify memory/docs_manager.py routes through paths.py."""
def test_no_hardcoded_docs_dir_name(self):
from praisonaiagents.memory import docs_manager
source = inspect.getsource(docs_manager.DocsManager)
assert 'DOCS_DIR_NAME = ".praisonai/docs"' not in source
def test_global_docs_path_uses_get_docs_dir(self):
from praisonaiagents.memory import docs_manager
source = inspect.getsource(docs_manager.DocsManager.__init__)
assert "get_docs_dir" in source
def test_no_hardcoded_home_praisonai_docs(self):
from praisonaiagents.memory import docs_manager
source = inspect.getsource(docs_manager.DocsManager.__init__)
assert 'Path.home() / ".praisonai" / "docs"' not in source
class TestRulesManagerWiring:
"""Verify memory/rules_manager.py routes through paths.py."""
def test_no_hardcoded_rules_dir_name(self):
from praisonaiagents.memory import rules_manager
source = inspect.getsource(rules_manager.RulesManager)
assert 'RULES_DIR_NAME = ".praisonai/rules"' not in source
def test_global_rules_path_uses_get_rules_dir(self):
from praisonaiagents.memory import rules_manager
source = inspect.getsource(rules_manager.RulesManager.__init__)
assert "get_rules_dir" in source
def test_no_hardcoded_home_praisonai_rules(self):
from praisonaiagents.memory import rules_manager
source = inspect.getsource(rules_manager.RulesManager.__init__)
assert 'Path.home() / ".praisonai" / "rules"' not in source
class TestMcpConfigWiring:
"""Verify memory/mcp_config.py routes through paths.py."""
def test_no_hardcoded_mcp_dir_name(self):
from praisonaiagents.memory import mcp_config
source = inspect.getsource(mcp_config.MCPConfigManager)
assert 'MCP_DIR_NAME = ".praisonai/mcp"' not in source
def test_global_mcp_path_uses_get_mcp_dir(self):
from praisonaiagents.memory import mcp_config
source = inspect.getsource(mcp_config.MCPConfigManager.__init__)
assert "get_mcp_dir" in source
def test_no_hardcoded_home_praisonai_mcp(self):
from praisonaiagents.memory import mcp_config
source = inspect.getsource(mcp_config.MCPConfigManager.__init__)
assert 'Path.home() / ".praisonai" / "mcp"' not in source
class TestFileMemoryWiring:
"""Verify memory/file_memory.py routes through paths.py."""
def test_no_hardcoded_base_path(self):
from praisonaiagents.memory import file_memory
source = inspect.getsource(file_memory.FileMemory.__init__)
assert 'Path(".praisonai/memory")' not in source
def test_uses_get_project_data_dir(self):
from praisonaiagents.memory import file_memory
source = inspect.getsource(file_memory.FileMemory.__init__)
assert "get_project_data_dir" in source
class TestMemoryWorkflowsWiring:
"""Verify memory/workflows.py routes through paths.py."""
def test_no_hardcoded_workflows_dir(self):
from praisonaiagents.memory import workflows
source = inspect.getsource(workflows.WorkflowManager)
assert 'WORKFLOWS_DIR = ".praisonai/workflows"' not in source
def test_uses_default_dir_name(self):
from praisonaiagents.memory import workflows
source = inspect.getsource(workflows.WorkflowManager)
assert "DEFAULT_DIR_NAME" in source or "_DIR_NAME" in source
def test_no_hardcoded_checkpoints_dir(self):
from praisonaiagents.memory import workflows
source = inspect.getsource(workflows.WorkflowManager._get_checkpoints_dir)
assert '".praisonai"' not in source
class TestWorkflowsWorkflowsWiring:
"""Verify workflows/workflows.py routes through paths.py."""
def test_no_hardcoded_workflows_dir(self):
from praisonaiagents.workflows import workflows
source = inspect.getsource(workflows.WorkflowManager)
assert 'WORKFLOWS_DIR = ".praisonai/workflows"' not in source
def test_uses_default_dir_name(self):
from praisonaiagents.workflows import workflows
source = inspect.getsource(workflows.WorkflowManager)
assert "DEFAULT_DIR_NAME" in source or "_DIR_NAME" in source
def test_no_hardcoded_checkpoints_dir(self):
from praisonaiagents.workflows import workflows
source = inspect.getsource(workflows.WorkflowManager._get_checkpoints_dir)
assert '".praisonai"' not in source
class TestConfigLoaderWiring:
"""Verify config/loader.py routes through paths.py."""
def test_no_hardcoded_plugin_dirs(self):
from praisonaiagents.config import loader
source = inspect.getsource(loader)
assert '"./.praisonai/plugins/"' not in source
assert '"~/.praisonai/plugins/"' not in source
def test_config_search_uses_paths(self):
from praisonaiagents.config import loader
source = inspect.getsource(loader._find_config_file)
assert "get_project_data_dir" in source or "get_data_dir" in source
def test_global_config_path_uses_get_data_dir(self):
from praisonaiagents.config import loader
source = inspect.getsource(loader._find_config_file)
assert "get_data_dir" in source
class TestAgentsAgentsWiring:
"""Verify agents/agents.py routes through paths.py."""
def test_no_hardcoded_memory_db(self):
from praisonaiagents.agents import agents
source = inspect.getsource(agents)
assert '"./.praisonai/memory.db"' not in source
def test_no_hardcoded_chroma_db(self):
from praisonaiagents.agents import agents
source = inspect.getsource(agents)
assert '"./.praisonai/chroma_db"' not in source
def test_uses_get_project_data_dir(self):
from praisonaiagents.agents import agents
source = inspect.getsource(agents)
assert "get_project_data_dir" in source
class TestLegacySessionWiring:
"""Verify session.py (legacy) routes through paths.py."""
def test_has_get_session_dir_method(self):
from praisonaiagents.session import Session
assert hasattr(Session, '_get_session_dir')
def test_get_session_dir_uses_paths(self):
from praisonaiagents.session import Session
source = inspect.getsource(Session._get_session_dir)
assert "get_project_sessions_dir" in source
def test_no_hardcoded_session_path(self):
from praisonaiagents import session
source = inspect.getsource(session.Session.__init__)
assert 'f".praisonai/sessions/' not in source
assert "makedirs(f\".praisonai" not in source
# ---------------------------------------------------------------------------
# Cross-cutting: comprehensive hardcoded path scan
# ---------------------------------------------------------------------------
class TestNoHardcodedPathsComprehensive:
"""
Scan all fixed modules' source code for any remaining hardcoded
.praisonai path construction patterns.
"""
FIXED_MODULES = [
"praisonaiagents.policy.config",
"praisonaiagents.agent.autonomy",
"praisonaiagents.scheduler.store",
"praisonaiagents.storage.backends",
"praisonaiagents.session.api",
"praisonaiagents.knowledge.knowledge",
"praisonaiagents.agent.context_agent",
"praisonaiagents.rag.summarizer",
"praisonaiagents.rag.retrieval_config",
"praisonaiagents.memory.memory",
"praisonaiagents.memory.hooks",
"praisonaiagents.memory.docs_manager",
"praisonaiagents.memory.rules_manager",
"praisonaiagents.memory.mcp_config",
"praisonaiagents.memory.file_memory",
"praisonaiagents.memory.workflows",
"praisonaiagents.workflows.workflows",
"praisonaiagents.config.loader",
"praisonaiagents.agents.agents",
"praisonaiagents.session",
]
# Patterns that indicate hardcoded path construction (not just docs/comments)
FORBIDDEN_PATTERNS = [
'Path(".praisonai"',
'Path(".praisonai/',
'os.makedirs(".praisonai"',
"os.makedirs('.praisonai'",
'"./.praisonai/',
"'./.praisonai/",
'= ".praisonai/',
"= '.praisonai/",
]
@pytest.mark.parametrize("module_name", FIXED_MODULES)
def test_no_forbidden_patterns_in_code(self, module_name):
"""Each fixed module should not contain forbidden path patterns in code."""
import importlib
mod = importlib.import_module(module_name)
source = inspect.getsource(mod)
# Extract only code lines (skip docstrings and comments)
in_docstring = False
code_lines = []
for line in source.split('\n'):
stripped = line.strip()
if stripped.startswith('"""') or stripped.startswith("'''"):
if stripped.count('"""') == 1 or stripped.count("'''") == 1:
in_docstring = not in_docstring
continue
if in_docstring:
continue
if stripped.startswith('#'):
continue
code_lines.append(line)
code_text = '\n'.join(code_lines)
for pattern in self.FORBIDDEN_PATTERNS:
assert pattern not in code_text, (
f"Module {module_name} still contains forbidden pattern: {pattern!r}"
)
@@ -0,0 +1,188 @@
"""
Tests for streaming integration - verifying stream_emitter receives events through main path.
These tests verify that:
1. stream_callback and emit_events are properly threaded through chat_completion_with_tools
2. Agent.stream_emitter.emit receives StreamEvents during agent.start(stream=True)
3. StreamMetrics are populated when enable_metrics() is called
"""
import pytest
from unittest.mock import MagicMock, patch, AsyncMock
from typing import List
class TestStreamCallbackThreading:
"""Test that stream_callback is properly threaded through the call chain."""
def test_chat_completion_with_tools_accepts_stream_callback(self):
"""chat_completion_with_tools should accept stream_callback parameter."""
from praisonaiagents.llm.openai_client import OpenAIClient
import inspect
sig = inspect.signature(OpenAIClient.chat_completion_with_tools)
params = list(sig.parameters.keys())
assert 'stream_callback' in params, "stream_callback parameter missing"
assert 'emit_events' in params, "emit_events parameter missing"
def test_achat_completion_with_tools_accepts_stream_callback(self):
"""achat_completion_with_tools should accept stream_callback parameter."""
from praisonaiagents.llm.openai_client import OpenAIClient
import inspect
sig = inspect.signature(OpenAIClient.achat_completion_with_tools)
params = list(sig.parameters.keys())
assert 'stream_callback' in params, "stream_callback parameter missing"
assert 'emit_events' in params, "emit_events parameter missing"
def test_process_stream_response_accepts_stream_callback(self):
"""process_stream_response should accept stream_callback parameter."""
from praisonaiagents.llm.openai_client import OpenAIClient
import inspect
sig = inspect.signature(OpenAIClient.process_stream_response)
params = list(sig.parameters.keys())
assert 'stream_callback' in params, "stream_callback parameter missing"
assert 'emit_events' in params, "emit_events parameter missing"
class TestAgentStreamEmitterWiring:
"""Test that Agent properly wires stream_emitter through main path."""
def test_agent_has_stream_emitter_property(self):
"""Agent should have stream_emitter property."""
from praisonaiagents import Agent
agent = Agent(name="test", instructions="Test agent")
assert hasattr(agent, 'stream_emitter'), "Agent missing stream_emitter property"
def test_stream_emitter_has_emit_method(self):
"""stream_emitter should have emit method."""
from praisonaiagents import Agent
agent = Agent(name="test", instructions="Test agent")
assert hasattr(agent.stream_emitter, 'emit'), "stream_emitter missing emit method"
assert callable(agent.stream_emitter.emit), "emit should be callable"
def test_stream_emitter_has_add_callback_method(self):
"""stream_emitter should have add_callback method."""
from praisonaiagents import Agent
agent = Agent(name="test", instructions="Test agent")
assert hasattr(agent.stream_emitter, 'add_callback'), "stream_emitter missing add_callback"
assert callable(agent.stream_emitter.add_callback), "add_callback should be callable"
def test_stream_emitter_callback_receives_events(self):
"""Callbacks added to stream_emitter should receive events when emit is called."""
from praisonaiagents import Agent
from praisonaiagents.streaming.events import StreamEvent, StreamEventType
agent = Agent(name="test", instructions="Test agent")
received_events: List[StreamEvent] = []
def callback(event: StreamEvent):
received_events.append(event)
agent.stream_emitter.add_callback(callback)
# Manually emit an event to test callback wiring
test_event = StreamEvent(type=StreamEventType.DELTA_TEXT, content="Hello")
agent.stream_emitter.emit(test_event)
assert len(received_events) == 1, "Callback should receive emitted event"
assert received_events[0].type == StreamEventType.DELTA_TEXT
assert received_events[0].content == "Hello"
class TestStreamEventTypes:
"""Test StreamEvent types are properly defined."""
def test_stream_event_types_exist(self):
"""All required StreamEventType values should exist."""
from praisonaiagents.streaming.events import StreamEventType
required_types = [
'REQUEST_START',
'HEADERS_RECEIVED',
'FIRST_TOKEN',
'DELTA_TEXT',
'DELTA_TOOL_CALL',
'TOOL_CALL_END',
'LAST_TOKEN',
'STREAM_END',
'ERROR'
]
for event_type in required_types:
assert hasattr(StreamEventType, event_type), f"Missing StreamEventType.{event_type}"
def test_stream_event_has_required_fields(self):
"""StreamEvent should have required fields."""
from praisonaiagents.streaming.events import StreamEvent, StreamEventType
event = StreamEvent(type=StreamEventType.DELTA_TEXT, content="test")
assert hasattr(event, 'type')
assert hasattr(event, 'timestamp')
assert hasattr(event, 'content')
assert hasattr(event, 'tool_call')
assert hasattr(event, 'is_reasoning')
assert hasattr(event, 'agent_id')
class TestStreamMetrics:
"""Test StreamMetrics functionality."""
def test_stream_emitter_has_enable_metrics(self):
"""stream_emitter should have enable_metrics method."""
from praisonaiagents import Agent
agent = Agent(name="test", instructions="Test agent")
assert hasattr(agent.stream_emitter, 'enable_metrics')
assert callable(agent.stream_emitter.enable_metrics)
def test_stream_emitter_has_get_metrics(self):
"""stream_emitter should have get_metrics method."""
from praisonaiagents import Agent
agent = Agent(name="test", instructions="Test agent")
assert hasattr(agent.stream_emitter, 'get_metrics')
assert callable(agent.stream_emitter.get_metrics)
def test_metrics_updated_on_emit(self):
"""Metrics should be updated when events are emitted."""
from praisonaiagents import Agent
from praisonaiagents.streaming.events import StreamEvent, StreamEventType
agent = Agent(name="test", instructions="Test agent")
agent.stream_emitter.enable_metrics()
# Emit events
agent.stream_emitter.emit(StreamEvent(type=StreamEventType.REQUEST_START))
agent.stream_emitter.emit(StreamEvent(type=StreamEventType.FIRST_TOKEN, content="H"))
agent.stream_emitter.emit(StreamEvent(type=StreamEventType.DELTA_TEXT, content="ello"))
agent.stream_emitter.emit(StreamEvent(type=StreamEventType.LAST_TOKEN))
agent.stream_emitter.emit(StreamEvent(type=StreamEventType.STREAM_END))
metrics = agent.stream_emitter.get_metrics()
assert metrics is not None, "Metrics should be available after enable_metrics()"
assert metrics.token_count >= 1, "Token count should be tracked"
class TestStreamCallbackInChatKwargs:
"""Test that _chat_completion passes stream_callback in chat_kwargs."""
def test_chat_kwargs_includes_stream_callback(self):
"""_chat_completion should include stream_callback in chat_kwargs."""
from praisonaiagents import Agent
import inspect
# Read the source code of _chat_completion to verify stream_callback is passed
source = inspect.getsource(Agent._chat_completion)
assert 'stream_callback' in source, "_chat_completion should reference stream_callback"
assert 'stream_emitter.emit' in source, "_chat_completion should use stream_emitter.emit"
assert 'emit_events' in source, "_chat_completion should reference emit_events"
@@ -0,0 +1,437 @@
"""
TDD Tests for ast_grep tool.
Tests cover:
1. Tool availability detection (installed vs not installed)
2. Graceful fallback when not installed
3. Tool assignment to Agent
4. Autonomy mode integration
5. Search, rewrite, and scan operations
"""
import pytest
import subprocess
from unittest.mock import patch, MagicMock
class TestAstGrepToolAvailability:
"""Test ast-grep tool availability detection."""
def test_is_available_when_installed(self):
"""Tool should report available when sg binary exists."""
from praisonaiagents.tools.ast_grep_tool import is_ast_grep_available
with patch('shutil.which') as mock_which:
mock_which.return_value = '/usr/local/bin/sg'
assert is_ast_grep_available() is True
def test_is_not_available_when_not_installed(self):
"""Tool should report not available when sg binary doesn't exist."""
from praisonaiagents.tools import ast_grep_tool
from praisonaiagents.tools.ast_grep_tool import is_ast_grep_available
# Clear cache to ensure fresh check
ast_grep_tool._availability_cache = None
with patch('praisonaiagents.tools.ast_grep_tool.shutil.which') as mock_which:
mock_which.return_value = None
result = is_ast_grep_available()
assert result is False
def test_availability_cached(self):
"""Availability check should be cached for performance."""
from praisonaiagents.tools import ast_grep_tool
# Clear cache
ast_grep_tool._availability_cache = None
with patch('shutil.which') as mock_which:
mock_which.return_value = '/usr/local/bin/sg'
# First call
result1 = ast_grep_tool.is_ast_grep_available()
# Second call should use cache
result2 = ast_grep_tool.is_ast_grep_available()
# which should only be called once due to caching
assert mock_which.call_count == 1
assert result1 is True
assert result2 is True
class TestAstGrepToolGracefulFallback:
"""Test graceful fallback when ast-grep is not installed."""
def test_search_returns_error_when_not_installed(self):
"""Search should return helpful error when not installed."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = False
result = ast_grep_search("def $FN($$$)", lang="python", path=".")
assert "not installed" in result.lower() or "not available" in result.lower()
assert "pip install ast-grep-cli" in result
def test_rewrite_returns_error_when_not_installed(self):
"""Rewrite should return helpful error when not installed."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_rewrite
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = False
result = ast_grep_rewrite("def $FN($$$)", "def $FN(*args)", lang="python", path=".")
assert "not installed" in result.lower() or "not available" in result.lower()
def test_scan_returns_error_when_not_installed(self):
"""Scan should return helpful error when not installed."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_scan
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = False
result = ast_grep_scan(path=".")
assert "not installed" in result.lower() or "not available" in result.lower()
def test_tool_does_not_crash_when_not_installed(self):
"""Tool should never crash, even when not installed."""
from praisonaiagents.tools.ast_grep_tool import (
ast_grep_search, ast_grep_rewrite, ast_grep_scan
)
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = False
# None of these should raise exceptions
try:
ast_grep_search("pattern", lang="python", path=".")
ast_grep_rewrite("old", "new", lang="python", path=".")
ast_grep_scan(path=".")
except Exception as e:
pytest.fail(f"Tool crashed when not installed: {e}")
class TestAstGrepToolOperations:
"""Test ast-grep tool operations when installed."""
def test_search_executes_subprocess(self):
"""Search should execute sg CLI with correct arguments."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
with patch('subprocess.run') as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
stdout='{"matches": []}',
stderr=''
)
result = ast_grep_search("def $FN($$$)", lang="python", path="/test/path")
mock_run.assert_called_once()
call_args = mock_run.call_args
cmd_list = call_args[0][0]
assert 'sg' in cmd_list[0] or cmd_list[0] == 'sg'
assert '--pattern' in cmd_list or 'def $FN($$$)' in cmd_list
assert '--lang' in cmd_list or 'python' in cmd_list
# Verify result is returned
assert result is not None
def test_search_handles_subprocess_error(self):
"""Search should handle subprocess errors gracefully."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
with patch('subprocess.run') as mock_run:
mock_run.side_effect = subprocess.SubprocessError("Command failed")
result = ast_grep_search("pattern", lang="python", path=".")
assert "error" in result.lower() or "failed" in result.lower()
def test_rewrite_dry_run_by_default(self):
"""Rewrite should use dry-run by default for safety."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_rewrite
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
with patch('subprocess.run') as mock_run:
mock_run.return_value = MagicMock(
returncode=0,
stdout='No changes',
stderr=''
)
result = ast_grep_rewrite("old", "new", lang="python", path=".")
# Should not actually modify files without explicit flag
call_args = mock_run.call_args
cmd_list = call_args[0][0]
# Check that --update-all is NOT in the command (dry-run mode)
assert '--update-all' not in cmd_list
assert mock_run.called
assert result is not None
class TestAstGrepToolAgentIntegration:
"""Test ast-grep tool integration with Agent."""
def test_tool_can_be_assigned_to_agent(self):
"""Tool should be assignable to Agent via tools parameter."""
from praisonaiagents import Agent
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
# Should not raise
agent = Agent(
name="test",
instructions="Test agent",
tools=[ast_grep_search],
)
assert ast_grep_search in agent.tools or any(
getattr(t, '__name__', '') == 'ast_grep_search' for t in agent.tools
)
def test_tool_works_when_not_installed_but_assigned(self):
"""Agent should work even if ast-grep tool is assigned but not installed."""
from praisonaiagents import Agent
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = False
# Should not raise during agent creation
agent = Agent(
name="test",
instructions="Test agent",
tools=[ast_grep_search],
)
# Agent should be created successfully
assert agent is not None
def test_tool_import_via_lazy_loading(self):
"""Tool should be importable via lazy loading from tools package."""
# This tests the TOOL_MAPPINGS integration
from praisonaiagents.tools import ast_grep_search
assert callable(ast_grep_search)
def test_all_ast_grep_tools_importable(self):
"""All ast-grep tools should be importable."""
from praisonaiagents.tools import (
ast_grep_search,
ast_grep_rewrite,
ast_grep_scan,
)
assert callable(ast_grep_search)
assert callable(ast_grep_rewrite)
assert callable(ast_grep_scan)
class TestAstGrepToolAutonomyIntegration:
"""Test ast-grep tool integration with autonomy mode."""
def test_autonomy_config_has_default_tools_field(self):
"""AutonomyConfig should have default_tools field."""
from praisonaiagents.agent.autonomy import AutonomyConfig
config = AutonomyConfig()
assert hasattr(config, 'default_tools')
def test_autonomy_default_tools_includes_ast_grep(self):
"""Autonomy default_tools should include ast_grep when available."""
from praisonaiagents.tools.ast_grep_tool import get_ast_grep_tools
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
tools = get_ast_grep_tools()
# Should include ast_grep tools
tool_names = [getattr(t, '__name__', str(t)) for t in tools]
assert any('ast_grep' in name for name in tool_names)
def test_autonomy_default_tools_graceful_fallback(self):
"""get_ast_grep_tools should return tools even when ast-grep not installed."""
from praisonaiagents.tools.ast_grep_tool import get_ast_grep_tools
# get_ast_grep_tools always returns the tools - they handle unavailability internally
tools = get_ast_grep_tools()
# Should return a list with the 3 ast-grep tools
assert isinstance(tools, list)
assert len(tools) == 3
def test_agent_with_autonomy_gets_default_tools(self):
"""Agent with autonomy enabled should get default tools injected."""
from praisonaiagents import Agent
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
agent = Agent(
name="test",
instructions="Test agent",
autonomy=True,
)
# Agent should have ast-grep tools injected
assert agent is not None
assert len(agent.tools) > 0, "Agent with autonomy=True should have default tools"
# Verify ast_grep_search is in the tools
tool_names = [getattr(t, '__name__', str(t)) for t in agent.tools]
assert 'ast_grep_search' in tool_names, f"ast_grep_search should be in agent.tools, got: {tool_names}"
def test_agent_with_custom_default_tools(self):
"""Agent with AutonomyConfig(default_tools=[X]) should use custom tools."""
from praisonaiagents import Agent
from praisonaiagents.agent.autonomy import AutonomyConfig
def custom_tool():
"""A custom tool for testing."""
return "custom"
config = AutonomyConfig(default_tools=[custom_tool])
agent = Agent(
name="test",
instructions="Test agent",
autonomy=config,
)
# Agent should have the custom tool
assert custom_tool in agent.tools, "Custom default_tools should be injected"
def test_agent_autonomy_default_tools_in_config_dict(self):
"""AutonomyConfig.default_tools should be serialized to agent.autonomy_config dict."""
from praisonaiagents import Agent
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
agent = Agent(
name="test",
instructions="Test agent",
autonomy=True,
)
# default_tools should be in the autonomy_config dict
assert 'default_tools' in agent.autonomy_config, "default_tools should be in autonomy_config dict"
class TestAstGrepToolSchema:
"""Test ast-grep tool schema generation for LLM."""
def test_search_has_docstring(self):
"""Search tool should have proper docstring for LLM."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
assert ast_grep_search.__doc__ is not None
assert len(ast_grep_search.__doc__) > 50
def test_rewrite_has_docstring(self):
"""Rewrite tool should have proper docstring for LLM."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_rewrite
assert ast_grep_rewrite.__doc__ is not None
assert len(ast_grep_rewrite.__doc__) > 50
def test_scan_has_docstring(self):
"""Scan tool should have proper docstring for LLM."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_scan
assert ast_grep_scan.__doc__ is not None
assert len(ast_grep_scan.__doc__) > 50
class TestAstGrepToolEdgeCases:
"""Test edge cases and error handling."""
def test_empty_pattern_handled(self):
"""Empty pattern should be handled gracefully."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
with patch('subprocess.run') as mock_run:
mock_run.return_value = MagicMock(returncode=1, stdout='', stderr='Invalid pattern')
result = ast_grep_search("", lang="python", path=".")
# Should not crash, should return error message
assert isinstance(result, str)
def test_invalid_language_handled(self):
"""Invalid language should be handled gracefully."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
with patch('subprocess.run') as mock_run:
mock_run.return_value = MagicMock(
returncode=1, stdout='', stderr='Unknown language'
)
result = ast_grep_search("pattern", lang="invalid_lang", path=".")
assert isinstance(result, str)
def test_nonexistent_path_handled(self):
"""Nonexistent path should be handled gracefully."""
from praisonaiagents.tools.ast_grep_tool import ast_grep_search
with patch('praisonaiagents.tools.ast_grep_tool.is_ast_grep_available') as mock_avail:
mock_avail.return_value = True
with patch('subprocess.run') as mock_run:
mock_run.return_value = MagicMock(
returncode=1, stdout='', stderr='Path not found'
)
result = ast_grep_search("pattern", lang="python", path="/nonexistent/path")
assert isinstance(result, str)
class TestAstGrepToolImportPerformance:
"""Test that tool import doesn't impact performance."""
def test_lazy_import_no_subprocess_on_import(self):
"""Importing the tool should not execute subprocess."""
with patch('subprocess.run') as mock_run:
# Re-import to test
import importlib
import praisonaiagents.tools.ast_grep_tool as module
importlib.reload(module)
# subprocess.run should not be called on import
mock_run.assert_not_called()
def test_availability_not_checked_on_import(self):
"""Availability should not be checked on module import."""
with patch('shutil.which') as mock_which:
import importlib
import praisonaiagents.tools.ast_grep_tool as module
# Clear cache to force fresh check
module._availability_cache = None
importlib.reload(module)
# which should not be called on import
mock_which.assert_not_called()
@@ -0,0 +1,363 @@
"""
TDD Tests for ToolProfile system.
Tests the ToolProfile dataclass, built-in profiles, and profile resolution.
"""
import pytest
class TestToolProfileDataclass:
"""Test ToolProfile dataclass structure."""
def test_tool_profile_has_required_fields(self):
"""ToolProfile should have name, tools, and description fields."""
from praisonaiagents.tools.profiles import ToolProfile
profile = ToolProfile(
name="test",
tools=["tool1", "tool2"],
description="Test profile"
)
assert profile.name == "test"
assert profile.tools == ["tool1", "tool2"]
assert profile.description == "Test profile"
def test_tool_profile_description_optional(self):
"""ToolProfile description should default to empty string."""
from praisonaiagents.tools.profiles import ToolProfile
profile = ToolProfile(name="test", tools=["tool1"])
assert profile.description == ""
def test_tool_profile_is_dataclass(self):
"""ToolProfile should be a dataclass."""
from dataclasses import is_dataclass
from praisonaiagents.tools.profiles import ToolProfile
assert is_dataclass(ToolProfile)
class TestBuiltinProfiles:
"""Test built-in profiles are defined correctly."""
def test_builtin_profiles_dict_exists(self):
"""BUILTIN_PROFILES dict should exist."""
from praisonaiagents.tools.profiles import BUILTIN_PROFILES
assert isinstance(BUILTIN_PROFILES, dict)
assert len(BUILTIN_PROFILES) > 0
def test_code_intelligence_profile_exists(self):
"""code_intelligence profile should include ast-grep tools."""
from praisonaiagents.tools.profiles import BUILTIN_PROFILES
assert "code_intelligence" in BUILTIN_PROFILES
profile = BUILTIN_PROFILES["code_intelligence"]
assert "ast_grep_search" in profile.tools
assert "ast_grep_rewrite" in profile.tools
assert "ast_grep_scan" in profile.tools
def test_file_ops_profile_exists(self):
"""file_ops profile should include file operation tools."""
from praisonaiagents.tools.profiles import BUILTIN_PROFILES
assert "file_ops" in BUILTIN_PROFILES
profile = BUILTIN_PROFILES["file_ops"]
assert "read_file" in profile.tools
assert "write_file" in profile.tools
assert "list_files" in profile.tools
def test_shell_profile_exists(self):
"""shell profile should include shell tools."""
from praisonaiagents.tools.profiles import BUILTIN_PROFILES
assert "shell" in BUILTIN_PROFILES
profile = BUILTIN_PROFILES["shell"]
assert "execute_command" in profile.tools
def test_web_profile_exists(self):
"""web profile should include web search tools."""
from praisonaiagents.tools.profiles import BUILTIN_PROFILES
assert "web" in BUILTIN_PROFILES
profile = BUILTIN_PROFILES["web"]
assert "internet_search" in profile.tools or "search_web" in profile.tools
def test_code_exec_profile_exists(self):
"""code_exec profile should include code execution tools."""
from praisonaiagents.tools.profiles import BUILTIN_PROFILES
assert "code_exec" in BUILTIN_PROFILES
profile = BUILTIN_PROFILES["code_exec"]
assert "execute_code" in profile.tools
def test_schedule_profile_exists(self):
"""schedule profile should include schedule tools."""
from praisonaiagents.tools.profiles import BUILTIN_PROFILES
assert "schedule" in BUILTIN_PROFILES
profile = BUILTIN_PROFILES["schedule"]
assert "schedule_add" in profile.tools
assert "schedule_list" in profile.tools
assert "schedule_remove" in profile.tools
class TestAutonomyProfile:
"""Test the composite AUTONOMY_PROFILE."""
def test_autonomy_profile_exists(self):
"""AUTONOMY_PROFILE should exist."""
from praisonaiagents.tools.profiles import AUTONOMY_PROFILE
assert AUTONOMY_PROFILE is not None
assert AUTONOMY_PROFILE.name == "autonomy"
def test_autonomy_profile_includes_file_ops(self):
"""AUTONOMY_PROFILE should include file_ops tools."""
from praisonaiagents.tools.profiles import AUTONOMY_PROFILE
assert "read_file" in AUTONOMY_PROFILE.tools
assert "write_file" in AUTONOMY_PROFILE.tools
def test_autonomy_profile_includes_shell(self):
"""AUTONOMY_PROFILE should include shell tools."""
from praisonaiagents.tools.profiles import AUTONOMY_PROFILE
assert "execute_command" in AUTONOMY_PROFILE.tools
def test_autonomy_profile_includes_web(self):
"""AUTONOMY_PROFILE should include web tools."""
from praisonaiagents.tools.profiles import AUTONOMY_PROFILE
assert "internet_search" in AUTONOMY_PROFILE.tools or "search_web" in AUTONOMY_PROFILE.tools
def test_autonomy_profile_includes_code_intelligence(self):
"""AUTONOMY_PROFILE should include code_intelligence tools."""
from praisonaiagents.tools.profiles import AUTONOMY_PROFILE
assert "ast_grep_search" in AUTONOMY_PROFILE.tools
def test_autonomy_profile_has_many_tools(self):
"""AUTONOMY_PROFILE should have 15+ tools (not just 3)."""
from praisonaiagents.tools.profiles import AUTONOMY_PROFILE
assert len(AUTONOMY_PROFILE.tools) >= 15, f"Expected 15+ tools, got {len(AUTONOMY_PROFILE.tools)}"
class TestProfileRegistry:
"""Test profile registration and retrieval."""
def test_get_profile_returns_builtin(self):
"""get_profile should return built-in profiles."""
from praisonaiagents.tools.profiles import get_profile
profile = get_profile("file_ops")
assert profile.name == "file_ops"
assert "read_file" in profile.tools
def test_get_profile_raises_for_unknown(self):
"""get_profile should raise KeyError for unknown profiles."""
from praisonaiagents.tools.profiles import get_profile
with pytest.raises(KeyError):
get_profile("nonexistent_profile")
def test_register_profile_adds_custom(self):
"""register_profile should add custom profiles."""
from praisonaiagents.tools.profiles import (
ToolProfile, register_profile, get_profile, _custom_profiles
)
# Clean up any previous test registrations
if "test_custom" in _custom_profiles:
del _custom_profiles["test_custom"]
custom = ToolProfile(
name="test_custom",
tools=["custom_tool1", "custom_tool2"],
description="Test custom profile"
)
register_profile(custom)
retrieved = get_profile("test_custom")
assert retrieved.name == "test_custom"
assert "custom_tool1" in retrieved.tools
# Cleanup
del _custom_profiles["test_custom"]
def test_custom_profile_overrides_builtin(self):
"""Custom profiles should override built-in profiles with same name."""
from praisonaiagents.tools.profiles import (
ToolProfile, register_profile, get_profile, _custom_profiles
)
# Register a custom profile with same name as builtin
custom = ToolProfile(
name="file_ops",
tools=["custom_read", "custom_write"],
description="Custom file ops"
)
register_profile(custom)
retrieved = get_profile("file_ops")
assert "custom_read" in retrieved.tools
# Cleanup - restore original
del _custom_profiles["file_ops"]
class TestResolveProfiles:
"""Test resolve_profiles function."""
def test_resolve_single_profile(self):
"""resolve_profiles should return tools from a single profile."""
from praisonaiagents.tools.profiles import resolve_profiles
tools = resolve_profiles("file_ops")
assert "read_file" in tools
assert "write_file" in tools
def test_resolve_multiple_profiles(self):
"""resolve_profiles should combine tools from multiple profiles."""
from praisonaiagents.tools.profiles import resolve_profiles
tools = resolve_profiles("file_ops", "shell")
assert "read_file" in tools
assert "execute_command" in tools
def test_resolve_profiles_deduplicates(self):
"""resolve_profiles should deduplicate tools."""
from praisonaiagents.tools.profiles import resolve_profiles
# autonomy includes file_ops, so calling both should not duplicate
tools = resolve_profiles("autonomy")
# Count occurrences of read_file
count = tools.count("read_file")
assert count == 1, f"Expected 1 occurrence of read_file, got {count}"
def test_resolve_profiles_returns_list(self):
"""resolve_profiles should return a list."""
from praisonaiagents.tools.profiles import resolve_profiles
tools = resolve_profiles("shell")
assert isinstance(tools, list)
def test_resolve_profiles_raises_for_unknown(self):
"""resolve_profiles should raise KeyError for unknown profiles."""
from praisonaiagents.tools.profiles import resolve_profiles
with pytest.raises(KeyError):
resolve_profiles("nonexistent")
class TestAgentAutonomyWithProfiles:
"""Test that Agent with autonomy=True uses AUTONOMY_PROFILE tools."""
def test_agent_autonomy_gets_profile_tools_when_no_tools_provided(self):
"""Agent(autonomy=True) with no tools should get AUTONOMY_PROFILE tools."""
from praisonaiagents import Agent
# No tools provided - should inject AUTONOMY_PROFILE
agent = Agent(
name="test",
instructions="Test agent",
autonomy=True,
)
# Should have more than just 3 ast-grep tools
assert len(agent.tools) >= 15, f"Expected 15+ tools, got {len(agent.tools)}"
# Check for file_ops tools
tool_names = [getattr(t, '__name__', str(t)) for t in agent.tools]
# At least one of these should be present
has_file_tool = any(name in tool_names for name in ['read_file', 'write_file', 'list_files'])
assert has_file_tool or len(agent.tools) >= 15, f"Expected file tools, got: {tool_names}"
def test_agent_autonomy_does_not_inject_when_tools_provided(self):
"""Agent(autonomy=True) with tools should NOT inject AUTONOMY_PROFILE (avoid duplicates)."""
from praisonaiagents import Agent
def my_tool():
"""My custom tool."""
return "custom"
# Tools provided - should NOT inject AUTONOMY_PROFILE
agent = Agent(
name="test",
instructions="Test agent",
tools=[my_tool],
autonomy=True,
)
# Should only have the user-provided tool, not AUTONOMY_PROFILE
assert len(agent.tools) == 1, f"Expected 1 tool (user-provided), got {len(agent.tools)}"
assert my_tool in agent.tools
def test_agent_custom_default_tools_override(self):
"""Agent with custom default_tools should use those instead."""
from praisonaiagents import Agent
from praisonaiagents.agent.autonomy import AutonomyConfig
def custom_tool():
"""Custom tool."""
return "custom"
config = AutonomyConfig(default_tools=[custom_tool])
agent = Agent(
name="test",
instructions="Test agent",
autonomy=config,
)
# Should have only the custom tool
assert custom_tool in agent.tools
# Should NOT have the full autonomy profile
assert len(agent.tools) == 1
def test_agent_empty_default_tools_disables_injection(self):
"""Agent with default_tools=[] should have no auto-injected tools."""
from praisonaiagents import Agent
from praisonaiagents.agent.autonomy import AutonomyConfig
config = AutonomyConfig(default_tools=[])
agent = Agent(
name="test",
instructions="Test agent",
autonomy=config,
)
# Should have no tools
assert len(agent.tools) == 0
class TestProfileExports:
"""Test that profiles are properly exported."""
def test_import_from_tools_package(self):
"""ToolProfile should be importable from praisonaiagents.tools."""
from praisonaiagents.tools import ToolProfile
assert ToolProfile is not None
def test_import_register_profile(self):
"""register_profile should be importable from praisonaiagents.tools."""
from praisonaiagents.tools import register_profile
assert callable(register_profile)
def test_import_resolve_profiles(self):
"""resolve_profiles should be importable from praisonaiagents.tools."""
from praisonaiagents.tools import resolve_profiles
assert callable(resolve_profiles)
def test_import_autonomy_profile(self):
"""AUTONOMY_PROFILE should be importable from praisonaiagents.tools."""
from praisonaiagents.tools import AUTONOMY_PROFILE
assert AUTONOMY_PROFILE is not None
+21 -2
View File
@@ -145,6 +145,21 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/39/e3/893e8757be2612e6c266d9bb58ad2e3651524b5b40cf56761e985a28b13e/asgiref-3.8.1-py3-none-any.whl", hash = "sha256:3e1e3ecc849832fe52ccf2cb6686b7a55f82bb1d6aee72a58826471390335e47", size = 23828, upload-time = "2024-03-22T14:39:34.521Z" },
]
[[package]]
name = "ast-grep-cli"
version = "0.41.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/05/9c/4b5fbc28a606d1d3217891d41c8428518de8a52e8f83564daabbd4d8ba18/ast_grep_cli-0.41.0.tar.gz", hash = "sha256:9aee19fa53d582f4408ace5cffe3ff5635dff954c7ab78a15530c4acf5893a0e", size = 220762, upload-time = "2026-02-22T19:06:01.576Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/e2/e7604406b8487d43928a99fcfb66ac3aa9fba787527e901b96e65dbbd6d6/ast_grep_cli-0.41.0-py3-none-macosx_10_12_x86_64.macosx_11_0_arm64.macosx_10_12_universal2.whl", hash = "sha256:93085ee153855fa9b1f486884f8aad788fb111f1e6a96bcfde240071d5033199", size = 14488134, upload-time = "2026-02-22T19:05:54.26Z" },
{ url = "https://files.pythonhosted.org/packages/65/b5/1f9ad69e28e2b98e0e99a4a7f981982937072c2a667f5ae808d8ffefa193/ast_grep_cli-0.41.0-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:8db8e877b365a553f83a2a0e158d8cfa17fc33f8033774ee3248665449dcad1b", size = 7261908, upload-time = "2026-02-22T19:06:02.5Z" },
{ url = "https://files.pythonhosted.org/packages/f8/8b/ac2e7946e2b566483fb0005ed715767110663fb85a666030b38e43dee10f/ast_grep_cli-0.41.0-py3-none-manylinux_2_28_aarch64.whl", hash = "sha256:c10f70cd6e45ded98773022d9ae0bebfc497b8b2e85ae17e15abca174e7d0f93", size = 7106987, upload-time = "2026-02-22T19:05:59.003Z" },
{ url = "https://files.pythonhosted.org/packages/f8/00/90cc9f094fad9116f29076c1a5393df50fba7d7b25c402d2611d96f7e3f9/ast_grep_cli-0.41.0-py3-none-manylinux_2_28_x86_64.whl", hash = "sha256:92d384608e5fd98a301a03a380e0ce011d46adcd5f6bcf53b0b9ba125faa43ad", size = 7417585, upload-time = "2026-02-22T19:06:00.307Z" },
{ url = "https://files.pythonhosted.org/packages/be/1b/390fd8ab3164e0f9377bd272196612f5047bfed35eb704df1c1e84776075/ast_grep_cli-0.41.0-py3-none-win32.whl", hash = "sha256:a7a4682cde7a1e1a8083dd8feed6f968d8f90a109e7c0917e1d5066807dfa46e", size = 6974015, upload-time = "2026-02-22T19:05:55.956Z" },
{ url = "https://files.pythonhosted.org/packages/0a/f5/71984cca3d3533c7e0a7dfe1784b7f6b12931463f6c5ce95ec36d5169219/ast_grep_cli-0.41.0-py3-none-win_amd64.whl", hash = "sha256:00570076cbf30e9c4c93f5577137c4d63245974420eff9e1fb3e3c6a9abefe97", size = 7413122, upload-time = "2026-02-22T19:05:57.586Z" },
{ url = "https://files.pythonhosted.org/packages/c1/ee/2add23700fa89661e84301c458a9a3218836edd5bbaeea49a7a283f4269a/ast_grep_cli-0.41.0-py3-none-win_arm64.whl", hash = "sha256:b8e0eeeec751fa213abc57ea20af4fdd8ae955a43e390914df85985f6933eb53", size = 7089061, upload-time = "2026-02-22T19:05:52.28Z" },
]
[[package]]
name = "async-timeout"
version = "4.0.3"
@@ -2549,7 +2564,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "1.5.15"
version = "1.5.32"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
@@ -2589,6 +2604,9 @@ auth = [
{ name = "python-jose", extra = ["cryptography"] },
{ name = "python-multipart" },
]
autonomy = [
{ name = "ast-grep-cli" },
]
graph = [
{ name = "chromadb" },
{ name = "mem0ai", extra = ["graph"] },
@@ -2629,6 +2647,7 @@ telemetry = [
[package.metadata]
requires-dist = [
{ name = "aiohttp", specifier = ">=3.8.0" },
{ name = "ast-grep-cli", marker = "extra == 'autonomy'", specifier = ">=0.30.0" },
{ name = "chonkie", marker = "extra == 'knowledge'", specifier = ">=1.0.2" },
{ name = "chromadb", marker = "extra == 'graph'", specifier = ">=1.0.0" },
{ name = "chromadb", marker = "extra == 'knowledge'", specifier = ">=1.0.0" },
@@ -2669,7 +2688,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "auth", "all"]
provides-extras = ["mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "auth", "autonomy", "all"]
[[package]]
name = "propcache"
+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.5.15.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.5.15.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.5.32.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.5.32.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+165 -1
View File
@@ -11,7 +11,7 @@ This module uses FULL LAZY LOADING for all heavy dependencies:
This ensures minimal import-time overhead.
"""
from pydantic import BaseModel
from typing import Dict, List, Optional, Type, TypeVar
from typing import Any, Dict, List, Optional, Type, TypeVar
import os
import json
import yaml
@@ -1377,4 +1377,168 @@ Example structure:
with open(full_path, 'w') as f:
yaml.dump(workflow_yaml, f, default_flow_style=False, sort_keys=False)
return full_path
# =============================================================================
# Job Workflow Auto Generator (Strategy 4)
# =============================================================================
class JobWorkflowStep(BaseModel):
"""A single step in a job workflow."""
name: str
step_type: str # "agent", "judge", "approve", "run", "action"
config: Dict[str, Any]
class JobWorkflowStructure(BaseModel):
"""Structure for a job workflow with agent-centric steps."""
name: str
description: str
steps: List[JobWorkflowStep]
class JobWorkflowAutoGenerator(BaseAutoGenerator):
"""
Auto-generates job workflow YAML files with agent-centric steps.
Generates workflows with `type: job` that include:
- agent: steps for AI agent execution
- judge: steps for quality gates
- approve: steps for approval gates
- run: steps for shell commands
- action: steps for built-in actions
Usage:
generator = JobWorkflowAutoGenerator(topic="Generate changelog and publish")
path = generator.generate()
"""
def __init__(self, topic: str = "Automate a task",
workflow_file: str = "job_workflow.yaml",
config_list: Optional[List[Dict]] = None):
"""
Initialize the JobWorkflowAutoGenerator.
Args:
topic: The task/topic for the workflow
workflow_file: Output file name
config_list: Optional LLM configuration
"""
super().__init__(config_list=config_list)
self.topic = topic
self.workflow_file = workflow_file
def generate(self, include_judge: bool = True, include_approve: bool = False) -> str:
"""
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 = self._structured_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())
prompt = f"""Generate a job workflow structure for: "{self.topic}"
A job workflow uses `type: job` and supports these step types:
1. **agent** - AI agent execution (role, instructions, prompt, model, tools, output_file)
2. **judge** - Quality gate with threshold (input_file, criteria, threshold, on_fail)
3. **approve** - Approval gate (description, risk_level, auto_approve)
4. **run** - Shell command execution
5. **action** - Built-in action (e.g., bump-version)
REQUIREMENTS:
- Create 2-4 steps that accomplish the task
- At least one step MUST be an "agent" step
{"- Include a 'judge' step to validate quality" if include_judge else ""}
{"- Include an 'approve' step for human approval before critical actions" if include_approve else ""}
- Each step should have a clear name and purpose
Available tools for agent steps: {tools_list}
STEP CONFIG FORMATS:
- agent: {{"role": "...", "instructions": "...", "prompt": "...", "model": "gpt-4o-mini", "tools": [], "output_file": "..."}}
- judge: {{"input_file": "...", "criteria": "...", "threshold": 7.0, "on_fail": "stop"}}
- approve: {{"description": "...", "risk_level": "medium", "auto_approve": false}}
- run: {{"command": "..."}}
- action: {{"name": "bump-version", "strategy": "patch"}}
Generate a workflow for: {self.topic}
"""
return prompt
def _save_workflow(self, data: JobWorkflowStructure) -> str:
"""Save the job workflow to a YAML file."""
# Build the workflow YAML structure
workflow_yaml = {
'type': 'job',
'name': data.name,
'description': data.description,
'steps': []
}
# Convert steps
for step in data.steps:
step_dict = {'name': step.name}
if step.step_type == 'agent':
step_dict['agent'] = {
'role': step.config.get('role', 'Assistant'),
'instructions': step.config.get('instructions', ''),
'prompt': step.config.get('prompt', step.config.get('instructions', '')),
'model': step.config.get('model', 'gpt-4o-mini'),
}
if step.config.get('tools'):
step_dict['agent']['tools'] = step.config['tools']
if step.config.get('output_file'):
step_dict['output_file'] = step.config['output_file']
elif step.step_type == 'judge':
step_dict['judge'] = {
'input_file': step.config.get('input_file', ''),
'criteria': step.config.get('criteria', 'Output is high quality'),
'threshold': step.config.get('threshold', 7.0),
'on_fail': step.config.get('on_fail', 'stop'),
}
elif step.step_type == 'approve':
step_dict['approve'] = {
'description': step.config.get('description', 'Approve this step'),
'risk_level': step.config.get('risk_level', 'medium'),
'auto_approve': step.config.get('auto_approve', False),
}
elif step.step_type == 'run':
step_dict['run'] = step.config.get('command', 'echo "Step executed"')
elif step.step_type == 'action':
step_dict['action'] = step.config.get('name', 'bump-version')
if step.config.get('strategy'):
step_dict['strategy'] = step.config['strategy']
workflow_yaml['steps'].append(step_dict)
# Write to file
full_path = os.path.abspath(self.workflow_file)
with open(full_path, 'w') as f:
yaml.dump(workflow_yaml, f, default_flow_style=False, sort_keys=False)
return full_path
+10 -1
View File
@@ -9,6 +9,15 @@ from __future__ import annotations
import asyncio
import logging
# Module-level deprecation warning
import warnings as _deprecation_warnings
_deprecation_warnings.warn(
"BotApprovalBackend is deprecated. Use platform-specific approval backends "
"(SlackApproval, TelegramApproval, DiscordApproval, WebhookApproval, HTTPApproval) instead.",
DeprecationWarning,
stacklevel=2
)
import time
from typing import Any, Dict
@@ -77,7 +86,7 @@ class BotApprovalBackend:
msg = f"🔐 Approval required:\n\nTool: {tool_name}\nArgs: {args_preview}\n\nReply 'yes' to approve or 'no' to deny."
# Create a future to wait for the response
loop = asyncio.get_event_loop()
loop = asyncio.get_running_loop()
future: asyncio.Future = loop.create_future()
self._pending[request_id] = future
+141
View File
@@ -0,0 +1,141 @@
"""
Rate Limiting Utilities for Bot Adapters.
Provides platform-aware rate limiting to prevent 429 errors from messaging APIs.
Platform limits (as of 2024):
- Telegram: ~30 messages/second to different users, stricter per-chat
- Discord: 5 messages per 5 seconds per channel (1 msg/sec effective)
- Slack: 1 message per second per channel
- WhatsApp: ~80 messages/second (Cloud API), varies for Web
"""
from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass, field
from typing import Dict, Optional
import logging
logger = logging.getLogger(__name__)
@dataclass
class RateLimitConfig:
"""Configuration for rate limiting.
Attributes:
messages_per_second: Maximum messages per second (global)
per_channel_delay: Minimum delay between messages to same channel (seconds)
burst_size: Number of messages allowed in a burst before throttling
"""
messages_per_second: float = 1.0
per_channel_delay: float = 1.0
burst_size: int = 5
# Platform-specific defaults
PLATFORM_LIMITS: Dict[str, RateLimitConfig] = {
"telegram": RateLimitConfig(messages_per_second=25.0, per_channel_delay=0.05, burst_size=30),
"discord": RateLimitConfig(messages_per_second=1.0, per_channel_delay=1.0, burst_size=5),
"slack": RateLimitConfig(messages_per_second=1.0, per_channel_delay=1.0, burst_size=1),
"whatsapp": RateLimitConfig(messages_per_second=50.0, per_channel_delay=0.1, burst_size=80),
}
class RateLimiter:
"""Async rate limiter with per-channel tracking.
Uses token bucket algorithm for global rate and per-channel delays.
Example:
limiter = RateLimiter.for_platform("telegram")
async def send(channel_id, msg):
await limiter.acquire(channel_id)
await api.send_message(channel_id, msg)
"""
def __init__(self, config: Optional[RateLimitConfig] = None):
"""Initialize rate limiter.
Args:
config: Rate limit configuration. Defaults to 1 msg/sec.
"""
self._config = config or RateLimitConfig()
self._tokens = float(self._config.burst_size)
self._last_refill = time.monotonic()
self._channel_last_send: Dict[str, float] = {}
self._lock = asyncio.Lock()
@classmethod
def for_platform(cls, platform: str) -> "RateLimiter":
"""Create a rate limiter with platform-specific defaults.
Args:
platform: Platform name (telegram, discord, slack, whatsapp)
Returns:
Configured RateLimiter instance
"""
config = PLATFORM_LIMITS.get(platform.lower(), RateLimitConfig())
return cls(config)
async def acquire(self, channel_id: Optional[str] = None) -> None:
"""Wait until rate limit allows sending.
Args:
channel_id: Optional channel ID for per-channel limiting
"""
async with self._lock:
now = time.monotonic()
# Refill tokens based on elapsed time
elapsed = now - self._last_refill
self._tokens = min(
self._config.burst_size,
self._tokens + elapsed * self._config.messages_per_second
)
self._last_refill = now
# Wait for global token
if self._tokens < 1.0:
wait_time = (1.0 - self._tokens) / self._config.messages_per_second
logger.debug(f"Rate limit: waiting {wait_time:.3f}s for global token")
await asyncio.sleep(wait_time)
self._tokens = 1.0
# Consume token
self._tokens -= 1.0
# Per-channel delay
if channel_id:
last_send = self._channel_last_send.get(channel_id, 0.0)
channel_elapsed = now - last_send
if channel_elapsed < self._config.per_channel_delay:
wait_time = self._config.per_channel_delay - channel_elapsed
logger.debug(f"Rate limit: waiting {wait_time:.3f}s for channel {channel_id}")
await asyncio.sleep(wait_time)
self._channel_last_send[channel_id] = time.monotonic()
def reset(self) -> None:
"""Reset rate limiter state."""
self._tokens = float(self._config.burst_size)
self._last_refill = time.monotonic()
self._channel_last_send.clear()
def get_rate_limiter(platform: str) -> RateLimiter:
"""Get a rate limiter for the specified platform.
This is a convenience function that creates a new limiter each time.
For persistent limiting, create and store a RateLimiter instance.
Args:
platform: Platform name
Returns:
Configured RateLimiter
"""
return RateLimiter.for_platform(platform)
+1 -1
View File
@@ -145,7 +145,7 @@ class BotSessionManager:
agent_lock = self._get_agent_lock(agent)
async with user_lock:
# Load history (may hit disk via run_in_executor for async safety)
loop = asyncio.get_event_loop()
loop = asyncio.get_running_loop()
user_history = await loop.run_in_executor(
None, self._load_history, user_id
)
+28
View File
@@ -27,6 +27,7 @@ from praisonaiagents.bots import (
from ._commands import format_status, format_help
from ._session import BotSessionManager
from ._debounce import InboundDebouncer
from ._ack import AckReactor
logger = logging.getLogger(__name__)
@@ -58,6 +59,7 @@ class DiscordBot(ChatCommandMixin, MessageHookMixin):
token: str,
agent: Optional["Agent"] = None,
config: Optional[BotConfig] = None,
**kwargs,
):
"""Initialize the Discord bot.
@@ -65,7 +67,10 @@ class DiscordBot(ChatCommandMixin, MessageHookMixin):
token: Discord bot token
agent: Optional agent to handle messages
config: Optional bot configuration
**kwargs: Additional arguments for forward compatibility
"""
# B9: Store extra kwargs for forward compatibility
self._extra_kwargs = kwargs
self._token = token
self._agent = agent
self.config = config or BotConfig(token=token)
@@ -81,6 +86,10 @@ class DiscordBot(ChatCommandMixin, MessageHookMixin):
self._debouncer: InboundDebouncer = InboundDebouncer(
debounce_ms=self.config.debounce_ms,
)
self._ack: AckReactor = AckReactor(
ack_emoji=self.config.ack_emoji,
done_emoji=self.config.done_emoji,
)
@property
def is_running(self) -> bool:
@@ -186,6 +195,22 @@ class DiscordBot(ChatCommandMixin, MessageHookMixin):
if self._agent:
user_id = str(message.author.id)
# Ack reaction - show processing indicator
ack_ctx = None
if self._ack.enabled:
async def _discord_react(emoji, **kw):
try:
await message.add_reaction(emoji)
except Exception:
pass # Reactions may not be supported
async def _discord_unreact(emoji, **kw):
try:
await message.remove_reaction(emoji, self._client.user)
except Exception:
pass
ack_ctx = await self._ack.ack(react_fn=_discord_react)
async def _send_agent_response():
text_to_send = await self._debouncer.debounce(user_id, bot_message.text)
response = await self._session.chat(self._agent, user_id, text_to_send)
@@ -196,6 +221,9 @@ class DiscordBot(ChatCommandMixin, MessageHookMixin):
return
await self._send_long_message(message.channel, send_result["content"], reference=message)
self.fire_message_sent(str(message.channel.id), send_result["content"])
# Done reaction - show completion
if ack_ctx:
await self._ack.done(ack_ctx, react_fn=_discord_react, unreact_fn=_discord_unreact)
if self.config.typing_indicator:
async with message.channel.typing():
+42 -1
View File
@@ -29,6 +29,7 @@ from .media import split_media_from_output, is_audio_file
from ._commands import format_status, format_help
from ._session import BotSessionManager
from ._debounce import InboundDebouncer
from ._ack import AckReactor
logger = logging.getLogger(__name__)
@@ -66,6 +67,7 @@ class SlackBot(ChatCommandMixin, MessageHookMixin):
signing_secret: Optional[str] = None,
agent: Optional["Agent"] = None,
config: Optional[BotConfig] = None,
**kwargs,
):
"""Initialize the Slack bot.
@@ -75,7 +77,10 @@ class SlackBot(ChatCommandMixin, MessageHookMixin):
signing_secret: Signing secret for webhook verification
agent: Optional agent to handle messages
config: Optional bot configuration
**kwargs: Additional arguments for forward compatibility
"""
# B9: Store extra kwargs for forward compatibility
self._extra_kwargs = kwargs
self._token = token
self._app_token = app_token
self._signing_secret = signing_secret
@@ -94,6 +99,10 @@ class SlackBot(ChatCommandMixin, MessageHookMixin):
self._debouncer: InboundDebouncer = InboundDebouncer(
debounce_ms=self.config.debounce_ms,
)
self._ack: AckReactor = AckReactor(
ack_emoji=self.config.ack_emoji,
done_emoji=self.config.done_emoji,
)
# Audio capabilities
self._stt_enabled: bool = False
@@ -194,6 +203,35 @@ class SlackBot(ChatCommandMixin, MessageHookMixin):
should_respond = True
if should_respond and self._agent:
channel_id = event.get("channel", "")
msg_ts = event.get("ts", "")
# Ack reaction - show processing indicator
ack_ctx = None
if self._ack.enabled and self._client:
async def _slack_react(emoji, **kw):
try:
# Slack uses emoji names without colons
emoji_name = emoji.strip(":")
await self._client.reactions_add(
channel=channel_id,
timestamp=msg_ts,
name=emoji_name,
)
except Exception:
pass # Reactions may fail silently
async def _slack_unreact(emoji, **kw):
try:
emoji_name = emoji.strip(":")
await self._client.reactions_remove(
channel=channel_id,
timestamp=msg_ts,
name=emoji_name,
)
except Exception:
pass
ack_ctx = await self._ack.ack(react_fn=_slack_react)
try:
user_id = event.get("user", "unknown")
logger.info(f"Message received: {text[:100]}...")
@@ -201,7 +239,6 @@ class SlackBot(ChatCommandMixin, MessageHookMixin):
response = await self._session.chat(self._agent, user_id, text)
logger.info(f"Response sent: {response[:100]}...")
channel_id = event.get("channel", "")
send_result = self.fire_message_sending(channel_id, str(response))
if send_result["cancel"]:
return
@@ -219,6 +256,10 @@ class SlackBot(ChatCommandMixin, MessageHookMixin):
channel_id, say, response, thread_ts=thread_ts
)
self.fire_message_sent(channel_id, response)
# Done reaction - show completion
if ack_ctx and self._client:
await self._ack.done(ack_ctx, react_fn=_slack_react, unreact_fn=_slack_unreact)
except Exception as e:
logger.error(f"Agent error: {e}")
error_msg = str(e)
+4
View File
@@ -62,6 +62,7 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
token: str,
agent: Optional["Agent"] = None,
config: Optional[BotConfig] = None,
**kwargs,
):
"""Initialize the Telegram bot.
@@ -69,7 +70,10 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
token: Telegram bot token from @BotFather
agent: Optional agent to handle messages
config: Optional bot configuration
**kwargs: Additional arguments for forward compatibility
"""
# B9: Store extra kwargs for forward compatibility
self._extra_kwargs = kwargs
self._token = token
self._agent = agent
self.config = config or BotConfig(token=token)
+60 -12
View File
@@ -45,6 +45,8 @@ from praisonaiagents.bots import (
from ._commands import format_status, format_help
from ._session import BotSessionManager
from ._rate_limit import RateLimiter
from ._ack import AckReactor
logger = logging.getLogger(__name__)
@@ -89,7 +91,10 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
allowed_numbers: Optional[List[str]] = None,
allowed_groups: Optional[List[str]] = None,
respond_to_all: bool = False,
**kwargs,
):
# B9: Store extra kwargs for forward compatibility
self._extra_kwargs = kwargs
# Mode: "cloud" (default, Meta Cloud API) or "web" (neonize, experimental)
self._mode = mode.lower().strip()
if self._mode not in ("cloud", "web"):
@@ -129,6 +134,13 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
self._message_handlers: List[Callable] = []
self._runner: Any = None
self._site: Any = None
self._http_session: Any = None # Shared aiohttp session for connection pooling
self._background_tasks: set = set() # Track background tasks to prevent silent failures
self._rate_limiter = RateLimiter.for_platform("whatsapp") # Rate limiting for API calls
self._ack: AckReactor = AckReactor(
ack_emoji=self.config.ack_emoji,
done_emoji=self.config.done_emoji,
)
# Web mode adapter (lazy initialized)
self._web_adapter: Any = None
@@ -216,6 +228,7 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
async def _start_cloud_mode(self) -> None:
"""Start the Cloud API webhook server."""
try:
import aiohttp
from aiohttp import web
except ImportError:
raise ImportError("aiohttp is required for WhatsApp bot. Install with: pip install aiohttp")
@@ -231,6 +244,7 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
self._is_running = True
self._started_at = time.time()
self._http_session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30))
self._bot_user = BotUser(
user_id=self._phone_number_id,
username="whatsapp_bot",
@@ -257,6 +271,7 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
⚠️ EXPERIMENTAL: Uses reverse-engineered protocol.
"""
import aiohttp
from ._whatsapp_web_adapter import WhatsAppWebAdapter
async def _on_web_message(event: Any) -> None:
@@ -447,6 +462,7 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
self._is_running = True
self._started_at = time.time()
self._http_session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30))
self._bot_user = BotUser(
user_id="web_user",
username="whatsapp_web_bot",
@@ -507,6 +523,15 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
await self._web_adapter.disconnect()
except Exception as e:
logger.warning(f"Web adapter disconnect error: {e}")
# Cancel pending background tasks
for task in self._background_tasks:
task.cancel()
if self._background_tasks:
await asyncio.gather(*self._background_tasks, return_exceptions=True)
self._background_tasks.clear()
if self._http_session:
await self._http_session.close()
self._http_session = None
logger.info("WhatsApp bot stopped")
# ── Webhook handlers ────────────────────────────────────────────
@@ -561,7 +586,10 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
return web.Response(status=400, text="Bad request")
# Process asynchronously so we respond 200 immediately
asyncio.create_task(self._process_webhook_data(data))
# Track task to prevent silent failures
task = asyncio.create_task(self._process_webhook_data(data))
self._background_tasks.add(task)
task.add_done_callback(self._background_tasks.discard)
# Always respond 200 to acknowledge receipt
return web.Response(status=200, text="OK")
@@ -681,6 +709,22 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
# Route to agent
if self._agent and content:
# Ack reaction - show processing indicator
ack_ctx = None
if self._ack.enabled:
async def _wa_react(emoji, **kw):
try:
await self.send_reaction(sender_id, msg_id, emoji)
except Exception:
pass
async def _wa_unreact(emoji, **kw):
try:
# WhatsApp doesn't have unreact, send empty reaction to clear
await self.send_reaction(sender_id, msg_id, "")
except Exception:
pass
ack_ctx = await self._ack.ack(react_fn=_wa_react)
try:
response = await self._session_mgr.chat(
self._agent, sender_id, content
@@ -690,6 +734,9 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
if not send_result["cancel"]:
await self.send_message(sender_id, send_result["content"])
self.fire_message_sent(sender_id, send_result["content"])
# Done reaction - show completion
if ack_ctx:
await self._ack.done(ack_ctx, react_fn=_wa_react, unreact_fn=_wa_unreact)
except Exception as e:
logger.error(f"Agent chat error: {e}")
await self.send_message(sender_id, "Sorry, I encountered an error processing your message.")
@@ -733,9 +780,11 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
if reply_to:
payload["context"] = {"message_id": reply_to}
# Rate limit before sending
await self._rate_limiter.acquire(to)
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
async with self._http_session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
result = await resp.json()
if "error" in result:
logger.error(f"WhatsApp send error: {result['error']}")
@@ -783,9 +832,11 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
if components:
payload["template"]["components"] = components
# Rate limit before sending
await self._rate_limiter.acquire(to)
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
async with self._http_session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
result = await resp.json()
if "error" in result:
logger.error(f"WhatsApp template error: {result['error']}")
@@ -820,8 +871,7 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
async with self._http_session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
await resp.json()
except Exception as e:
logger.error(f"Failed to send reaction: {e}")
@@ -845,8 +895,7 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
async with self._http_session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=10)) as resp:
await resp.json()
except Exception as e:
logger.error(f"Failed to mark as read: {e}")
@@ -883,8 +932,7 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
import aiohttp
url = f"{GRAPH_API_BASE}/{self._phone_number_id}"
headers = {"Authorization": f"Bearer {self._token}"}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=10)) as resp:
async with self._http_session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=10)) as resp:
elapsed = (time.time() - started) * 1000
if resp.status == 200:
data = await resp.json()
@@ -904,7 +952,7 @@ class WhatsAppBot(ChatCommandMixin, MessageHookMixin):
from praisonaiagents.bots import HealthResult
probe_result = await self.probe()
uptime = (time.time() - self._started_at) if self._started_at else None
session_count = len(self._session._histories) if hasattr(self._session, '_histories') else 0
session_count = len(self._session_mgr._histories) if hasattr(self._session_mgr, '_histories') else 0
return HealthResult(
ok=self._is_running and probe_result.ok, platform="whatsapp",
is_running=self._is_running, uptime_seconds=uptime,
+4
View File
@@ -184,6 +184,7 @@ def register_commands():
from .commands.run import app as run_app
from .commands.profile import app as profile_app
from .commands.benchmark import app as benchmark_app
from .commands.paths import app as paths_app
# Import new command modules - Previously legacy-only commands
from .commands.chat import app as chat_app
@@ -209,6 +210,7 @@ def register_commands():
from .commands.todo import app as todo_app
from .commands.docs import app as docs_app
from .commands.commit import app as commit_app
from .commands.publish import app as publish_app
from .commands.hooks import app as hooks_app
from .commands.rules import app as rules_app
from .commands.registry import app as registry_app
@@ -251,6 +253,7 @@ def register_commands():
app.add_typer(run_app, name="run", help="Run agents")
app.add_typer(profile_app, name="profile", help="Performance profiling and diagnostics")
app.add_typer(benchmark_app, name="benchmark", help="Comprehensive performance benchmarking")
app.add_typer(paths_app, name="paths", help="Storage path inspection and migration")
# Register sub-apps - Terminal-native commands
app.add_typer(chat_app, name="chat", help="Terminal-native interactive chat (REPL)")
@@ -278,6 +281,7 @@ def register_commands():
app.add_typer(todo_app, name="todo", help="Todo/task management")
app.add_typer(docs_app, name="docs", help="Documentation management")
app.add_typer(commit_app, name="commit", help="AI-assisted git commits")
app.add_typer(publish_app, name="publish", help="Package publishing")
app.add_typer(hooks_app, name="hooks", help="Hook management")
app.add_typer(rules_app, name="rules", help="Rules management")
app.add_typer(registry_app, name="registry", help="Registry management")
@@ -0,0 +1,106 @@
"""
Paths command group for PraisonAI CLI.
Provides path inspection:
- paths show: Display all resolved storage paths
"""
import os
from pathlib import Path
import typer
from ..output.console import get_output_controller
app = typer.Typer(help="Storage path inspection and migration")
@app.command("show")
def paths_show(
json_output: bool = typer.Option(
False,
"--json",
help="Output in JSON format",
),
):
"""
Show all resolved storage paths.
Displays global, project-local, and environment-overridden paths.
"""
output = get_output_controller()
try:
from praisonaiagents.paths import (
ENV_VAR,
DEFAULT_DIR_NAME,
LEGACY_DIR_NAME,
get_data_dir,
get_all_paths,
get_project_data_dir,
)
except ImportError:
output.print_error("praisonaiagents package not installed")
raise typer.Exit(1)
env_home = os.environ.get(ENV_VAR)
global_dir = get_data_dir()
project_dir = get_project_data_dir()
legacy_dir = Path.home() / LEGACY_DIR_NAME
all_paths = get_all_paths()
if json_output or (hasattr(output, "is_json_mode") and output.is_json_mode):
data = {
"env_var": ENV_VAR,
"env_value": env_home,
"global_dir": str(global_dir),
"project_dir": str(project_dir),
"legacy_dir_exists": legacy_dir.exists(),
"paths": {k: str(v) for k, v in all_paths.items()},
}
output.print_json(data)
return
# Text output using Rich console directly
from rich.console import Console
from rich.table import Table
console = Console()
console.print("\n[bold cyan]PraisonAI Storage Paths[/bold cyan]\n")
# Environment override
if env_home:
console.print(f" [green]✓[/green] {ENV_VAR} = {env_home}")
else:
console.print(f" [dim]{ENV_VAR} is not set (using defaults)[/dim]")
console.print()
# Table of paths
table = Table(title="Resolved Paths", show_lines=False)
table.add_column("Category", style="cyan")
table.add_column("Path", style="white")
table.add_column("Exists", justify="center")
table.add_row(
"Global data dir",
str(global_dir),
"✅" if global_dir.exists() else "—",
)
table.add_row(
"Project data dir",
str(project_dir),
"✅" if project_dir.exists() else "—",
)
for name, path_val in sorted(all_paths.items()):
p = Path(str(path_val))
table.add_row(name, str(p), "✅" if p.exists() else "—")
console.print(table)
# Legacy warning
if legacy_dir.exists():
console.print(
f"\n [yellow]⚠ Legacy directory {legacy_dir} exists.[/yellow]"
)
@@ -0,0 +1,193 @@
"""
Publish command group for PraisonAI CLI.
Provides package publishing with automatic version bumping.
"""
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path
import typer
from rich import print as rprint
app = typer.Typer(help="Package publishing")
def _find_pyproject() -> Path:
"""Find pyproject.toml in current directory."""
p = Path.cwd() / "pyproject.toml"
if not p.exists():
rprint("[red]Error: No pyproject.toml found in current directory[/red]")
raise typer.Exit(1)
return p
def _read_version(pyproject: Path) -> str:
"""Read version from pyproject.toml."""
content = pyproject.read_text()
match = re.search(r'^version\s*=\s*"([^"]+)"', content, re.MULTILINE)
if not match:
rprint("[red]Error: Could not find version in pyproject.toml[/red]")
raise typer.Exit(1)
return match.group(1)
def _bump_version(version: str, major: bool = False, minor: bool = False) -> str:
"""Bump version string.
Default bumps patch (last number): 1.5.21 -> 1.5.22
--minor: 1.5.21 -> 1.6.0
--major: 1.5.21 -> 2.0.0
"""
parts = version.split(".")
if len(parts) != 3:
rprint(f"[red]Error: Version '{version}' is not in X.Y.Z format[/red]")
raise typer.Exit(1)
maj, min_, patch = int(parts[0]), int(parts[1]), int(parts[2])
if major:
return f"{maj + 1}.0.0"
elif minor:
return f"{maj}.{min_ + 1}.0"
else:
return f"{maj}.{min_}.{patch + 1}"
def _write_version(pyproject: Path, old_version: str, new_version: str):
"""Write new version to pyproject.toml."""
content = pyproject.read_text()
content = content.replace(f'version = "{old_version}"', f'version = "{new_version}"', 1)
pyproject.write_text(content)
def _run_step(cmd: list[str], label: str, dry_run: bool = False) -> bool:
"""Run a build step, printing status."""
cmd_str = " ".join(cmd)
if dry_run:
rprint(f" [dim]→ {cmd_str}[/dim]")
return True
rprint(f"[cyan]→ {cmd_str}[/cyan]")
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
rprint(f"[red]✗ {label} failed[/red]")
if result.stderr:
rprint(f"[red]{result.stderr.strip()}[/red]")
if result.stdout:
rprint(f"[dim]{result.stdout.strip()}[/dim]")
return False
if result.stdout:
# Show concise output
lines = result.stdout.strip().split("\n")
for line in lines[-5:]: # last 5 lines
rprint(f" [dim]{line}[/dim]")
rprint(f"[green]✓ {label}[/green]")
return True
@app.callback(invoke_without_command=True)
def publish_main(ctx: typer.Context):
"""
Publish packages to registries.
Use a subcommand like 'pypi' to specify the target.
Examples:
praisonai publish pypi
praisonai publish pypi --minor
praisonai publish pypi --dry-run
"""
if ctx.invoked_subcommand is None:
rprint("[yellow]Usage: praisonai publish pypi [OPTIONS][/yellow]")
rprint("Run [bold]praisonai publish pypi --help[/bold] for options.")
raise typer.Exit(0)
@app.command("pypi")
def publish_pypi(
major: bool = typer.Option(False, "--major", help="Bump major version (X.0.0)"),
minor: bool = typer.Option(False, "--minor", help="Bump minor version (X.Y.0)"),
no_bump: bool = typer.Option(False, "--no-bump", help="Skip version increment"),
dry_run: bool = typer.Option(False, "--dry-run", "-n", help="Show what would happen without executing"),
token_var: str = typer.Option("PYPI_TOKEN", "--token-var", help="Environment variable name for PyPI token"),
):
"""
Publish package to PyPI.
Auto-increments the patch version (last number) by default,
then runs: rm -rf dist && uv lock && uv build && uv publish
Examples:
praisonai publish pypi # bump patch: 1.5.21 -> 1.5.22
praisonai publish pypi --minor # bump minor: 1.5.21 -> 1.6.0
praisonai publish pypi --major # bump major: 1.5.21 -> 2.0.0
praisonai publish pypi --no-bump # publish without version change
praisonai publish pypi --dry-run # preview without executing
praisonai publish pypi --token-var UV_PUBLISH_TOKEN
"""
# 1. Find pyproject.toml
pyproject = _find_pyproject()
old_version = _read_version(pyproject)
# 2. Bump version
if no_bump:
new_version = old_version
rprint(f"[yellow]Skipping version bump — publishing {old_version}[/yellow]")
else:
if major and minor:
rprint("[red]Error: Cannot use --major and --minor together[/red]")
raise typer.Exit(1)
new_version = _bump_version(old_version, major=major, minor=minor)
bump_type = "major" if major else ("minor" if minor else "patch")
rprint(f"[cyan]Version bump ({bump_type}): {old_version} → {new_version}[/cyan]")
# 3. Get PyPI token
token = os.environ.get(token_var)
if not token:
rprint(f"[red]Error: {token_var} environment variable not set[/red]")
rprint(f"[dim]Set it with: export {token_var}=pypi-...[/dim]")
raise typer.Exit(1)
rprint(f"[green]✓ PyPI token found ({token_var})[/green]")
# 4. Dry run summary
if dry_run:
rprint("\n[yellow]Dry run — no changes will be made:[/yellow]")
if not no_bump:
rprint(f" [dim]→ Update pyproject.toml: {old_version} → {new_version}[/dim]")
_run_step(["rm", "-rf", "dist"], "clean", dry_run=True)
_run_step(["uv", "lock"], "lock", dry_run=True)
_run_step(["uv", "build"], "build", dry_run=True)
_run_step(["uv", "publish", "--token", f"${token_var}"], "publish", dry_run=True)
rprint("[green]Dry run complete.[/green]")
return
# 5. Write new version
if not no_bump:
_write_version(pyproject, old_version, new_version)
rprint(f"[green]✓ Updated pyproject.toml[/green]")
# 6. Clean dist
dist_dir = Path.cwd() / "dist"
if dist_dir.exists():
shutil.rmtree(dist_dir)
rprint("[green]✓ Cleaned dist/[/green]")
# 7. uv lock
if not _run_step(["uv", "lock"], "uv lock"):
raise typer.Exit(1)
# 8. uv build
if not _run_step(["uv", "build"], "uv build"):
raise typer.Exit(1)
# 9. uv publish
if not _run_step(["uv", "publish", "--token", token], "uv publish"):
raise typer.Exit(1)
rprint(f"\n[bold green]✓ Published {new_version} to PyPI[/bold green]")
@@ -15,24 +15,30 @@ def workflow_run(
file: str = typer.Argument(..., help="Workflow file path"),
model: str = typer.Option(None, "--model", "-m", help="LLM model to use"),
verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
dry_run: bool = typer.Option(False, "--dry-run", help="Preview steps without executing (job workflows only)"),
):
"""Run a workflow.
Use --var key=value to pass variables to YAML workflows.
Use --dry-run to preview job workflows without executing.
Can be specified multiple times.
Example: praisonai workflow run my_workflow.yaml --var topic="AI Tools" --var style="comprehensive"
Example:
praisonai workflow run my_workflow.yaml --var topic="AI Tools"
praisonai workflow run publish-pypi.yaml --dry-run
"""
from praisonai.cli.main import PraisonAI
import sys
argv = ['workflow', 'run', file]
# Pass through extra args (includes --var arguments)
# Pass through extra args (includes --var arguments and custom flags)
argv.extend(ctx.args)
if model:
argv.extend(['--model', model])
if verbose:
argv.append('--verbose')
if dry_run:
argv.append('--dry-run')
original_argv = sys.argv
sys.argv = ['praisonai'] + argv
@@ -58,7 +58,6 @@ def _execute_core(
agent_config = {
"name": request.agent_name,
"verbose": False,
}
if request.agent_instructions:
@@ -183,7 +182,6 @@ async def execute_async_stream(
agent_config = {
"name": request.agent_name,
"verbose": False,
}
if request.agent_instructions:
@@ -293,11 +293,18 @@ def create_agent_centric_tools(
JSON string with list of symbols and their locations
"""
async def _list():
result = await router.handle_query(
f"list all functions and classes in {file_path}",
file_path=file_path
)
return json.dumps(result.to_dict())
# Add 10s timeout to prevent LSP from blocking autonomy
try:
result = await asyncio.wait_for(
router.handle_query(
f"list all functions and classes in {file_path}",
file_path=file_path
),
timeout=10.0
)
return json.dumps(result.to_dict())
except asyncio.TimeoutError:
return json.dumps({"error": "LSP list_symbols timed out (10s)", "success": False})
return run_async(_list())
@@ -317,8 +324,15 @@ def create_agent_centric_tools(
if file_path:
query += f" in {file_path}"
result = await router.handle_query(query, file_path=file_path)
return json.dumps(result.to_dict())
# Add 10s timeout to prevent LSP from blocking autonomy
try:
result = await asyncio.wait_for(
router.handle_query(query, file_path=file_path),
timeout=10.0
)
return json.dumps(result.to_dict())
except asyncio.TimeoutError:
return json.dumps({"error": "LSP find_definition timed out (10s)", "success": False})
return run_async(_find())
@@ -338,8 +352,15 @@ def create_agent_centric_tools(
if file_path:
query += f" in {file_path}"
result = await router.handle_query(query, file_path=file_path)
return json.dumps(result.to_dict())
# Add 10s timeout to prevent LSP from blocking autonomy
try:
result = await asyncio.wait_for(
router.handle_query(query, file_path=file_path),
timeout=10.0
)
return json.dumps(result.to_dict())
except asyncio.TimeoutError:
return json.dumps({"error": "LSP find_references timed out (10s)", "success": False})
return run_async(_find())
@@ -358,8 +379,15 @@ def create_agent_centric_tools(
if file_path:
query += f" in {file_path}"
result = await router.handle_query(query, file_path=file_path)
return json.dumps(result.to_dict())
# Add 10s timeout to prevent LSP from blocking autonomy
try:
result = await asyncio.wait_for(
router.handle_query(query, file_path=file_path),
timeout=10.0
)
return json.dumps(result.to_dict())
except asyncio.TimeoutError:
return json.dumps({"error": "LSP diagnostics timed out (10s)", "success": False})
return run_async(_get())
@@ -0,0 +1,173 @@
"""
JSONL Display Mode for PraisonAI CLI.
Provides structured JSONL output matching Gemini CLI's STREAM_JSON format.
Each event is a single JSON line written to stderr.
Usage:
praisonai "prompt" --display jsonl 2>/tmp/events.jsonl
Event Types:
- init: Session start with model info
- tool_use: Tool call with parameters
- tool_result: Tool completion with status/output
- llm_start: LLM call start
- llm_end: LLM call end with metrics
- autonomy_step: Autonomy iteration info (PraisonAI unique)
- error: Error event
- result: Final result with stats
"""
import json
import sys
import time
from datetime import datetime, timezone
from typing import Any, Dict, Optional, TextIO
class JsonlDisplay:
"""JSONL event emitter matching Gemini CLI's STREAM_JSON format.
Thread-safe for multi-agent execution.
"""
def __init__(self, file: TextIO = None):
self.file = file or sys.stderr
self._start_time = time.time()
self._tool_calls = 0
self._tokens_in = 0
self._tokens_out = 0
def emit(self, event_type: str, **kwargs) -> None:
"""Emit a JSONL event. Thread-safe."""
event = {
"type": event_type,
"ts": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
}
# Filter None values
event.update({k: v for k, v in kwargs.items() if v is not None})
try:
self.file.write(json.dumps(event, default=str) + "\n")
self.file.flush()
except Exception:
pass # Silently ignore write errors
def on_init(self, model: str = None, session_id: str = None) -> None:
"""Emit init event at session start."""
self.emit("init", model=model, session_id=session_id)
def on_tool_call(
self,
message: str = None,
tool_name: str = None,
tool_input: Dict[str, Any] = None,
tool_output: str = None,
**kwargs
) -> None:
"""Callback for tool calls - emits tool_use or tool_result."""
if tool_name:
if tool_output is None:
# Tool starting
self._tool_calls += 1
self.emit("tool_use", tool=tool_name, args=tool_input)
else:
# Tool completed
output_str = str(tool_output)[:500] if tool_output else None
self.emit("tool_result", tool=tool_name, status="ok", output=output_str)
def on_llm_start(self, model: str = None, agent_name: str = None, **kwargs) -> None:
"""Callback for LLM call start."""
self.emit("llm_start", model=model, agent=agent_name)
def on_llm_end(
self,
model: str = None,
tokens_in: int = 0,
tokens_out: int = 0,
cost: float = None,
latency_ms: float = None,
**kwargs
) -> None:
"""Callback for LLM call end with metrics."""
self._tokens_in += tokens_in
self._tokens_out += tokens_out
self.emit(
"llm_end",
model=model,
tokens_in=tokens_in,
tokens_out=tokens_out,
cost=cost,
latency_ms=latency_ms
)
def on_error(self, message: str = None, **kwargs) -> None:
"""Callback for errors."""
if message:
self.emit("error", severity="error", message=str(message))
def on_interaction(self, response: str = None, agent_name: str = None, **kwargs) -> None:
"""Callback for agent response."""
if response:
# Truncate for JSONL
self.emit("message", role="assistant", content=response[:2000], agent=agent_name)
# P3/G2: Autonomy-specific events (PraisonAI unique!)
def on_autonomy_iteration(
self,
iteration: int = None,
max_iterations: int = None,
stage: str = None,
**kwargs
) -> None:
"""Callback for autonomy iteration."""
self.emit("autonomy_step", iteration=iteration, max=max_iterations, stage=stage)
def on_autonomy_stage_change(
self,
from_stage: str = None,
to_stage: str = None,
**kwargs
) -> None:
"""Callback for autonomy stage escalation."""
self.emit("autonomy_stage", from_stage=from_stage, to_stage=to_stage)
def on_autonomy_doom_loop(
self,
iteration: int = None,
recovery_action: str = None,
**kwargs
) -> None:
"""Callback for doom loop detection."""
self.emit("autonomy_doom_loop", iteration=iteration, recovery=recovery_action)
def on_complete(
self,
reason: str = None,
iterations: int = None,
duration_ms: float = None,
**kwargs
) -> None:
"""Emit final result event with stats."""
duration = duration_ms or (time.time() - self._start_time) * 1000
self.emit(
"result",
status="success" if reason in ("goal", "promise") else "stopped",
reason=reason,
iterations=iterations,
duration_ms=duration,
stats={
"total_tokens": self._tokens_in + self._tokens_out,
"input_tokens": self._tokens_in,
"output_tokens": self._tokens_out,
"tool_calls": self._tool_calls,
}
)
def create_jsonl_display(file: TextIO = None) -> JsonlDisplay:
"""Create a JSONL display instance."""
return JsonlDisplay(file=file)
__all__ = ["JsonlDisplay", "create_jsonl_display"]
@@ -0,0 +1,324 @@
"""
Hybrid Workflow Executor for PraisonAI CLI.
Executes workflows with `type: hybrid` that combine:
- Deterministic steps (run, python, script, action) from JobWorkflowExecutor
- Agent-centric steps (agent, judge, approve) from JobWorkflowExecutor
- Multi-agent workflow steps (using praisonaiagents)
This provides a unified executor that can handle both paradigms in a single workflow.
Usage:
praisonai workflow run hybrid-workflow.yaml
praisonai workflow run hybrid-workflow.yaml --dry-run
"""
import os
import time
from pathlib import Path
from typing import Any, Dict, List, Optional
class HybridWorkflowExecutor:
"""
Execute a `type: hybrid` workflow — combines job and agent workflow capabilities.
Step types (detected by key):
From JobWorkflowExecutor (deterministic + agent-centric):
run: Shell command via subprocess
python: Python script file
script: Inline Python code
action: Built-in or YAML-defined action
agent: Single AI agent execution
judge: AI quality gate with threshold
approve: AI or human approval gate
From Agent Workflows (multi-agent):
workflow: Execute a multi-agent workflow step
parallel: Execute multiple steps in parallel
"""
def __init__(self, data: Dict[str, Any], file_path: str):
self._data = data
self._file_path = file_path
self._name = data.get("name", Path(file_path).stem)
self._description = data.get("description", "")
self._steps = data.get("steps", [])
self._agents = data.get("agents", {})
self._vars = data.get("vars", {})
self._flag_defs = data.get("flags", {})
self._cwd = str(Path(file_path).parent.resolve())
# Lazy-loaded executors
self._job_executor = None
def _get_job_executor(self):
"""Lazy-load JobWorkflowExecutor for deterministic/agent-centric steps."""
if self._job_executor is None:
from .job_workflow import JobWorkflowExecutor
self._job_executor = JobWorkflowExecutor(self._data, self._file_path)
return self._job_executor
def run(self, args: List[str]) -> Optional[Dict[str, Any]]:
"""
Execute the hybrid workflow.
Args:
args: Raw CLI args (may contain --dry-run, etc.)
Returns:
Result dict with status and step outcomes.
"""
from rich.console import Console
from rich.panel import Panel
console = Console()
dry_run = "--dry-run" in args
flags = self._parse_flags(args)
# Header
mode = "[yellow]DRY RUN[/yellow]" if dry_run else "[green]EXECUTE[/green]"
console.print(Panel(
f"[bold]{self._name}[/bold]\n{self._description}",
title=f"🔀 Hybrid Workflow — {mode}",
border_style="magenta",
))
if flags:
flag_str = ", ".join(f"--{k}" for k, v in flags.items() if v)
console.print(f" [dim]Flags:[/dim] {flag_str}")
console.print()
results: List[Dict[str, Any]] = []
failed = False
for i, step in enumerate(self._steps, 1):
step_name = step.get("name", f"Step {i}")
# Conditional: if
condition = step.get("if")
if condition and not self._eval_condition(condition, flags):
console.print(f" [dim]⊘ {step_name} — skipped (condition false)[/dim]")
results.append({"name": step_name, "status": "skipped"})
continue
# Detect step type and route to appropriate executor
step_type = self._detect_step_type(step)
if dry_run:
display = self._format_dry_run_display(step_type, step)
console.print(f" [cyan]● {step_name}[/cyan] — [dim]{display}[/dim]")
results.append({"name": step_name, "status": "dry-run", "type": step_type})
continue
# Execute
console.print(f" [bold]▸ {step_name}[/bold] ", end="")
t0 = time.time()
try:
if step_type in ("shell", "python", "script", "action", "agent", "judge", "approve"):
# Route to JobWorkflowExecutor
result = self._execute_job_step(step_type, step, flags)
elif step_type == "workflow":
# Execute multi-agent workflow step
result = self._execute_workflow_step(step, flags)
elif step_type == "parallel":
# Execute parallel steps
result = self._execute_parallel_step(step, flags)
else:
result = {"ok": False, "error": f"Unknown step type: {step_type}"}
except Exception as e:
result = {"ok": False, "error": str(e)}
elapsed = time.time() - t0
if result["ok"]:
console.print(f"[green]✓[/green] [dim]({elapsed:.1f}s)[/dim]")
else:
console.print(f"[red]✗[/red] [dim]({elapsed:.1f}s)[/dim]")
if result.get("error"):
console.print(f" [red]{result['error']}[/red]")
failed = True
results.append({
"name": step_name,
"status": "ok" if result["ok"] else "error",
"type": step_type,
"elapsed": elapsed,
**({k: v for k, v in result.items() if k not in ("ok",)}),
})
# Stop on failure unless continue_on_error
if failed and not step.get("continue_on_error", False):
break
# Summary
console.print()
if dry_run:
console.print(f"[yellow]🔀 Dry run complete — {len(results)} steps planned[/yellow]")
elif failed:
console.print("[red]✗ Hybrid workflow failed[/red]")
else:
console.print(f"[green]✓ Hybrid workflow completed — {len(results)} steps[/green]")
return {"name": self._name, "results": results, "ok": not failed, "dry_run": dry_run}
def _detect_step_type(self, step: Dict) -> str:
"""Detect step type from keys."""
# Job workflow step types
if "run" in step:
return "shell"
if "python" in step:
return "python"
if "script" in step:
return "script"
if "action" in step:
return "action"
if "agent" in step:
return "agent"
if "judge" in step:
return "judge"
if "approve" in step:
return "approve"
# Hybrid-specific step types
if "workflow" in step:
return "workflow"
if "parallel" in step:
return "parallel"
return "unknown"
def _format_dry_run_display(self, step_type: str, step: Dict) -> str:
"""Format dry-run display based on step type."""
if step_type == "agent":
agent_info = step.get("agent", {})
if isinstance(agent_info, dict):
role = agent_info.get("role", "Assistant")
model = agent_info.get("model", "gpt-4o-mini")
return f"agent: {role} (model: {model})"
return f"agent: {agent_info}"
elif step_type == "judge":
judge_info = step.get("judge", {})
threshold = judge_info.get("threshold", 7.0) if isinstance(judge_info, dict) else 7.0
return f"judge: threshold={threshold}"
elif step_type == "approve":
approve_info = step.get("approve", {})
risk = approve_info.get("risk_level", "medium") if isinstance(approve_info, dict) else "medium"
return f"approve: risk={risk}"
elif step_type == "workflow":
workflow_info = step.get("workflow", {})
agent_ref = workflow_info.get("agent", "unknown") if isinstance(workflow_info, dict) else workflow_info
return f"workflow: agent={agent_ref}"
elif step_type == "parallel":
parallel_steps = step.get("parallel", [])
return f"parallel: {len(parallel_steps)} steps"
elif step_type == "shell":
cmd = step.get("run", "")
return f"shell: {cmd[:60]}..." if len(cmd) > 60 else f"shell: {cmd}"
else:
return f"{step_type}"
def _execute_job_step(self, step_type: str, step: Dict, flags: Dict) -> Dict:
"""Execute a step using JobWorkflowExecutor."""
executor = self._get_job_executor()
# Get the target based on step type
type_to_key = {
"shell": "run",
"python": "python",
"script": "script",
"action": "action",
"agent": "agent",
"judge": "judge",
"approve": "approve",
}
key = type_to_key.get(step_type)
target = step.get(key)
return executor._execute_step(step_type, target, step, flags)
def _execute_workflow_step(self, step: Dict, flags: Dict) -> Dict:
"""Execute a multi-agent workflow step."""
workflow_config = step.get("workflow", {})
# Get agent reference
agent_ref = workflow_config.get("agent") if isinstance(workflow_config, dict) else workflow_config
if not agent_ref:
return {"ok": False, "error": "Workflow step requires 'agent' reference"}
# Look up agent in agents: block
agent_config = self._agents.get(agent_ref)
if not agent_config:
return {"ok": False, "error": f"Agent not found: {agent_ref}"}
try:
from praisonaiagents import Agent
agent = Agent(
name=agent_config.get("name", agent_ref),
role=agent_config.get("role", "Assistant"),
instructions=agent_config.get("instructions", ""),
llm=agent_config.get("model", "gpt-4o-mini"),
)
action = workflow_config.get("action", agent_config.get("goal", ""))
result = agent.chat(action)
return {"ok": True, "output": str(result) if result else ""}
except ImportError:
return {"ok": False, "error": "praisonaiagents not installed"}
except Exception as e:
return {"ok": False, "error": str(e)}
def _execute_parallel_step(self, step: Dict, flags: Dict) -> Dict:
"""Execute multiple steps in parallel."""
parallel_steps = step.get("parallel", [])
if not parallel_steps:
return {"ok": True, "output": "No parallel steps to execute"}
# For now, execute sequentially (true parallelism would require asyncio)
# This maintains compatibility while providing the parallel syntax
results = []
all_ok = True
for sub_step in parallel_steps:
step_type = self._detect_step_type(sub_step)
if step_type in ("shell", "python", "script", "action", "agent", "judge", "approve"):
result = self._execute_job_step(step_type, sub_step, flags)
elif step_type == "workflow":
result = self._execute_workflow_step(sub_step, flags)
else:
result = {"ok": False, "error": f"Unknown step type in parallel: {step_type}"}
results.append(result)
if not result.get("ok"):
all_ok = False
return {
"ok": all_ok,
"results": results,
"output": f"Executed {len(parallel_steps)} parallel steps",
}
def _parse_flags(self, args: List[str]) -> Dict[str, bool]:
"""Parse CLI flags based on workflow flag definitions."""
flags = {}
for flag_name in self._flag_defs:
cli_flag = f"--{flag_name}"
flags[flag_name.replace("-", "_")] = cli_flag in args
return flags
def _eval_condition(self, condition: str, flags: Dict) -> bool:
"""Evaluate a simple condition expression."""
condition = condition.strip()
if condition.startswith("{{") and condition.endswith("}}"):
condition = condition[2:-2].strip()
context = {"flags": type("Flags", (), flags)(), "env": os.environ}
try:
return bool(eval(condition, {"__builtins__": {}}, context))
except Exception:
return False
@@ -0,0 +1,745 @@
"""
Job Workflow Executor for PraisonAI CLI.
General-purpose step executor for workflows with `type: job`.
Supports multiple step types:
**Deterministic steps** (no LLM):
- run: Shell command via subprocess
- python: Python script file
- script: Inline Python code
- action: Built-in or YAML-defined action
**Agent-centric steps** (AI-powered):
- agent: Single AI agent execution
- judge: AI quality gate with threshold
- approve: AI or human approval gate
Usage:
praisonai workflow run publish-pypi.yaml
praisonai workflow run publish-pypi.yaml --dry-run
praisonai workflow run publish-pypi.yaml --major
"""
import os
import re
import subprocess
import sys
import time
from pathlib import Path
from typing import Any, Dict, List, Optional
class JobWorkflowExecutor:
"""
Execute a `type: job` workflow — ordered steps with optional AI integration.
Step types (detected by key):
Deterministic (no LLM):
run: Shell command via subprocess
python: Python script file
script: Inline Python code
action: Built-in or YAML-defined action
Agent-centric (AI-powered):
agent: Single AI agent execution
judge: AI quality gate with threshold
approve: AI or human approval gate
"""
# Built-in actions registry
BUILT_IN_ACTIONS = {"bump-version"}
def __init__(self, data: Dict[str, Any], file_path: str):
self._data = data
self._file_path = file_path
self._name = data.get("name", Path(file_path).stem)
self._description = data.get("description", "")
self._steps = data.get("steps", [])
self._vars = data.get("vars", {})
self._flag_defs = data.get("flags", {})
self._cwd = str(Path(file_path).parent.resolve())
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def run(self, args: List[str]) -> Optional[Dict[str, Any]]:
"""
Execute the workflow.
Args:
args: Raw CLI args (may contain --dry-run, --major, etc.)
Returns:
Result dict with status and step outcomes.
"""
from rich.console import Console
from rich.panel import Panel
console = Console()
dry_run = "--dry-run" in args
flags = self._parse_flags(args)
# Header
mode = "[yellow]DRY RUN[/yellow]" if dry_run else "[green]EXECUTE[/green]"
console.print(Panel(
f"[bold]{self._name}[/bold]\n{self._description}",
title=f"⚡ Job Workflow — {mode}",
border_style="cyan",
))
if flags:
flag_str = ", ".join(f"--{k}" for k, v in flags.items() if v)
console.print(f" [dim]Flags:[/dim] {flag_str}")
console.print()
results: List[Dict[str, Any]] = []
failed = False
for i, step in enumerate(self._steps, 1):
step_name = step.get("name", f"Step {i}")
# Conditional: if
condition = step.get("if")
if condition and not self._eval_condition(condition, flags):
console.print(f" [dim]⊘ {step_name} — skipped (condition false)[/dim]")
results.append({"name": step_name, "status": "skipped"})
continue
# Detect step type
step_type, step_target = self._detect_step_type(step)
if not step_type:
console.print(f" [red]✗ {step_name}[/red] — unknown step type (need run/python/script/action/agent/judge/approve)")
results.append({"name": step_name, "status": "error", "error": "unknown step type"})
failed = True
break
# Resolve variables in the target (only for string targets)
if isinstance(step_target, str):
step_target = self._resolve_vars(step_target, flags)
if dry_run:
# Format dry-run display based on step type
if step_type == "agent":
agent_info = step_target if isinstance(step_target, dict) else {"instructions": step_target}
role = agent_info.get("role", "Assistant")
model = agent_info.get("model", "gpt-4o-mini")
display = f"agent: {role} (model: {model})"
elif step_type == "judge":
judge_info = step_target if isinstance(step_target, dict) else {"criteria": step_target}
threshold = judge_info.get("threshold", 7.0)
display = f"judge: threshold={threshold}"
elif step_type == "approve":
approve_info = step_target if isinstance(step_target, dict) else {"description": step_target}
risk = approve_info.get("risk_level", "medium")
display = f"approve: risk={risk}"
else:
display = f"{step_type}: {self._truncate(str(step_target), 80)}"
console.print(f" [cyan]● {step_name}[/cyan] — [dim]{display}[/dim]")
results.append({"name": step_name, "status": "dry-run", "type": step_type})
continue
# Execute
console.print(f" [bold]▸ {step_name}[/bold] ", end="")
t0 = time.time()
result = self._execute_step(step_type, step_target, step, flags)
elapsed = time.time() - t0
if result["ok"]:
console.print(f"[green]✓[/green] [dim]({elapsed:.1f}s)[/dim]")
else:
console.print(f"[red]✗[/red] [dim]({elapsed:.1f}s)[/dim]")
if result.get("error"):
console.print(f" [red]{result['error']}[/red]")
failed = True
results.append({
"name": step_name,
"status": "ok" if result["ok"] else "error",
"type": step_type,
"elapsed": elapsed,
**({k: v for k, v in result.items() if k not in ("ok",)}),
})
# Stop on failure unless continue_on_error
if failed and not step.get("continue_on_error", False):
break
# Summary
console.print()
if dry_run:
console.print(f"[yellow]⚡ Dry run complete — {len(results)} steps planned[/yellow]")
elif failed:
console.print("[red]✗ Workflow failed[/red]")
else:
console.print(f"[green]✓ Workflow completed — {len(results)} steps[/green]")
return {"name": self._name, "results": results, "ok": not failed, "dry_run": dry_run}
# ------------------------------------------------------------------
# Step type detection
# ------------------------------------------------------------------
def _detect_step_type(self, step: Dict) -> tuple:
"""Detect step type from keys. Returns (type, target) or (None, None)."""
# Deterministic steps
if "run" in step:
return "shell", step["run"]
if "python" in step:
return "python", step["python"]
if "script" in step:
return "script", step["script"]
if "action" in step:
return "action", step["action"]
# Agent-centric steps
if "agent" in step:
return "agent", step["agent"]
if "judge" in step:
return "judge", step["judge"]
if "approve" in step:
return "approve", step["approve"]
return None, None
# ------------------------------------------------------------------
# Step executors
# ------------------------------------------------------------------
def _execute_step(self, step_type: str, target: Any, step: Dict, flags: Dict) -> Dict:
"""Route to the correct executor."""
try:
# Deterministic steps
if step_type == "shell":
return self._exec_shell(target, step)
elif step_type == "python":
return self._exec_python_script(target, step)
elif step_type == "script":
return self._exec_inline_python(target, step, flags)
elif step_type == "action":
return self._exec_action(target, step, flags)
# Agent-centric steps
elif step_type == "agent":
return self._exec_agent_step(target, step, flags)
elif step_type == "judge":
return self._exec_judge_step(target, step, flags)
elif step_type == "approve":
return self._exec_approve_step(target, step, flags)
else:
return {"ok": False, "error": f"Unknown step type: {step_type}"}
except Exception as e:
return {"ok": False, "error": str(e)}
def _exec_shell(self, cmd: str, step: Dict) -> Dict:
"""Execute a shell command."""
cwd = step.get("cwd", self._cwd)
env = self._build_env(step)
result = subprocess.run(
cmd, shell=True, cwd=cwd, env=env,
capture_output=True, text=True,
timeout=step.get("timeout", 300),
)
if result.returncode != 0:
error = result.stderr.strip() or result.stdout.strip() or f"Exit code {result.returncode}"
return {"ok": False, "error": error, "returncode": result.returncode}
return {"ok": True, "output": result.stdout.strip()}
def _exec_python_script(self, script_path: str, step: Dict) -> Dict:
"""Execute a Python script file."""
# Resolve relative to workflow file
resolved = Path(self._cwd) / script_path
if not resolved.exists():
return {"ok": False, "error": f"Script not found: {resolved}"}
args_str = step.get("args", "")
args_list = args_str.split() if isinstance(args_str, str) else args_str
cmd = [sys.executable, str(resolved)] + args_list
env = self._build_env(step)
result = subprocess.run(
cmd, cwd=self._cwd, env=env,
capture_output=True, text=True,
timeout=step.get("timeout", 300),
)
if result.returncode != 0:
error = result.stderr.strip() or result.stdout.strip() or f"Exit code {result.returncode}"
return {"ok": False, "error": error, "returncode": result.returncode}
return {"ok": True, "output": result.stdout.strip()}
def _exec_inline_python(self, code: str, step: Dict, flags: Dict) -> Dict:
"""Execute inline Python code in an isolated namespace."""
namespace = {
"flags": flags,
"vars": {k: self._resolve_var_value(v) for k, v in self._vars.items()},
"env": os.environ.copy(),
"cwd": self._cwd,
"__builtins__": __builtins__,
}
try:
exec(code, namespace)
return {"ok": True, "output": namespace.get("result", "")}
except Exception as e:
return {"ok": False, "error": str(e)}
def _exec_action(self, action_name: str, step: Dict, flags: Dict) -> Dict:
"""
Execute an action with dynamic resolution.
Resolution order:
1. YAML-defined actions (actions: block in the workflow)
2. File-based actions (.py files from actions/ or .praison/actions/)
3. Built-in actions (bump-version, etc.)
"""
# 1. YAML-defined actions
yaml_actions = self._data.get("actions", {})
if action_name in yaml_actions:
return self._exec_yaml_action(action_name, yaml_actions[action_name], step, flags)
# 2. File-based actions
action_file = self._find_action_file(action_name)
if action_file:
return self._exec_file_action(action_file, step, flags)
# 3. Built-in actions
builtin = self._get_builtin_action(action_name)
if builtin:
return builtin(step, flags)
return {"ok": False, "error": f"Unknown action: {action_name}"}
def _exec_yaml_action(self, name: str, action_def: Dict, step: Dict, flags: Dict) -> Dict:
"""Execute an action defined inline in the YAML actions: block."""
# Agent-powered action
if "agent" in action_def:
return self._exec_agent_step(action_def["agent"], {**action_def, **step}, flags)
if "script" in action_def:
# Merge action-level config into namespace
merged_step = {**action_def, **step}
return self._exec_inline_python(action_def["script"], merged_step, flags)
elif "run" in action_def:
merged_step = {**action_def, **step}
return self._exec_shell(action_def["run"], merged_step)
elif "python" in action_def:
merged_step = {**action_def, **step}
return self._exec_python_script(action_def["python"], merged_step)
return {"ok": False, "error": f"Action '{name}' has no run/script/python/agent key"}
def _find_action_file(self, action_name: str) -> Optional[Path]:
"""
Find an action .py file by name.
Search order:
1. ./actions/{action_name}.py (next to workflow)
2. .praison/actions/{action_name}.py (project-level)
"""
# Normalize name: bump-version → bump_version.py
file_name = action_name.replace("-", "_") + ".py"
# Search next to workflow file
local = Path(self._cwd) / "actions" / file_name
if local.exists():
return local
# Search project-level
project = Path(self._cwd) / ".praison" / "actions" / file_name
if project.exists():
return project
return None
def _exec_file_action(self, action_file: Path, step: Dict, flags: Dict) -> Dict:
"""
Execute a file-based action.
The action file must define a `run(step, flags, cwd)` function
that returns a dict with at minimum {"ok": True/False}.
"""
import importlib.util
try:
spec = importlib.util.spec_from_file_location("action_module", str(action_file))
if not spec or not spec.loader:
return {"ok": False, "error": f"Cannot load action: {action_file}"}
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
if not hasattr(module, "run"):
return {"ok": False, "error": f"Action file {action_file.name} missing run(step, flags, cwd) function"}
result = module.run(step=step, flags=flags, cwd=self._cwd)
if not isinstance(result, dict):
return {"ok": True, "output": str(result)}
return result
except Exception as e:
return {"ok": False, "error": str(e)}
def _get_builtin_action(self, action_name: str):
"""Return built-in action handler or None."""
builtins = {
"bump-version": self._action_bump_version,
}
return builtins.get(action_name)
# ------------------------------------------------------------------
# Agent-centric step executors
# ------------------------------------------------------------------
def _exec_agent_step(self, agent_config: Any, step: Dict, flags: Dict) -> Dict:
"""
Execute an agent step using praisonaiagents.Agent.
Args:
agent_config: Agent configuration (dict with role, instructions, tools, etc.)
step: Full step definition (may include output_file)
flags: Workflow flags
Returns:
Dict with ok, output, and optionally error
"""
try:
from praisonaiagents import Agent
except ImportError:
return {"ok": False, "error": "praisonaiagents not installed. Install with: pip install praisonaiagents"}
# Handle both dict config and simple string
if isinstance(agent_config, str):
agent_config = {"instructions": agent_config}
# Build agent
agent_name = agent_config.get("name", step.get("name", "workflow-agent"))
role = agent_config.get("role", "Assistant")
instructions = agent_config.get("instructions", "")
model = agent_config.get("model", "gpt-4o-mini")
tools = self._resolve_agent_tools(agent_config.get("tools", []))
try:
agent = Agent(
name=agent_name,
role=role,
instructions=instructions,
tools=tools,
llm=model,
)
# Get prompt - use prompt field, or instructions, or goal
prompt = agent_config.get("prompt", agent_config.get("goal", instructions))
prompt = self._resolve_vars(str(prompt), flags)
# Execute agent
result = agent.chat(prompt)
output = str(result) if result else ""
# Write to file if specified
output_file = step.get("output_file") or agent_config.get("output_file")
if output_file:
output_path = Path(self._cwd) / output_file
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(output)
return {"ok": True, "output": output}
except Exception as e:
error_msg = str(e)
# Check for API key issues
if "api_key" in error_msg.lower() or "authentication" in error_msg.lower():
return {"ok": False, "error": f"Agent step '{agent_name}' requires an LLM API key. Set OPENAI_API_KEY or use --model ollama/llama3 for local execution. Original error: {error_msg}"}
return {"ok": False, "error": error_msg}
def _exec_judge_step(self, judge_config: Any, step: Dict, flags: Dict) -> Dict:
"""
Execute a judge step for quality gating.
Args:
judge_config: Judge configuration (dict with input_file, criteria, threshold, etc.)
step: Full step definition
flags: Workflow flags
Returns:
Dict with ok, score, feedback, and optionally error
"""
try:
from praisonaiagents.eval import Judge
except ImportError:
return {"ok": False, "error": "praisonaiagents.eval not available. Install with: pip install praisonaiagents"}
# Handle both dict config and simple string (criteria)
if isinstance(judge_config, str):
judge_config = {"criteria": judge_config}
# Get input
input_file = judge_config.get("input_file")
if input_file:
input_path = Path(self._cwd) / input_file
if not input_path.exists():
return {"ok": False, "error": f"Input file not found: {input_path}"}
input_text = input_path.read_text()
else:
input_text = judge_config.get("input", "")
if not input_text:
return {"ok": False, "error": "Judge step requires input_file or input"}
criteria = judge_config.get("criteria", "Output is high quality")
threshold = float(judge_config.get("threshold", 7.0))
model = judge_config.get("model", "gpt-4o-mini")
on_fail = judge_config.get("on_fail", "stop") # stop | warn | retry
try:
judge = Judge(criteria=criteria, model=model)
result = judge.evaluate(input_text)
score = result.score if hasattr(result, 'score') else 0.0
feedback = result.feedback if hasattr(result, 'feedback') else str(result)
passed = score >= threshold
if not passed and on_fail == "warn":
# Warn but continue
return {"ok": True, "score": score, "feedback": feedback, "passed": False, "warning": f"Score {score} below threshold {threshold}"}
return {"ok": passed, "score": score, "feedback": feedback, "passed": passed}
except Exception as e:
return {"ok": False, "error": str(e)}
def _exec_approve_step(self, approve_config: Any, step: Dict, flags: Dict) -> Dict:
"""
Execute an approval gate step.
Args:
approve_config: Approval configuration (dict with description, risk_level, etc.)
step: Full step definition
flags: Workflow flags
Returns:
Dict with ok, reason, and optionally error
"""
try:
from praisonaiagents.approval.backends import ConsoleBackend, AutoApproveBackend
from praisonaiagents.approval.protocols import ApprovalRequest
except ImportError:
return {"ok": False, "error": "praisonaiagents.approval not available. Install with: pip install praisonaiagents"}
# Handle both dict config and simple string (description)
if isinstance(approve_config, str):
approve_config = {"description": approve_config}
description = approve_config.get("description", step.get("name", "Workflow step"))
risk_level = approve_config.get("risk_level", "medium")
auto_approve = approve_config.get("auto_approve", False)
# Build context from previous step results if available
context = {
"step_name": step.get("name", "Unknown"),
"workflow": self._name,
"description": description,
}
request = ApprovalRequest(
tool_name="workflow_step",
arguments=context,
risk_level=risk_level,
agent_name=self._name,
)
try:
if auto_approve:
backend = AutoApproveBackend()
else:
backend = ConsoleBackend()
decision = backend.request_approval_sync(request)
return {
"ok": decision.approved,
"reason": decision.reason,
"approver": getattr(decision, 'approver', 'user'),
}
except Exception as e:
return {"ok": False, "error": str(e)}
def _resolve_agent_tools(self, tool_names: List) -> List:
"""
Resolve tool names to actual tool functions.
Args:
tool_names: List of tool names (strings) or tool functions
Returns:
List of resolved tool functions
"""
if not tool_names:
return []
resolved = []
for tool in tool_names:
if callable(tool):
# Already a function
resolved.append(tool)
elif isinstance(tool, str):
# Try to resolve from registry
try:
from praisonaiagents.tools import get_tool
resolved_tool = get_tool(tool)
if resolved_tool:
resolved.append(resolved_tool)
except (ImportError, Exception):
# Try common tools
if tool == "execute_command":
try:
from praisonaiagents.tools import execute_command
resolved.append(execute_command)
except ImportError:
pass
elif tool == "read_file":
try:
from praisonaiagents.tools import read_file
resolved.append(read_file)
except ImportError:
pass
elif tool == "write_file":
try:
from praisonaiagents.tools import write_file
resolved.append(write_file)
except ImportError:
pass
return resolved
# ------------------------------------------------------------------
# Built-in actions
# ------------------------------------------------------------------
def _action_bump_version(self, step: Dict, flags: Dict) -> Dict:
"""Bump version in pyproject.toml (or similar)."""
file_path = step.get("file", "pyproject.toml")
resolved = Path(self._cwd) / file_path
if not resolved.exists():
return {"ok": False, "error": f"File not found: {resolved}"}
# Determine strategy from flags, falling back to step config
if flags.get("major"):
strategy = "major"
elif flags.get("minor"):
strategy = "minor"
else:
strategy = step.get("strategy", "patch")
try:
content = resolved.read_text()
# Match version = "X.Y.Z" in pyproject.toml
pattern = r'(version\s*=\s*["\'])(\d+)\.(\d+)\.(\d+)(["\'])'
match = re.search(pattern, content)
if not match:
return {"ok": False, "error": f"No version found in {file_path}"}
major, minor, patch = int(match.group(2)), int(match.group(3)), int(match.group(4))
old_version = f"{major}.{minor}.{patch}"
if strategy == "major":
major += 1
minor = 0
patch = 0
elif strategy == "minor":
minor += 1
patch = 0
else: # patch
patch += 1
new_version = f"{major}.{minor}.{patch}"
new_content = re.sub(
pattern,
rf'\g<1>{new_version}\g<5>',
content,
count=1,
)
resolved.write_text(new_content)
return {"ok": True, "output": f"{old_version} → {new_version}", "old_version": old_version, "new_version": new_version}
except Exception as e:
return {"ok": False, "error": str(e)}
# ------------------------------------------------------------------
# Variable / env resolution
# ------------------------------------------------------------------
def _resolve_vars(self, text: str, flags: Dict) -> str:
"""Resolve ${{ env.VAR }} and {{ flags.X }} in text."""
if not isinstance(text, str):
return text
# ${{ env.VAR_NAME }}
def env_replacer(m):
var_name = m.group(1).strip()
return os.environ.get(var_name, f"${{{{ env.{var_name} }}}}")
text = re.sub(r'\$\{\{\s*env\.(\w+)\s*\}\}', env_replacer, text)
# {{ flags.X }}
def flag_replacer(m):
flag_name = m.group(1).strip()
return str(flags.get(flag_name, ""))
text = re.sub(r'\{\{\s*flags\.(\w+)\s*\}\}', flag_replacer, text)
return text
def _resolve_var_value(self, var_def) -> str:
"""Resolve a var definition to its value."""
if isinstance(var_def, dict):
default = var_def.get("default", "")
# Check env var with same name
env_name = str(default)
return os.environ.get(env_name, default)
return str(var_def)
def _build_env(self, step: Dict) -> Dict[str, str]:
"""Build environment for subprocess, merging step-level env."""
env = os.environ.copy()
step_env = step.get("env", {})
if step_env:
env.update({k: str(v) for k, v in step_env.items()})
return env
# ------------------------------------------------------------------
# Flag parsing
# ------------------------------------------------------------------
def _parse_flags(self, args: List[str]) -> Dict[str, bool]:
"""Parse CLI flags based on workflow flag definitions."""
flags = {}
for flag_name in self._flag_defs:
# Normalize: no-bump → no_bump for Python, but match --no-bump on CLI
cli_flag = f"--{flag_name}"
flags[flag_name.replace("-", "_")] = cli_flag in args
return flags
# ------------------------------------------------------------------
# Condition evaluation
# ------------------------------------------------------------------
def _eval_condition(self, condition: str, flags: Dict) -> bool:
"""Evaluate a simple condition expression."""
# Strip {{ }} if present
condition = condition.strip()
if condition.startswith("{{") and condition.endswith("}}"):
condition = condition[2:-2].strip()
# Build evaluation context
context = {"flags": type("Flags", (), flags)(), "env": os.environ}
try:
return bool(eval(condition, {"__builtins__": {}}, context))
except Exception:
return False
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _truncate(text: str, max_len: int) -> str:
"""Truncate text for display."""
text = text.replace("\n", " ").strip()
if len(text) > max_len:
return text[:max_len - 3] + "..."
return text
@@ -18,6 +18,7 @@ class OutputMode(Enum):
COMPACT = "compact"
VERBOSE = "verbose"
QUIET = "quiet"
EDITOR = "editor" # User-friendly conversation format
@dataclass
@@ -42,6 +43,8 @@ def get_default_mode() -> OutputMode:
return OutputMode.COMPACT
elif env_mode == "quiet":
return OutputMode.QUIET
elif env_mode == "editor":
return OutputMode.EDITOR
return OutputMode.VERBOSE
@@ -354,6 +354,10 @@ Example:
"""
Run a YAML workflow file.
Supports two modes:
- type: job → deterministic step executor (shell, python, actions)
- default → agent workflow (LLM-based)
Args:
args: [file.yaml, --var, key=value, ...]
@@ -369,6 +373,24 @@ Example:
self.print_status(f"File not found: {file_path}", "error")
return None
# Detect type: job or hybrid → route to appropriate executor
try:
import yaml
with open(file_path, 'r') as f:
raw_data = yaml.safe_load(f)
if isinstance(raw_data, dict):
workflow_type = raw_data.get("type")
if workflow_type == "job":
from .job_workflow import JobWorkflowExecutor
executor = JobWorkflowExecutor(raw_data, file_path)
return executor.run(args)
elif workflow_type == "hybrid":
from .hybrid_workflow import HybridWorkflowExecutor
executor = HybridWorkflowExecutor(raw_data, file_path)
return executor.run(args)
except Exception:
pass # Fall through to agent workflow
manager = self._get_manager()
if not manager:
return None
@@ -564,6 +586,7 @@ Example:
table = Table(title="Available Workflows")
table.add_column("File", style="cyan")
table.add_column("Name", style="green")
table.add_column("Type", style="magenta")
table.add_column("Steps", style="yellow")
parser = self._get_parser()
@@ -571,18 +594,33 @@ Example:
for yaml_file in yaml_files:
try:
if parser:
# Check if it's a job workflow
import yaml
with open(yaml_file, 'r') as f:
raw = yaml.safe_load(f)
if isinstance(raw, dict) and raw.get("type") == "job":
wf_name = raw.get("name", yaml_file.stem)
step_count = len(raw.get("steps", []))
table.add_row(
yaml_file.name,
wf_name,
"[bold magenta]job[/bold magenta]",
str(step_count),
)
elif parser:
workflow = parser.parse_file(yaml_file)
table.add_row(
yaml_file.name,
workflow.name,
str(len(workflow.steps))
"[dim]agent[/dim]",
str(len(workflow.steps)),
)
else:
table.add_row(yaml_file.name, "N/A", "N/A")
table.add_row(yaml_file.name, "N/A", "?", "N/A")
workflow_names.append(yaml_file.name)
except Exception as e:
table.add_row(yaml_file.name, f"Error: {e}", "N/A")
table.add_row(yaml_file.name, f"Error: {e}", "?", "N/A")
rprint(table)
return workflow_names
@@ -647,7 +685,7 @@ Example:
Auto-generate a workflow from a topic description.
Args:
args: ["topic description", --pattern, <pattern>, --output, <file.yaml>]
args: ["topic description", --pattern, <pattern>, --output, <file.yaml>, --type, <job|agent>]
Returns:
Path to created file
@@ -656,6 +694,9 @@ Example:
topic = None
pattern = "sequential"
output_file = None
workflow_type = "agent" # Default to agent workflow
include_judge = False
include_approve = False
i = 0
while i < len(args):
@@ -665,6 +706,15 @@ Example:
elif args[i] == "--output" and i + 1 < len(args):
output_file = args[i + 1]
i += 2
elif args[i] == "--type" and i + 1 < len(args):
workflow_type = args[i + 1]
i += 2
elif args[i] == "--judge":
include_judge = True
i += 1
elif args[i] == "--approve":
include_approve = True
i += 1
elif not args[i].startswith("--") and topic is None:
topic = args[i]
i += 1
@@ -672,40 +722,63 @@ Example:
i += 1
if not topic:
self.print_status('Usage: praisonai workflow auto "topic" --pattern <pattern>', "error")
self.print_status('Usage: praisonai workflow auto "topic" [--type job|agent] [--pattern <pattern>]', "error")
self.print_status("Types: job (agent-centric steps), agent (multi-agent workflow)", "info")
self.print_status("Patterns: sequential, routing, parallel", "info")
return None
# Validate pattern
valid_patterns = ["sequential", "routing", "parallel", "loop", "orchestrator-workers", "evaluator-optimizer"]
if pattern not in valid_patterns:
self.print_status(f"Unknown pattern: {pattern}", "error")
self.print_status(f"Valid patterns: {', '.join(valid_patterns)}", "info")
# Validate workflow type
if workflow_type not in ["job", "agent"]:
self.print_status(f"Unknown type: {workflow_type}", "error")
self.print_status("Valid types: job, agent", "info")
return None
# Validate pattern (only for agent workflows)
if workflow_type == "agent":
valid_patterns = ["sequential", "routing", "parallel", "loop", "orchestrator-workers", "evaluator-optimizer"]
if pattern not in valid_patterns:
self.print_status(f"Unknown pattern: {pattern}", "error")
self.print_status(f"Valid patterns: {', '.join(valid_patterns)}", "info")
return None
# Default output file
if not output_file:
# Create safe filename from topic
safe_name = "".join(c if c.isalnum() else "_" for c in topic[:30]).lower()
output_file = f"{safe_name}_workflow.yaml"
suffix = "job_workflow" if workflow_type == "job" else "workflow"
output_file = f"{safe_name}_{suffix}.yaml"
# Check if file exists
if os.path.exists(output_file):
self.print_status(f"File already exists: {output_file}", "error")
return None
self.print_status(f"Generating {pattern} workflow for: {topic}", "info")
self.print_status(f"Generating {workflow_type} workflow for: {topic}", "info")
try:
# Lazy import to avoid performance impact
from praisonai.auto import WorkflowAutoGenerator
generator = WorkflowAutoGenerator(
topic=topic,
workflow_file=output_file
)
result_path = generator.generate(pattern=pattern)
if workflow_type == "job":
# Generate job workflow with agent-centric steps
from praisonai.auto import JobWorkflowAutoGenerator
generator = JobWorkflowAutoGenerator(
topic=topic,
workflow_file=output_file
)
result_path = generator.generate(
include_judge=include_judge,
include_approve=include_approve
)
else:
# Generate agent workflow (existing behavior)
from praisonai.auto import WorkflowAutoGenerator
generator = WorkflowAutoGenerator(
topic=topic,
workflow_file=output_file
)
result_path = generator.generate(pattern=pattern)
self.print_status(f"✓ Created workflow: {result_path}", "success")
self.print_status(f"Run with: praisonai workflow run {output_file}", "info")
+7 -2
View File
@@ -13,7 +13,7 @@ from typing import List, Tuple
LEGACY_COMMANDS = {
'chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research',
'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools',
'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile',
'todo', 'docs', 'mcp', 'commit', 'publish', 'serve', 'schedule', 'skills', 'profile',
'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy',
'templates', 'recipe', 'endpoints', 'audio', 'embed', 'images', 'moderate',
'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants',
@@ -34,7 +34,7 @@ TYPER_COMMANDS = {
# Previously legacy-only commands (now in Typer)
'chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research',
'memory', 'workflow', 'tools', 'knowledge', 'deploy', 'agents', 'skills',
'eval', 'templates', 'recipe', 'todo', 'docs', 'commit', 'hooks', 'rules',
'eval', 'templates', 'recipe', 'todo', 'docs', 'commit', 'publish', 'hooks', 'rules',
'registry', 'package', 'endpoints', 'test', 'examples', 'batch',
# Replay commands
'replay',
@@ -99,11 +99,16 @@ def is_legacy_invocation(argv: List[str]) -> bool:
'-p', '--autonomy', '--trust',
'--sandbox', '--external-agent', '--compare', '--interval', '--timeout',
'--max-cost', '--rpm', '--tpm', '--temperature',
# Display flags (verbosity ladder + machine output)
'--output', '--flow',
]
for arg in argv:
if arg in legacy_flags:
return True
# Handle -vv, -qq, -vvv etc. (count flags not in Typer)
if len(arg) > 2 and arg.startswith('-') and not arg.startswith('--') and arg[1:] in ('vv', 'qq', 'vvv', 'qqq'):
return True
# Check if first arg is a legacy command not in Typer
if first_arg in LEGACY_COMMANDS and first_arg not in TYPER_COMMANDS:
+285 -261
View File
@@ -260,8 +260,15 @@ class PraisonAI:
try:
# Check if stdin is not a terminal (i.e., has piped input)
if not sys.stdin.isatty():
stdin_content = sys.stdin.read().strip()
return stdin_content if stdin_content else None
import select
# Non-blocking check: only read if data is actually available.
# Without this, sys.stdin.read() blocks forever in non-TTY
# environments (subprocesses, CI/CD, IDE terminals, Docker)
# where stdin is a pipe with no EOF.
if select.select([sys.stdin], [], [], 0.0)[0]:
stdin_content = sys.stdin.read().strip()
return stdin_content if stdin_content else None
return None
except Exception:
# If there's any error reading stdin, ignore it
pass
@@ -794,7 +801,7 @@ class PraisonAI:
return default_args
# Define special commands
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy', 'templates', 'recipe', 'endpoints', 'audio', 'embed', 'embedding', 'images', 'moderate', 'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants', 'fine-tuning', 'completions', 'messages', 'guardrails', 'rag', 'videos', 'a2a', 'containers', 'passthrough', 'responses', 'search', 'realtime-api', 'doctor', 'registry', 'package', 'install', 'uninstall', 'acp', 'debug', 'lsp', 'diag', 'browser', 'replay', 'bot', 'gateway', 'sandbox', 'wizard', 'migrate', 'security', 'persistence']
special_commands = ['chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research', 'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools', 'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile', 'eval', 'agents', 'run', 'thinking', 'compaction', 'output', 'deploy', 'templates', 'recipe', 'endpoints', 'audio', 'embed', 'embedding', 'images', 'moderate', 'files', 'batches', 'vector-stores', 'rerank', 'ocr', 'assistants', 'fine-tuning', 'completions', 'messages', 'guardrails', 'rag', 'videos', 'a2a', 'containers', 'passthrough', 'responses', 'search', 'realtime-api', 'doctor', 'registry', 'package', 'install', 'uninstall', 'acp', 'debug', 'lsp', 'diag', 'browser', 'replay', 'bot', 'gateway', 'sandbox', 'wizard', 'migrate', 'security', 'persistence', 'paths']
parser = argparse.ArgumentParser(prog="praisonai", description="praisonAI command-line interface")
parser.add_argument("--framework", choices=["crewai", "autogen", "praisonai"], help="Specify the framework")
@@ -831,7 +838,14 @@ class PraisonAI:
parser.add_argument("--no-acp", action="store_true", help="Disable ACP tools (agentic file operations with plan/approve/apply)")
parser.add_argument("--no-lsp", action="store_true", help="Disable LSP tools (code intelligence: symbols, definitions, references)")
parser.add_argument("--save", "-s", action="store_true", help="Save research output to file (output/research/)")
parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose output for research")
parser.add_argument("-v", "--verbose", action="count", default=0,
help="Increase verbosity (-v=verbose, -vv=debug)")
parser.add_argument("-q", "--quiet", action="count", default=0,
help="Decrease verbosity (-q=quiet, -qq=silent)")
parser.add_argument("--output", type=str, choices=["json", "jsonl", "editor"], dest="output_format",
help="Output format (json, jsonl, or editor for user-friendly conversation-style)")
parser.add_argument("--flow", action="store_true",
help="Show visual agent→tool flow chart")
parser.add_argument("--web", "--web-search", action="store_true", help="Enable native web search (OpenAI, Gemini, Anthropic, xAI, Perplexity)")
parser.add_argument("--web-fetch", action="store_true", help="Enable web fetch to retrieve URL content (Anthropic only)")
parser.add_argument("--prompt-caching", action="store_true", help="Enable prompt caching to reduce costs (OpenAI, Anthropic, Bedrock, Deepseek)")
@@ -938,8 +952,19 @@ class PraisonAI:
# Direct prompt flag - alternative to positional command
parser.add_argument("-p", "--prompt", type=str, dest="prompt_flag", help="Direct prompt to execute (alternative to positional argument)")
# Autonomy Mode - control AI action approval
parser.add_argument("--autonomy", type=str, choices=["suggest", "auto_edit", "full_auto"], help="Set autonomy mode for AI actions")
# Autonomy Mode - full_auto by default for multi-turn autonomous execution
# ACP tools disabled by default for speed; use --acp to enable
parser.add_argument("--autonomy", type=str, choices=["suggest", "auto_edit", "full_auto", "disable"],
default=None, help="Set autonomy mode. Only use for multi-turn tasks that need iterative execution.")
parser.add_argument("--acp", action="store_true",
help="Enable ACP tools in autonomy mode (slower but more powerful file operations)")
parser.add_argument("--lsp", action="store_true",
help="Enable LSP tools in autonomy mode (slower but provides code intelligence)")
# P8/G11: Tool timeout - prevent slow tools from blocking
parser.add_argument("--tool-timeout", type=int, default=60,
help="Timeout in seconds for each tool call (default: 60)")
# Tool Approval - control tool execution approval
parser.add_argument("--trust", action="store_true", help="Auto-approve all tool executions (skip approval prompts)")
@@ -1686,6 +1711,19 @@ class PraisonAI:
exit_code = handle_audit_command(unknown_args)
sys.exit(exit_code)
elif args.command == 'paths':
# Paths command - storage path inspection
# Routes to Typer CLI for paths commands (show)
from .app import app as typer_app, register_commands
register_commands()
import sys as _sys
_sys.argv = ['praisonai', 'paths'] + unknown_args
try:
typer_app()
except SystemExit as e:
sys.exit(e.code if e.code else 0)
sys.exit(0)
# Only check framework availability for agent-related operations
if not args.command and (args.init or args.auto or args.framework):
if not CREWAI_AVAILABLE and not AUTOGEN_AVAILABLE and not PRAISONAI_AVAILABLE:
@@ -3491,7 +3529,7 @@ class PraisonAI:
return
# Handle auto mode
auto_mode = '-a' in args or '--auto' in args
auto_mode = '-a' in args or '--auto' in args or '--all' in args
if auto_mode:
print("[cyan]Auto-staging all changes...[/cyan]")
subprocess.run(["git", "add", "-A"], capture_output=True)
@@ -3861,15 +3899,45 @@ Provide ONLY the commit message, no explanations."""
"backstory": "You are a helpful AI assistant"
}
# Set output mode based on --verbose flag
# Uses consolidated 'output' param instead of deprecated 'verbose'
if hasattr(self, 'args') and getattr(self.args, 'verbose', False):
# Autonomy Mode - full_auto by default for multi-turn autonomous execution
# ACP tools disabled by default for speed (~3s vs 174s+); use --acp to enable
autonomy_mode = getattr(self.args, 'autonomy', None) if hasattr(self, 'args') else None
if autonomy_mode and autonomy_mode not in ('disable', None):
agent_config["autonomy"] = {"level": autonomy_mode, "enabled": True}
# Set SDK output preset based on verbosity flags
# The display dispatcher handles CLI rendering; this controls SDK-level behavior
v = getattr(self.args, 'verbose', 0) if hasattr(self, 'args') else 0
if v >= 2:
agent_config["output"] = "verbose" # SDK debug-level detail
elif v >= 1:
agent_config["output"] = "verbose"
else:
agent_config["output"] = "minimal"
# Load default tools (same as interactive mode) unless --no-tools is set
# CRITICAL: For autonomy mode, disable slow ACP and LSP tools by default
# ACP tools go through complex orchestration (174s+ delays)
# LSP tools require starting language server (176s+ delays)
# Use --acp or --lsp flags to explicitly enable them.
if not getattr(self.args, 'no_tools', False):
# Check if autonomy mode is enabled - disable ACP/LSP by default for speed
is_autonomy = autonomy_mode and autonomy_mode not in ('disable', None)
use_acp = getattr(self.args, 'acp', False) # Explicit --acp flag
use_lsp = getattr(self.args, 'lsp', False) # Explicit --lsp flag
if is_autonomy:
if not use_acp:
# Force disable ACP for autonomy mode (use basic tools only)
# This makes full_auto fast (~3s vs 174s+)
self.args.no_acp = True
logging.debug("Autonomy mode: ACP tools disabled for speed (use --acp to enable)")
if not use_lsp:
# Force disable LSP for autonomy mode
# LSP requires starting language server (176s+ delays)
self.args.no_lsp = True
logging.debug("Autonomy mode: LSP tools disabled for speed (use --lsp to enable)")
default_tools = self._load_interactive_tools()
if default_tools:
agent_config["tools"] = default_tools
@@ -3934,6 +4002,11 @@ Provide ONLY the commit message, no explanations."""
if getattr(self.args, 'planning_reasoning', False):
agent_config["planning_reasoning"] = True
# P8/G11: Tool timeout - prevent slow tools from blocking
tool_timeout = getattr(self.args, 'tool_timeout', 60)
if tool_timeout and tool_timeout > 0:
agent_config["tool_timeout"] = tool_timeout
# Memory
if getattr(self.args, 'memory', False):
agent_config["memory"] = True
@@ -4211,16 +4284,175 @@ Provide ONLY the commit message, no explanations."""
else:
result = auto_rag.chat(prompt)
else:
# Run with minimal status display when verbose=False
is_verbose = agent_config.get("verbose", False)
if not is_verbose:
result = self._run_with_status_display(agent, prompt)
else:
# Try start method first, fallback to chat
# Resolve display mode from CLI flags
display_mode = self._resolve_display_mode()
if display_mode == 'silent':
# -qq: No output at all, exit code only
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
elif display_mode == 'quiet':
# -q: Result only, no spinners or status
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
if result is not None:
output = getattr(result, 'output', None) or (str(result) if result else None)
if output:
print(output)
elif display_mode == 'verbose':
# -v: SDK StatusOutput with timestamps and metrics
try:
from praisonaiagents.output.status import enable_status_output, disable_status_output
enable_status_output(show_timestamps=True, show_metrics=True)
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
disable_status_output()
except ImportError:
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
elif display_mode == 'debug':
# -vv: SDK TraceOutput with markdown rendering
try:
from praisonaiagents.output.trace import enable_trace_output, disable_trace_output
enable_trace_output(use_markdown=True)
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
disable_trace_output()
except ImportError:
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
elif display_mode == 'jsonl':
# --output jsonl: JSONL structured output for CI/CD
from .features.display_jsonl import JsonlDisplay
from praisonaiagents.main import register_display_callback as _reg_cb
jsonl = JsonlDisplay()
jsonl.on_init(model=agent_config.get('llm'))
_reg_cb('tool_call', jsonl.on_tool_call)
_reg_cb('llm_start', jsonl.on_llm_start)
_reg_cb('autonomy_iteration', jsonl.on_autonomy_iteration)
_reg_cb('autonomy_stage_change', jsonl.on_autonomy_stage_change)
_reg_cb('autonomy_doom_loop', jsonl.on_autonomy_doom_loop)
_reg_cb('autonomy_complete', lambda **kw: jsonl.on_complete(
reason=kw.get('completion_reason'),
iterations=kw.get('iterations'),
duration_ms=(kw.get('duration_seconds') or 0) * 1000,
))
start_time = time.time()
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
# Emit final result
reason = getattr(result, 'completion_reason', None) if hasattr(result, 'completion_reason') else 'complete'
iters = getattr(result, 'iterations', None) if hasattr(result, 'iterations') else None
jsonl.on_complete(reason=reason or 'complete', iterations=iters,
duration_ms=(time.time() - start_time) * 1000)
# Print result to stdout
if hasattr(result, 'output') and result.output:
print(result.output)
elif isinstance(result, str):
print(result)
elif display_mode == 'json':
# --output json: JSON envelope output
import json as json_mod
start_time = time.time()
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
output = result.output if hasattr(result, 'output') else str(result)
envelope = {
'result': output,
'model': agent_config.get('llm'),
'duration_ms': round((time.time() - start_time) * 1000),
}
if hasattr(result, 'completion_reason'):
envelope['completion_reason'] = result.completion_reason
if hasattr(result, 'iterations'):
envelope['iterations'] = result.iterations
print(json_mod.dumps(envelope, indent=2))
elif display_mode == 'flow':
# --flow: SDK FlowDisplay - visual agent→tool chart
try:
from praisonaiagents.flow_display import track_workflow
flow = track_workflow()
flow.start()
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
flow.stop()
except ImportError:
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
elif display_mode == 'editor':
# --output editor: User-friendly step-by-step format
# Uses SDK EditorOutput (Step 1: 📄 Creating file → ✓ Done)
from praisonaiagents.output.editor import enable_editor_output, disable_editor_output
editor = enable_editor_output(use_color=True)
# Run agent
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
# Display final result
output = result.output if hasattr(result, 'output') else str(result)
if output:
editor.output(output)
# Show summary
elapsed = editor.elapsed_time()
editor.summary("Completed", [
f"Duration: {elapsed:.1f}s",
f"Blocks: {len(editor.get_blocks())}",
])
disable_editor_output()
else:
# Default: SDK status output — clean inline progress
# Shows: spinner + tool calls, no panels, no timestamps
try:
from praisonaiagents.output.status import enable_status_output, disable_status_output
enable_status_output(show_timestamps=False, show_metrics=False)
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
disable_status_output()
except ImportError:
if hasattr(agent, 'start'):
result = agent.start(prompt)
else:
result = agent.chat(prompt)
# ===== POST-PROCESSING WITH NEW FEATURES =====
@@ -4404,257 +4636,49 @@ Now, {final_instruction.lower()}:"""
# Return the actual result for any downstream processing
return result.output
def _run_with_status_display(self, agent, prompt):
def _resolve_display_mode(self):
"""Map CLI flags to a display mode string.
Priority: --output > --flow > --display (deprecated) > -v/-q > default.
Returns one of: 'silent', 'quiet', 'verbose', 'debug', 'json', 'jsonl', 'flow', 'status', 'cursor'.
"""
Run agent with minimal status display (spinner + tool/handoff updates).
# Machine formats take highest priority
output_fmt = getattr(self.args, 'output_format', None)
if output_fmt:
return output_fmt # "json", "jsonl", or "cursor"
Shows:
- "Generating..." with spinner while processing
- Real-time tool call notifications via registered callback
- Agent handoff notifications
"""
import threading
import time
from rich.console import Console
from rich.live import Live
from rich.text import Text
# --flow is an independent feature flag
if getattr(self.args, 'flow', False) or getattr(self.args, 'flow_display', False):
return 'flow'
console = Console()
# Import callback registration from praisonaiagents
# Store in local variable for safe access throughout the function
_sync_display_callbacks = None
_register_display_callback = None
# Check Typer global state (for Typer subcommands)
try:
from praisonaiagents import register_display_callback, sync_display_callbacks
_sync_display_callbacks = sync_display_callbacks
_register_display_callback = register_display_callback
except ImportError:
# Fallback if callbacks not available - just run agent directly
try:
result = agent.start(prompt)
except AttributeError:
# Try chat method if start not available
result = agent.chat(prompt)
if result:
console.print(result)
elif result == "" or result is None:
console.print("[dim]No response generated[/dim]")
return result
# Get tool names for display
tool_names = []
if hasattr(agent, 'tools') and agent.tools:
for tool in agent.tools:
if hasattr(tool, '__name__'):
tool_names.append(tool.__name__)
elif hasattr(tool, 'name'):
tool_names.append(tool.name)
status_info = {
'status': 'Generating...',
'tool_calls': [],
'handoffs': [],
'done': False,
'result': None,
'error': None,
'start_time': time.time(),
'available_tools': tool_names,
'approval_pending': False, # Flag to pause Live display during approval
'live_instance': None # Reference to Live instance for stopping
}
def tool_call_callback(message):
"""Callback triggered when a tool is called."""
# Extract tool name from message
if "Calling function:" in message:
# Format: "Calling function: function_name"
parts = message.split("Calling function:")
if len(parts) > 1:
tool_name = parts[1].strip()
if tool_name and tool_name not in status_info['tool_calls']:
status_info['tool_calls'].append(tool_name)
status_info['status'] = f"Using {tool_name}..."
elif "Function " in message and " returned:" in message:
# Format: "Function function_name returned: ..."
# Tool execution completed, update status
status_info['status'] = "Processing result..."
# Register callback for tool calls (use local variable)
_register_display_callback('tool_call', tool_call_callback)
def build_status_display():
"""Build the status display text."""
elapsed = time.time() - status_info['start_time']
# Main status with spinner
text = Text()
text.append("⏳ ", style="cyan")
text.append(f"{status_info['status']} ", style="bold")
text.append(f"({elapsed:.1f}s)", style="dim")
# Show available tools on first line
if status_info['available_tools'] and not status_info['tool_calls']:
tools_str = ', '.join(status_info['available_tools'][:3])
if len(status_info['available_tools']) > 3:
tools_str += f" +{len(status_info['available_tools']) - 3} more"
text.append(f"\n 🔧 Tools: {tools_str}", style="dim")
# Show recent tool calls
if status_info['tool_calls']:
text.append("\n")
for tool in status_info['tool_calls'][-3:]: # Show last 3
text.append(f" ⚙ {tool}", style="dim yellow")
text.append("\n")
# Show handoffs
if status_info['handoffs']:
for handoff in status_info['handoffs'][-2:]: # Show last 2
text.append(f" → {handoff}", style="dim cyan")
text.append("\n")
return text
def run_agent():
"""Run the agent in background thread."""
try:
status_info['result'] = agent.start(prompt)
except Exception as e:
status_info['error'] = e
finally:
status_info['done'] = True
# Set up approval callback that stops Live display before prompting
# Only if not using --trust (which auto-approves everything)
if not getattr(self.args, 'trust', False):
from praisonaiagents.approval import set_approval_callback, ApprovalDecision
from rich.prompt import Confirm
from rich.panel import Panel
def cli_approval_with_live_pause(function_name, arguments, risk_level):
"""Approval callback that stops Live display before prompting."""
# Signal to stop Live display
status_info['approval_pending'] = True
# Wait a moment for Live to stop
time.sleep(0.2)
# Now show the approval prompt
risk_colors = {"critical": "bold red", "high": "red", "medium": "yellow", "low": "blue"}
risk_color = risk_colors.get(risk_level, "white")
tool_info = f"[bold]Function:[/] {function_name}\n"
tool_info += f"[bold]Risk Level:[/] [{risk_color}]{risk_level.upper()}[/{risk_color}]\n"
tool_info += "[bold]Arguments:[/]\n"
for key, value in arguments.items():
str_value = str(value)[:100] + "..." if len(str(value)) > 100 else str(value)
tool_info += f" {key}: {str_value}\n"
console.print(Panel(tool_info.strip(), title="🔒 Tool Approval Required", border_style=risk_color))
try:
approved = Confirm.ask(f"[{risk_color}]Execute this {risk_level} risk tool?[/{risk_color}]", default=False)
status_info['approval_pending'] = False
if approved:
console.print("[green]✅ Approved[/green]")
return ApprovalDecision(approved=True, reason="User approved")
else:
console.print("[red]❌ Denied[/red]")
console.print("[dim]Tip: Use --trust to auto-approve all tools[/dim]")
return ApprovalDecision(approved=False, reason="User denied")
except (KeyboardInterrupt, EOFError):
status_info['approval_pending'] = False
console.print("\n[red]❌ Cancelled[/red]")
return ApprovalDecision(approved=False, reason="User cancelled")
# Only set if not already set by --approve-level
if not getattr(self.args, 'approve_level', None):
set_approval_callback(cli_approval_with_live_pause)
# Start agent in background thread
thread = threading.Thread(target=run_agent, daemon=True)
thread.start()
# Show live status while processing
# Loop handles unlimited approval interruptions (each approval
# pauses the Live display, then we restart it afterwards)
try:
while not status_info['done']:
with Live(build_status_display(), console=console, refresh_per_second=4, transient=True) as live:
status_info['live_instance'] = live
while not status_info['done']:
# Check if approval is pending - stop Live to show prompt
if status_info['approval_pending']:
break
live.update(build_status_display())
time.sleep(0.1)
# If approval was pending, wait for it to complete then loop
# back to restart the Live display
while status_info['approval_pending']:
time.sleep(0.1)
except KeyboardInterrupt:
console.print("\n[dim]Interrupted[/dim]")
# Unregister callback (use local variable with None check)
if _sync_display_callbacks is not None and 'tool_call' in _sync_display_callbacks:
del _sync_display_callbacks['tool_call']
return None
# Wait for thread to complete (generous timeout for long-running tasks)
thread.join(timeout=5.0)
# Unregister callback to avoid memory leaks (use local variable with None check)
if _sync_display_callbacks is not None and 'tool_call' in _sync_display_callbacks:
del _sync_display_callbacks['tool_call']
# Handle result
if status_info['error']:
console.print(f"[red]Error: {status_info['error']}[/red]")
return None
# Show tool calls that were made (if any)
if status_info['tool_calls']:
console.print(f"[dim]Tools used: {', '.join(status_info['tool_calls'])}[/dim]")
# Get the result and print it directly here to ensure it's displayed
result = status_info['result']
# Handle AutonomyResult from run_autonomous()
display_text = None
if result is not None:
if hasattr(result, 'output') and result.output:
display_text = result.output
elif isinstance(result, str) and result:
display_text = result
elif result:
display_text = str(result)
if display_text:
console.print(display_text)
else:
# Last resort: extract from agent's chat history
extracted = None
if hasattr(agent, 'chat_history') and agent.chat_history:
for msg in reversed(agent.chat_history):
if msg.get('role') == 'assistant' and msg.get('content'):
extracted = msg['content']
break
if extracted:
console.print(extracted)
elif status_info['tool_calls']:
console.print("[dim]Task completed (tools executed successfully)[/dim]")
else:
console.print("[dim]No response generated[/dim]")
# Cleanup LSP/ACP runtime
try:
from .features.interactive_tools import cleanup_runtime
cleanup_runtime()
except Exception:
from .app import state as typer_state
if typer_state.quiet:
return 'quiet'
if hasattr(typer_state, 'output_format'):
if typer_state.output_format.value == 'json':
return 'json'
elif typer_state.output_format.value == 'stream-json':
return 'jsonl'
if typer_state.screen_reader:
return 'verbose' # Accessible: timestamps but no spinners
except (ImportError, AttributeError):
pass
return result
# Verbosity ladder: -v/-vv/-q/-qq
v = getattr(self.args, 'verbose', 0)
q = getattr(self.args, 'quiet', 0)
if q >= 2:
return 'silent'
if q >= 1:
return 'quiet'
if v >= 2:
return 'debug'
if v >= 1:
return 'verbose'
return 'editor' # Default: User-friendly editor-style output
def _handle_serve_command(self, args, unknown_args):
"""
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==4.5.15 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==4.5.32 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+222 -4
View File
@@ -109,6 +109,66 @@ class WebSocketGateway:
await gateway.start()
"""
@classmethod
def from_config_file(
cls,
config_path: str = "config.yaml",
section: str = "gateway",
) -> "WebSocketGateway":
"""Create a gateway from a YAML configuration file.
Gap S5: Enables gateway config persistence via standard config files.
Args:
config_path: Path to the YAML config file
section: Config section containing gateway settings (default: "gateway")
Returns:
Configured WebSocketGateway instance
Example config.yaml:
gateway:
host: "0.0.0.0"
port: 8765
auth_token: "${GATEWAY_AUTH_TOKEN}"
max_connections: 1000
heartbeat_interval: 30
"""
import yaml
if not os.path.exists(config_path):
logger.warning(f"Config file not found: {config_path}, using defaults")
return cls()
with open(config_path, "r") as f:
raw = yaml.safe_load(f) or {}
gateway_config = raw.get(section, {})
if not gateway_config:
logger.warning(f"No '{section}' section in {config_path}, using defaults")
return cls()
# Substitute environment variables
def _substitute(value):
if isinstance(value, str):
return cls._substitute_env_vars(value)
return value
# Build GatewayConfig
config = GatewayConfig(
host=_substitute(gateway_config.get("host", "127.0.0.1")),
port=int(gateway_config.get("port", 8765)),
auth_token=_substitute(gateway_config.get("auth_token")),
max_connections=int(gateway_config.get("max_connections", 1000)),
heartbeat_interval=int(gateway_config.get("heartbeat_interval", 30)),
reconnect_timeout=int(gateway_config.get("reconnect_timeout", 60)),
ssl_cert=_substitute(gateway_config.get("ssl_cert")),
ssl_key=_substitute(gateway_config.get("ssl_key")),
)
logger.info(f"Gateway config loaded from {config_path}")
return cls(config=config)
def __init__(
self,
host: str = "127.0.0.1",
@@ -437,9 +497,36 @@ class WebSocketGateway:
except Exception as e:
logger.error(f"Error sending to client {client_id}: {e}")
def register_agent(self, agent: "Agent", agent_id: Optional[str] = None) -> str:
"""Register an agent with the gateway."""
def register_agent(
self,
agent: "Agent",
agent_id: Optional[str] = None,
*,
overwrite: bool = True,
) -> str:
"""Register an agent with the gateway.
Args:
agent: The agent instance to register
agent_id: Optional custom ID (auto-generated if not provided)
overwrite: If False, raises ValueError on duplicate ID (default: True for backward compat)
Returns:
The agent ID used for registration
Raises:
ValueError: If overwrite=False and agent_id already exists
"""
aid = agent_id or getattr(agent, "agent_id", None) or str(uuid.uuid4())
# B10: Handle duplicate registration
if aid in self._agents:
if not overwrite:
raise ValueError(
f"Agent '{aid}' already registered. Use overwrite=True to replace."
)
logger.warning(f"Overwriting existing agent: {aid}")
self._agents[aid] = agent
logger.info(f"Agent registered: {aid}")
return aid
@@ -460,6 +547,83 @@ class WebSocketGateway:
"""List all registered agent IDs."""
return list(self._agents.keys())
# ── Client Management (Public API) ────────────────────────────────
def add_client(self, client_id: str, websocket: Any) -> None:
"""Register a client WebSocket connection.
Args:
client_id: Unique client identifier
websocket: WebSocket connection object
"""
self._clients[client_id] = websocket
logger.debug(f"Client added: {client_id}")
def remove_client(self, client_id: str) -> bool:
"""Unregister a client connection.
Args:
client_id: The client ID to remove
Returns:
True if client was found and removed, False otherwise
"""
removed = self._clients.pop(client_id, None) is not None
if removed:
logger.debug(f"Client removed: {client_id}")
return removed
def get_client(self, client_id: str) -> Optional[Any]:
"""Get a client WebSocket by ID.
Args:
client_id: The client ID to look up
Returns:
The WebSocket connection or None if not found
"""
return self._clients.get(client_id)
def list_clients(self) -> List[str]:
"""List all connected client IDs.
Returns:
List of client IDs
"""
return list(self._clients.keys())
# ── Channel Bot Management (Public API - B7) ─────────────────────────
def list_channel_bots(self) -> List[str]:
"""List all registered channel bot names.
Returns:
List of channel bot names (e.g., ['discord', 'telegram'])
"""
return list(self._channel_bots.keys())
def get_channel_bot(self, name: str) -> Optional[Any]:
"""Get a channel bot by name.
Args:
name: The channel bot name (e.g., 'discord', 'telegram')
Returns:
The bot instance or None if not found
"""
return self._channel_bots.get(name)
def has_channel_bot(self, name: str) -> bool:
"""Check if a channel bot is registered.
Args:
name: The channel bot name
Returns:
True if the bot exists, False otherwise
"""
return name in self._channel_bots
def create_session(
self,
agent_id: str,
@@ -651,22 +815,76 @@ class WebSocketGateway:
return _resolve(raw)
def _create_agents_from_config(self, agents_cfg: Dict[str, Dict[str, Any]]) -> None:
"""Create and register Agent instances from the agents section of gateway.yaml."""
"""Create and register Agent instances from the agents section of gateway.yaml.
Supports all agent configuration options including:
- tools: List of tool names to resolve via ToolResolver
- tool_choice: Tool selection mode ('auto', 'required', 'none')
- reflection: Enable reflection/interactive mode (default: True)
- allow_delegation: Allow task delegation
- role: Agent role (CrewAI-style)
- goal: Agent goal (CrewAI-style)
- backstory: Agent backstory (CrewAI-style)
"""
from praisonaiagents import Agent
# G1: Resolve tool names to callables (same pattern as agents_generator)
tool_resolver = None
try:
from praisonai.tool_resolver import ToolResolver
tool_resolver = ToolResolver()
except ImportError:
logger.debug("ToolResolver not available, agents will have no tools")
for agent_id, agent_def in agents_cfg.items():
instructions = agent_def.get("instructions", "")
model = agent_def.get("model", None)
memory = agent_def.get("memory", False)
# G4: Support role/goal/backstory for CrewAI-style agents
role = agent_def.get("role", None)
goal = agent_def.get("goal", None)
backstory = agent_def.get("backstory", None)
# G1: Resolve tools from YAML config
agent_tools = []
yaml_tool_names = agent_def.get("tools", [])
if yaml_tool_names and tool_resolver:
for tool_name in yaml_tool_names:
if not tool_name or not isinstance(tool_name, str):
continue
tool_name = tool_name.strip()
resolved = tool_resolver.resolve(tool_name)
if resolved:
agent_tools.append(resolved)
logger.debug(f"Resolved tool '{tool_name}' for agent '{agent_id}'")
else:
logger.warning(f"Tool '{tool_name}' not found for agent '{agent_id}'")
# Additional agent options from YAML
tool_choice = agent_def.get("tool_choice", None)
reflection = agent_def.get("reflection", True) # Default: enable reflection/interactive mode
allow_delegation = agent_def.get("allow_delegation", False)
agent = Agent(
name=agent_id,
instructions=instructions,
llm=model,
memory=memory,
tools=agent_tools if agent_tools else None,
reflection=reflection,
allow_delegation=allow_delegation,
role=role,
goal=goal,
backstory=backstory,
)
# Store tool_choice for later use in chat()
if tool_choice:
agent._yaml_tool_choice = tool_choice
self.register_agent(agent, agent_id=agent_id)
logger.info(f"Created agent '{agent_id}' (model={model})")
logger.info(f"Created agent '{agent_id}' (model={model}, tools={len(agent_tools)}, reflection={reflection})")
def _determine_routing_context(
self, channel_type: str, message_metadata: Dict[str, Any]
+1 -1
View File
@@ -1 +1 @@
__version__ = "4.5.15"
__version__ = "4.5.32"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=1.5.15",
"praisonaiagents>=1.5.32",
"python-dotenv>=0.19.0",
"litellm>=1.81.0",
"PyYAML>=6.0",
Loaded 100 of 107 files, more files were not shown because too many files have changed in this diff. Show more