mirror of
https://github.com/MervinPraison/PraisonAI.git
synced 2026-11-09 03:39:37 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3d26cedd5d | ||
|
|
8c735a60b2 |
No files matched your search
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.69" \
|
||||
"praisonai>=2.3.71" \
|
||||
"praisonai[chat]" \
|
||||
"embedchain[github,youtube]"
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.69" \
|
||||
"praisonai>=2.3.71" \
|
||||
"praisonai[ui]" \
|
||||
"praisonai[chat]" \
|
||||
"praisonai[realtime]" \
|
||||
|
||||
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
|
||||
# Install Python packages (using latest versions)
|
||||
RUN pip install --no-cache-dir \
|
||||
praisonai_tools \
|
||||
"praisonai>=2.3.69" \
|
||||
"praisonai>=2.3.71" \
|
||||
"praisonai[ui]" \
|
||||
"praisonai[crewai]"
|
||||
|
||||
|
||||
@@ -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/v2.3.69.tar.gz"
|
||||
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.69.tar.gz | shasum -a 256`.split.first
|
||||
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.71.tar.gz"
|
||||
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.71.tar.gz | shasum -a 256`.split.first
|
||||
license "MIT"
|
||||
|
||||
depends_on "[email protected]"
|
||||
|
||||
@@ -47,10 +47,14 @@ class AgentSchedulerHandler:
|
||||
return AgentSchedulerHandler._handle_describe(unknown_args, state_manager, daemon_manager)
|
||||
elif subcommand == "save":
|
||||
return AgentSchedulerHandler._handle_save(unknown_args, state_manager)
|
||||
elif subcommand == "stop-all":
|
||||
return AgentSchedulerHandler._handle_stop_all(state_manager, daemon_manager)
|
||||
elif subcommand == "stats":
|
||||
return AgentSchedulerHandler._handle_stats(state_manager, unknown_args, daemon_manager)
|
||||
else:
|
||||
print(f"❌ Unknown subcommand: {subcommand}")
|
||||
print("\nAvailable commands:")
|
||||
print(" start, list, stop, logs, restart, delete, describe, save")
|
||||
print(" start, list, stop, logs, restart, delete, describe, save, stop-all, stats")
|
||||
return 1
|
||||
|
||||
@staticmethod
|
||||
@@ -126,7 +130,8 @@ class AgentSchedulerHandler:
|
||||
print(f" Logs: ~/.praisonai/logs/{name}.log")
|
||||
print(f"\nManage:")
|
||||
print(f" praisonai schedule list")
|
||||
print(f" praisonai schedule logs {name} -f")
|
||||
print(f" praisonai schedule describe {name}")
|
||||
print(f" praisonai schedule logs {name} --follow")
|
||||
print(f" praisonai schedule stop {name}")
|
||||
|
||||
return 0
|
||||
@@ -163,6 +168,94 @@ class AgentSchedulerHandler:
|
||||
print(f"\nTotal: {len(states)} scheduler(s)")
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def _handle_stats(state_manager, unknown_args=None, daemon_manager=None) -> int:
|
||||
"""Handle 'schedule stats' command - show aggregate or individual statistics."""
|
||||
# If a name is provided, show individual stats (alias for describe)
|
||||
if unknown_args and len(unknown_args) > 0:
|
||||
name = unknown_args[0]
|
||||
return AgentSchedulerHandler._handle_describe([name], state_manager, daemon_manager)
|
||||
|
||||
# Otherwise show aggregate stats
|
||||
states = state_manager.list_all()
|
||||
|
||||
if not states:
|
||||
print("No schedulers found.")
|
||||
print("\nStart one with:")
|
||||
print(' praisonai schedule start <name> "Your task" --interval hourly')
|
||||
return 0
|
||||
|
||||
# Calculate aggregate stats
|
||||
total_schedulers = len(states)
|
||||
running = sum(1 for s in states if s.get('status') == 'running')
|
||||
stopped = total_schedulers - running
|
||||
total_executions = sum(s.get('executions', 0) for s in states)
|
||||
total_cost = sum(s.get('cost', 0.0) for s in states)
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📊 Aggregate Scheduler Statistics")
|
||||
print(f"{'='*60}")
|
||||
print(f"Total Schedulers: {total_schedulers}")
|
||||
print(f" 🟢 Running: {running}")
|
||||
print(f" 🔴 Stopped: {stopped}")
|
||||
print(f"\nTotal Executions: {total_executions}")
|
||||
print(f"Total Cost: ${total_cost:.4f}")
|
||||
|
||||
if total_executions > 0:
|
||||
avg_cost = total_cost / total_executions
|
||||
print(f"Avg Cost/Execution: ${avg_cost:.4f}")
|
||||
|
||||
print(f"{'='*60}\n")
|
||||
|
||||
# Show per-scheduler breakdown
|
||||
print("Per-Scheduler Breakdown:")
|
||||
print(f"{'Name':<20} {'Status':<10} {'Executions':<12} {'Cost':<12}")
|
||||
print("="*60)
|
||||
|
||||
for state in sorted(states, key=lambda x: x.get('executions', 0), reverse=True):
|
||||
name = state['name'][:20]
|
||||
status = state.get('status', 'unknown')
|
||||
executions = state.get('executions', 0)
|
||||
cost = state.get('cost', 0.0)
|
||||
|
||||
status_icon = "🟢" if status == "running" else "🔴"
|
||||
print(f"{name:<20} {status_icon} {status:<8} {executions:<12} ${cost:<11.4f}")
|
||||
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def _handle_stop_all(state_manager, daemon_manager) -> int:
|
||||
"""Handle 'schedule stop-all' command."""
|
||||
states = state_manager.list_all()
|
||||
|
||||
if not states:
|
||||
print("No schedulers running.")
|
||||
return 0
|
||||
|
||||
print(f"\n🛑 Stopping all schedulers ({len(states)} total)...\n")
|
||||
|
||||
stopped = 0
|
||||
failed = 0
|
||||
|
||||
for state in states:
|
||||
name = state['name']
|
||||
pid = state['pid']
|
||||
|
||||
try:
|
||||
if daemon_manager.stop_daemon(pid):
|
||||
state_manager.delete_state(name)
|
||||
print(f"✅ Stopped '{name}' (PID: {pid})")
|
||||
stopped += 1
|
||||
else:
|
||||
print(f"❌ Failed to stop '{name}' (PID: {pid})")
|
||||
failed += 1
|
||||
except Exception as e:
|
||||
print(f"❌ Error stopping '{name}': {e}")
|
||||
failed += 1
|
||||
|
||||
print(f"\n📊 Summary: {stopped} stopped, {failed} failed")
|
||||
return 0 if failed == 0 else 1
|
||||
|
||||
@staticmethod
|
||||
def _handle_stop(unknown_args, state_manager, daemon_manager) -> int:
|
||||
"""Handle 'schedule stop' command."""
|
||||
@@ -482,41 +575,33 @@ class AgentSchedulerHandler:
|
||||
print(' praisonai schedule "Check news every hour" --interval hourly')
|
||||
return 1
|
||||
|
||||
task_prompt = first_arg
|
||||
print(f"🤖 Creating scheduler for task: {task_prompt[:60]}...")
|
||||
task = first_arg
|
||||
|
||||
# Create a simple agent with the prompt
|
||||
# Create agent
|
||||
agent = Agent(
|
||||
name="Scheduled Agent",
|
||||
role="Task Executor",
|
||||
goal=task_prompt,
|
||||
instructions=f"Execute this task: {task_prompt}",
|
||||
verbose=verbose
|
||||
goal=task,
|
||||
instructions=task,
|
||||
verbose=False
|
||||
)
|
||||
|
||||
# Create scheduler
|
||||
scheduler = AgentScheduler(
|
||||
agent=agent,
|
||||
task=task_prompt,
|
||||
task=task,
|
||||
timeout=timeout_override,
|
||||
max_cost=max_cost_override
|
||||
)
|
||||
|
||||
# Display configuration
|
||||
agent_name = getattr(scheduler.agent, 'name', 'Agent')
|
||||
print(f"\n{'='*60}")
|
||||
print(f"🤖 Starting 24/7 {agent_name}")
|
||||
print(f"{'='*60}")
|
||||
print(f"Task: {scheduler.task[:80]}{'...' if len(scheduler.task) > 80 else ''}")
|
||||
# Get configuration
|
||||
interval = interval_override
|
||||
max_retries = max_retries_override
|
||||
|
||||
# Get schedule info
|
||||
if is_yaml_mode:
|
||||
schedule_config = scheduler._yaml_schedule_config
|
||||
interval = interval_override or schedule_config.get('interval', 'hourly')
|
||||
max_retries = max_retries_override or schedule_config.get('max_retries', 3)
|
||||
else:
|
||||
interval = interval_override
|
||||
max_retries = max_retries_override
|
||||
|
||||
print(f"Schedule: {interval}")
|
||||
print(f"Max Retries: {max_retries}")
|
||||
|
||||
@@ -0,0 +1,729 @@
|
||||
"""
|
||||
Agent Scheduler CLI Feature.
|
||||
|
||||
Provides CLI commands for scheduling agents to run 24/7 at regular intervals.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AgentSchedulerHandler:
|
||||
"""Handler for agent scheduler CLI commands."""
|
||||
|
||||
@staticmethod
|
||||
def handle_daemon_command(subcommand: str, args, unknown_args=None) -> int:
|
||||
"""
|
||||
Handle daemon management commands (start, list, stop, logs, restart).
|
||||
|
||||
Args:
|
||||
subcommand: Command to execute (start, list, stop, logs, restart)
|
||||
args: Parsed command-line arguments
|
||||
unknown_args: Additional arguments
|
||||
|
||||
Returns:
|
||||
Exit code
|
||||
"""
|
||||
from praisonai.scheduler.state_manager import SchedulerStateManager
|
||||
from praisonai.scheduler.daemon_manager import DaemonManager
|
||||
|
||||
state_manager = SchedulerStateManager()
|
||||
daemon_manager = DaemonManager()
|
||||
|
||||
if subcommand == "start":
|
||||
return AgentSchedulerHandler._handle_start(args, unknown_args, state_manager, daemon_manager)
|
||||
elif subcommand == "list":
|
||||
return AgentSchedulerHandler._handle_list(state_manager)
|
||||
elif subcommand == "stop":
|
||||
return AgentSchedulerHandler._handle_stop(unknown_args, state_manager, daemon_manager)
|
||||
elif subcommand == "logs":
|
||||
return AgentSchedulerHandler._handle_logs(unknown_args, daemon_manager)
|
||||
elif subcommand == "restart":
|
||||
return AgentSchedulerHandler._handle_restart(unknown_args, state_manager, daemon_manager)
|
||||
elif subcommand == "delete":
|
||||
return AgentSchedulerHandler._handle_delete(unknown_args, state_manager)
|
||||
elif subcommand == "describe":
|
||||
return AgentSchedulerHandler._handle_describe(unknown_args, state_manager, daemon_manager)
|
||||
elif subcommand == "save":
|
||||
return AgentSchedulerHandler._handle_save(unknown_args, state_manager)
|
||||
elif subcommand == "stop-all":
|
||||
return AgentSchedulerHandler._handle_stop_all(state_manager, daemon_manager)
|
||||
elif subcommand == "stats":
|
||||
return AgentSchedulerHandler._handle_stats(state_manager, unknown_args, daemon_manager)
|
||||
else:
|
||||
print(f"❌ Unknown subcommand: {subcommand}")
|
||||
print("\nAvailable commands:")
|
||||
print(" start, list, stop, logs, restart, delete, describe, save, stop-all, stats")
|
||||
return 1
|
||||
|
||||
@staticmethod
|
||||
def _handle_start(args, unknown_args, state_manager, daemon_manager) -> int:
|
||||
"""Handle 'schedule start' command."""
|
||||
from datetime import datetime
|
||||
|
||||
if not unknown_args:
|
||||
print("❌ Error: Please provide scheduler name and task")
|
||||
print("\nUsage:")
|
||||
print(' praisonai schedule start <name> "Your task" --interval hourly')
|
||||
print("\nExample:")
|
||||
print(' praisonai schedule start news-checker "Check AI news" --interval hourly')
|
||||
return 1
|
||||
|
||||
# Parse arguments
|
||||
name = unknown_args[0]
|
||||
task = unknown_args[1] if len(unknown_args) > 1 else None
|
||||
|
||||
if not task:
|
||||
print("❌ Error: Please provide a task")
|
||||
return 1
|
||||
|
||||
# Get options
|
||||
interval = getattr(args, 'schedule_interval', None) or 'hourly'
|
||||
max_retries = getattr(args, 'schedule_max_retries', None) or 3
|
||||
timeout = getattr(args, 'timeout', None)
|
||||
max_cost = getattr(args, 'max_cost', None)
|
||||
|
||||
# Check if name already exists
|
||||
existing = state_manager.load_state(name)
|
||||
if existing and daemon_manager.get_status(existing.get('pid', 0))['is_alive']:
|
||||
print(f"❌ Error: Scheduler '{name}' is already running (PID: {existing['pid']})")
|
||||
print(f" Use 'praisonai schedule stop {name}' to stop it first")
|
||||
return 1
|
||||
|
||||
# Start daemon
|
||||
print(f"🚀 Starting scheduler '{name}'...")
|
||||
print(f" Task: {task}")
|
||||
print(f" Interval: {interval}")
|
||||
if timeout:
|
||||
print(f" Timeout: {timeout}s")
|
||||
if max_cost:
|
||||
print(f" Budget: ${max_cost}")
|
||||
|
||||
pid = daemon_manager.start_scheduler_daemon(
|
||||
name=name,
|
||||
task=task,
|
||||
interval=interval,
|
||||
max_cost=max_cost,
|
||||
timeout=timeout,
|
||||
max_retries=max_retries
|
||||
)
|
||||
|
||||
# Save state
|
||||
state = {
|
||||
"name": name,
|
||||
"pid": pid,
|
||||
"task": task,
|
||||
"interval": interval,
|
||||
"timeout": timeout,
|
||||
"max_cost": max_cost,
|
||||
"max_retries": max_retries,
|
||||
"status": "running",
|
||||
"started_at": datetime.now().isoformat(),
|
||||
"executions": 0,
|
||||
"cost": 0.0
|
||||
}
|
||||
state_manager.save_state(name, state)
|
||||
|
||||
print(f"\n✅ Scheduler '{name}' started successfully!")
|
||||
print(f" PID: {pid}")
|
||||
print(f" Logs: ~/.praisonai/logs/{name}.log")
|
||||
print(f"\nManage:")
|
||||
print(f" praisonai schedule list")
|
||||
print(f" praisonai schedule describe {name}")
|
||||
print(f" praisonai schedule logs {name} --follow")
|
||||
print(f" praisonai schedule stop {name}")
|
||||
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def _handle_list(state_manager) -> int:
|
||||
"""Handle 'schedule list' command."""
|
||||
states = state_manager.list_all()
|
||||
|
||||
if not states:
|
||||
print("No schedulers running.")
|
||||
print("\nStart one with:")
|
||||
print(' praisonai schedule start <name> "Your task" --interval hourly')
|
||||
return 0
|
||||
|
||||
# Clean up dead processes
|
||||
state_manager.cleanup_dead_processes()
|
||||
states = state_manager.list_all()
|
||||
|
||||
# Display table
|
||||
print(f"\n{'Name':<20} {'Status':<10} {'PID':<8} {'Interval':<12} {'Task':<40}")
|
||||
print("=" * 90)
|
||||
|
||||
for state in states:
|
||||
name = state.get('name', 'unknown')[:20]
|
||||
pid = state.get('pid', 0)
|
||||
status = "running" if state_manager.is_process_alive(pid) else "stopped"
|
||||
interval = state.get('interval', 'unknown')[:12]
|
||||
task = state.get('task', '')[:40]
|
||||
|
||||
status_icon = "🟢" if status == "running" else "🔴"
|
||||
print(f"{name:<20} {status_icon} {status:<8} {pid:<8} {interval:<12} {task:<40}")
|
||||
|
||||
print(f"\nTotal: {len(states)} scheduler(s)")
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def _handle_stats(state_manager, unknown_args=None, daemon_manager=None) -> int:
|
||||
"""Handle 'schedule stats' command - show aggregate or individual statistics."""
|
||||
# If a name is provided, show individual stats (alias for describe)
|
||||
if unknown_args and len(unknown_args) > 0:
|
||||
name = unknown_args[0]
|
||||
return AgentSchedulerHandler._handle_describe([name], state_manager, None)
|
||||
|
||||
# Otherwise show aggregate stats
|
||||
states = state_manager.list_all()
|
||||
|
||||
if not states:
|
||||
print("No schedulers found.")
|
||||
print("\nStart one with:")
|
||||
print(' praisonai schedule start <name> "Your task" --interval hourly')
|
||||
return 0
|
||||
|
||||
# Calculate aggregate stats
|
||||
total_schedulers = len(states)
|
||||
running = sum(1 for s in states if s.get('status') == 'running')
|
||||
stopped = total_schedulers - running
|
||||
total_executions = sum(s.get('executions', 0) for s in states)
|
||||
total_cost = sum(s.get('cost', 0.0) for s in states)
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📊 Aggregate Scheduler Statistics")
|
||||
print(f"{'='*60}")
|
||||
print(f"Total Schedulers: {total_schedulers}")
|
||||
print(f" 🟢 Running: {running}")
|
||||
print(f" 🔴 Stopped: {stopped}")
|
||||
print(f"\nTotal Executions: {total_executions}")
|
||||
print(f"Total Cost: ${total_cost:.4f}")
|
||||
|
||||
if total_executions > 0:
|
||||
avg_cost = total_cost / total_executions
|
||||
print(f"Avg Cost/Execution: ${avg_cost:.4f}")
|
||||
|
||||
print(f"{'='*60}\n")
|
||||
|
||||
# Show per-scheduler breakdown
|
||||
print("Per-Scheduler Breakdown:")
|
||||
print(f"{'Name':<20} {'Status':<10} {'Executions':<12} {'Cost':<12}")
|
||||
print("="*60)
|
||||
|
||||
for state in sorted(states, key=lambda x: x.get('executions', 0), reverse=True):
|
||||
name = state['name'][:20]
|
||||
status = state.get('status', 'unknown')
|
||||
executions = state.get('executions', 0)
|
||||
cost = state.get('cost', 0.0)
|
||||
|
||||
status_icon = "🟢" if status == "running" else "🔴"
|
||||
print(f"{name:<20} {status_icon} {status:<8} {executions:<12} ${cost:<11.4f}")
|
||||
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def _handle_stop_all(state_manager, daemon_manager) -> int:
|
||||
"""Handle 'schedule stop-all' command."""
|
||||
states = state_manager.list_all()
|
||||
|
||||
if not states:
|
||||
print("No schedulers running.")
|
||||
return 0
|
||||
|
||||
print(f"\n🛑 Stopping all schedulers ({len(states)} total)...\n")
|
||||
|
||||
stopped = 0
|
||||
failed = 0
|
||||
|
||||
for state in states:
|
||||
name = state['name']
|
||||
pid = state['pid']
|
||||
|
||||
try:
|
||||
if daemon_manager.stop_daemon(pid):
|
||||
state_manager.delete_state(name)
|
||||
print(f"✅ Stopped '{name}' (PID: {pid})")
|
||||
stopped += 1
|
||||
else:
|
||||
print(f"❌ Failed to stop '{name}' (PID: {pid})")
|
||||
failed += 1
|
||||
except Exception as e:
|
||||
print(f"❌ Error stopping '{name}': {e}")
|
||||
failed += 1
|
||||
|
||||
print(f"\n📊 Summary: {stopped} stopped, {failed} failed")
|
||||
return 0 if failed == 0 else 1
|
||||
|
||||
@staticmethod
|
||||
def _handle_stop(unknown_args, state_manager, daemon_manager) -> int:
|
||||
"""Handle 'schedule stop' command."""
|
||||
if not unknown_args:
|
||||
print("❌ Error: Please provide scheduler name")
|
||||
print("\nUsage: praisonai schedule stop <name>")
|
||||
return 1
|
||||
|
||||
name = unknown_args[0]
|
||||
state = state_manager.load_state(name)
|
||||
|
||||
if not state:
|
||||
print(f"❌ Error: Scheduler '{name}' not found")
|
||||
print("\nList schedulers with: praisonai schedule list")
|
||||
return 1
|
||||
|
||||
pid = state.get('pid')
|
||||
if not pid:
|
||||
print(f"❌ Error: No PID found for scheduler '{name}'")
|
||||
return 1
|
||||
|
||||
print(f"🛑 Stopping scheduler '{name}' (PID: {pid})...")
|
||||
|
||||
success = daemon_manager.stop_daemon(pid)
|
||||
|
||||
if success:
|
||||
state['status'] = 'stopped'
|
||||
state_manager.save_state(name, state)
|
||||
print(f"✅ Scheduler '{name}' stopped successfully")
|
||||
return 0
|
||||
else:
|
||||
print(f"❌ Failed to stop scheduler '{name}'")
|
||||
return 1
|
||||
|
||||
@staticmethod
|
||||
def _handle_logs(unknown_args, daemon_manager) -> int:
|
||||
"""Handle 'schedule logs' command."""
|
||||
if not unknown_args:
|
||||
print("❌ Error: Please provide scheduler name")
|
||||
print("\nUsage: praisonai schedule logs <name> [-f]")
|
||||
return 1
|
||||
|
||||
name = unknown_args[0]
|
||||
follow = '-f' in unknown_args or '--follow' in unknown_args
|
||||
|
||||
if follow:
|
||||
print(f"📋 Following logs for '{name}' (Ctrl+C to stop)...")
|
||||
import subprocess
|
||||
log_file = daemon_manager.log_dir / f"{name}.log"
|
||||
if not log_file.exists():
|
||||
print(f"❌ Error: Log file not found for '{name}'")
|
||||
return 1
|
||||
|
||||
try:
|
||||
subprocess.run(['tail', '-f', str(log_file)])
|
||||
except KeyboardInterrupt:
|
||||
print("\n✅ Stopped following logs")
|
||||
return 0
|
||||
else:
|
||||
logs = daemon_manager.read_logs(name, lines=50)
|
||||
if logs:
|
||||
print(f"📋 Last 50 lines of logs for '{name}':\n")
|
||||
print(logs)
|
||||
return 0
|
||||
else:
|
||||
print(f"❌ Error: No logs found for '{name}'")
|
||||
return 1
|
||||
|
||||
@staticmethod
|
||||
def _handle_restart(unknown_args, state_manager, daemon_manager) -> int:
|
||||
"""Handle 'schedule restart' command."""
|
||||
from datetime import datetime
|
||||
import time
|
||||
|
||||
if not unknown_args:
|
||||
print("❌ Error: Please provide scheduler name")
|
||||
print("\nUsage: praisonai schedule restart <name>")
|
||||
return 1
|
||||
|
||||
name = unknown_args[0]
|
||||
state = state_manager.load_state(name)
|
||||
|
||||
if not state:
|
||||
print(f"❌ Error: Scheduler '{name}' not found")
|
||||
return 1
|
||||
|
||||
print(f"🔄 Restarting scheduler '{name}'...")
|
||||
|
||||
# Stop if running
|
||||
pid = state.get('pid')
|
||||
if pid and state_manager.is_process_alive(pid):
|
||||
daemon_manager.stop_daemon(pid)
|
||||
time.sleep(1)
|
||||
|
||||
# Start again
|
||||
new_pid = daemon_manager.start_scheduler_daemon(
|
||||
name=name,
|
||||
task=state['task'],
|
||||
interval=state['interval'],
|
||||
max_cost=state.get('max_cost'),
|
||||
timeout=state.get('timeout'),
|
||||
max_retries=state.get('max_retries', 3)
|
||||
)
|
||||
|
||||
state['pid'] = new_pid
|
||||
state['status'] = 'running'
|
||||
state['started_at'] = datetime.now().isoformat()
|
||||
state_manager.save_state(name, state)
|
||||
|
||||
print(f"✅ Scheduler '{name}' restarted (PID: {new_pid})")
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def _handle_delete(unknown_args, state_manager) -> int:
|
||||
"""Handle 'schedule delete' command."""
|
||||
if not unknown_args:
|
||||
print("❌ Error: Please provide scheduler name")
|
||||
print("\nUsage: praisonai schedule delete <name>")
|
||||
return 1
|
||||
|
||||
name = unknown_args[0]
|
||||
|
||||
if state_manager.delete_state(name):
|
||||
print(f"✅ Scheduler '{name}' deleted from list")
|
||||
return 0
|
||||
else:
|
||||
print(f"❌ Error: Scheduler '{name}' not found")
|
||||
return 1
|
||||
|
||||
@staticmethod
|
||||
def _handle_describe(unknown_args, state_manager, daemon_manager) -> int:
|
||||
"""Handle 'schedule describe' command."""
|
||||
from datetime import datetime
|
||||
|
||||
if not unknown_args:
|
||||
print("❌ Error: Please provide scheduler name")
|
||||
print("\nUsage: praisonai schedule describe <name>")
|
||||
return 1
|
||||
|
||||
name = unknown_args[0]
|
||||
state = state_manager.load_state(name)
|
||||
|
||||
if not state:
|
||||
print(f"❌ Error: Scheduler '{name}' not found")
|
||||
print("\nList schedulers with: praisonai schedule list")
|
||||
return 1
|
||||
|
||||
# Get process status
|
||||
pid = state.get('pid', 0)
|
||||
is_alive = state_manager.is_process_alive(pid)
|
||||
status = "🟢 running" if is_alive else "🔴 stopped"
|
||||
|
||||
# Calculate uptime
|
||||
started_at = state.get('started_at')
|
||||
uptime = "N/A"
|
||||
if started_at:
|
||||
try:
|
||||
start_time = datetime.fromisoformat(started_at)
|
||||
uptime_delta = datetime.now() - start_time
|
||||
hours = int(uptime_delta.total_seconds() // 3600)
|
||||
minutes = int((uptime_delta.total_seconds() % 3600) // 60)
|
||||
uptime = f"{hours}h {minutes}m"
|
||||
except:
|
||||
pass
|
||||
|
||||
# Display detailed info
|
||||
print(f"\n{'='*60}")
|
||||
print(f"📋 Scheduler Details: {name}")
|
||||
print(f"{'='*60}")
|
||||
print(f"Status: {status}")
|
||||
print(f"PID: {pid}")
|
||||
print(f"Uptime: {uptime}")
|
||||
print(f"Task: {state.get('task', 'N/A')}")
|
||||
print(f"Interval: {state.get('interval', 'N/A')}")
|
||||
print(f"Max Retries: {state.get('max_retries', 'N/A')}")
|
||||
|
||||
if state.get('timeout'):
|
||||
print(f"Timeout: {state['timeout']}s")
|
||||
|
||||
if state.get('max_cost'):
|
||||
print(f"Budget: ${state['max_cost']}")
|
||||
|
||||
print(f"Executions: {state.get('executions', 0)}")
|
||||
print(f"Total Cost: ${state.get('cost', 0.0):.4f}")
|
||||
print(f"Started: {started_at or 'N/A'}")
|
||||
|
||||
# Log file location
|
||||
log_file = daemon_manager.log_dir / f"{name}.log"
|
||||
print(f"Logs: {log_file}")
|
||||
|
||||
print(f"{'='*60}\n")
|
||||
|
||||
return 0
|
||||
|
||||
@staticmethod
|
||||
def _handle_save(unknown_args, state_manager) -> int:
|
||||
"""Handle 'schedule save' command."""
|
||||
import yaml
|
||||
|
||||
if not unknown_args:
|
||||
print("❌ Error: Please provide scheduler name")
|
||||
print("\nUsage: praisonai schedule save <name> [output.yaml]")
|
||||
return 1
|
||||
|
||||
name = unknown_args[0]
|
||||
output_file = unknown_args[1] if len(unknown_args) > 1 else f"{name}.yaml"
|
||||
|
||||
state = state_manager.load_state(name)
|
||||
|
||||
if not state:
|
||||
print(f"❌ Error: Scheduler '{name}' not found")
|
||||
return 1
|
||||
|
||||
# Create YAML config from state
|
||||
yaml_config = {
|
||||
'framework': 'praisonai',
|
||||
'agents': [{
|
||||
'name': name,
|
||||
'role': 'Task Executor',
|
||||
'goal': state.get('task', ''),
|
||||
'instructions': state.get('task', ''),
|
||||
'verbose': True
|
||||
}],
|
||||
'task': state.get('task', ''),
|
||||
'schedule': {
|
||||
'interval': state.get('interval', 'hourly'),
|
||||
'max_retries': state.get('max_retries', 3),
|
||||
'run_immediately': True
|
||||
}
|
||||
}
|
||||
|
||||
# Add optional fields
|
||||
if state.get('timeout'):
|
||||
yaml_config['schedule']['timeout'] = state['timeout']
|
||||
|
||||
if state.get('max_cost'):
|
||||
yaml_config['schedule']['max_cost'] = state['max_cost']
|
||||
|
||||
# Write to file
|
||||
try:
|
||||
with open(output_file, 'w') as f:
|
||||
yaml.dump(yaml_config, f, default_flow_style=False, sort_keys=False)
|
||||
|
||||
print(f"✅ Configuration saved to: {output_file}")
|
||||
print(f"\nRun with:")
|
||||
print(f" praisonai schedule {output_file}")
|
||||
return 0
|
||||
except Exception as e:
|
||||
print(f"❌ Error saving configuration: {e}")
|
||||
return 1
|
||||
|
||||
@staticmethod
|
||||
def handle_schedule_command(args, unknown_args=None) -> int:
|
||||
"""
|
||||
Handle the schedule command for running agents periodically.
|
||||
|
||||
Supports two modes:
|
||||
1. YAML mode: praisonai schedule agents.yaml
|
||||
2. Prompt mode: praisonai schedule "Your task here" --interval hourly
|
||||
|
||||
Args:
|
||||
args: Parsed command-line arguments
|
||||
unknown_args: Additional arguments after the command
|
||||
|
||||
Returns:
|
||||
Exit code (0 for success, 1 for error)
|
||||
"""
|
||||
try:
|
||||
from praisonai.scheduler import AgentScheduler
|
||||
from praisonaiagents import Agent
|
||||
except ImportError:
|
||||
print("Error: praisonai.scheduler or praisonaiagents module not found")
|
||||
print("Please ensure PraisonAI is properly installed")
|
||||
print("pip install praisonai praisonaiagents")
|
||||
return 1
|
||||
|
||||
# Check if first arg is a YAML file or a prompt
|
||||
first_arg = unknown_args[0] if unknown_args else None
|
||||
is_yaml_mode = first_arg and (first_arg.endswith('.yaml') or first_arg.endswith('.yml'))
|
||||
|
||||
# Get overrides from CLI
|
||||
interval_override = getattr(args, 'schedule_interval', None) or 'hourly'
|
||||
max_retries_override = getattr(args, 'schedule_max_retries', None) or 3
|
||||
timeout_override = getattr(args, 'timeout', None)
|
||||
max_cost_override = getattr(args, 'max_cost', None)
|
||||
verbose = getattr(args, 'verbose', False)
|
||||
|
||||
# Set up logging - only show logs if verbose
|
||||
if verbose:
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='[%(asctime)s] %(levelname)s - %(message)s',
|
||||
datefmt='%Y-%m-%d %H:%M:%S'
|
||||
)
|
||||
else:
|
||||
# Suppress scheduler logs in non-verbose mode
|
||||
logging.getLogger('praisonai.scheduler').setLevel(logging.WARNING)
|
||||
|
||||
try:
|
||||
if is_yaml_mode:
|
||||
# YAML mode: Load from agents.yaml
|
||||
yaml_path = first_arg
|
||||
print(f"🤖 Loading agent configuration from: {yaml_path}")
|
||||
scheduler = AgentScheduler.from_yaml(
|
||||
yaml_path=yaml_path,
|
||||
interval_override=interval_override,
|
||||
max_retries_override=max_retries_override,
|
||||
timeout_override=timeout_override,
|
||||
max_cost_override=max_cost_override
|
||||
)
|
||||
else:
|
||||
# Prompt mode: Create agent from direct prompt
|
||||
if not first_arg:
|
||||
print("❌ Error: Please provide either a YAML file or a task prompt")
|
||||
print("\nExamples:")
|
||||
print(" praisonai schedule agents.yaml")
|
||||
print(' praisonai schedule "Check news every hour" --interval hourly')
|
||||
return 1
|
||||
|
||||
task = first_arg
|
||||
|
||||
# Create agent
|
||||
agent = Agent(
|
||||
name="Scheduled Agent",
|
||||
role="Task Executor",
|
||||
goal=task,
|
||||
instructions=task,
|
||||
verbose=False
|
||||
)
|
||||
|
||||
# Create scheduler
|
||||
scheduler = AgentScheduler(
|
||||
agent=agent,
|
||||
task=task,
|
||||
timeout=timeout_override,
|
||||
max_cost=max_cost_override
|
||||
)
|
||||
|
||||
# Get configuration
|
||||
interval = interval_override
|
||||
max_retries = max_retries_override
|
||||
|
||||
if is_yaml_mode:
|
||||
schedule_config = scheduler._yaml_schedule_config
|
||||
interval = interval_override or schedule_config.get('interval', 'hourly')
|
||||
max_retries = max_retries_override or schedule_config.get('max_retries', 3)
|
||||
|
||||
print(f"Schedule: {interval}")
|
||||
print(f"Max Retries: {max_retries}")
|
||||
if scheduler.timeout:
|
||||
print(f"Timeout: {scheduler.timeout}s")
|
||||
if scheduler.max_cost:
|
||||
print(f"Budget: ${scheduler.max_cost}")
|
||||
print(f"{'='*60}\n")
|
||||
|
||||
# Start scheduler
|
||||
print("⏰ Starting scheduler... (Press Ctrl+C to stop)\n")
|
||||
if is_yaml_mode:
|
||||
scheduler.start_from_yaml_config()
|
||||
else:
|
||||
scheduler.start(
|
||||
schedule_expr=interval,
|
||||
max_retries=max_retries,
|
||||
run_immediately=True
|
||||
)
|
||||
|
||||
# Keep running until interrupted
|
||||
try:
|
||||
while scheduler.is_running:
|
||||
import time
|
||||
time.sleep(1)
|
||||
except KeyboardInterrupt:
|
||||
print("\n\n🛑 Stopping scheduler...")
|
||||
scheduler.stop()
|
||||
|
||||
# Display final stats
|
||||
stats = scheduler.get_stats()
|
||||
print("\n📊 Final Statistics:")
|
||||
print(f" Total Executions: {stats['total_executions']}")
|
||||
print(f" Successful: {stats['successful_executions']}")
|
||||
print(f" Failed: {stats['failed_executions']}")
|
||||
print(f" Success Rate: {stats['success_rate']:.1f}%")
|
||||
print("\n✅ Agent stopped successfully\n")
|
||||
|
||||
return 0
|
||||
|
||||
except FileNotFoundError as e:
|
||||
print(f"❌ Error: {e}")
|
||||
print(f"\nMake sure {yaml_path} exists in the current directory.")
|
||||
print("\nExample agents.yaml structure:")
|
||||
print("""
|
||||
framework: praisonai
|
||||
|
||||
agents:
|
||||
- name: "AI News Monitor"
|
||||
role: "News Analyst"
|
||||
instructions: "Search and summarize AI news"
|
||||
tools:
|
||||
- search_tool
|
||||
|
||||
task: "Search for latest AI news"
|
||||
|
||||
schedule:
|
||||
interval: "hourly"
|
||||
max_retries: 3
|
||||
run_immediately: true
|
||||
""")
|
||||
return 1
|
||||
|
||||
except ValueError as e:
|
||||
print(f"❌ Configuration Error: {e}")
|
||||
return 1
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error: {e}")
|
||||
if verbose:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return 1
|
||||
|
||||
@staticmethod
|
||||
def add_schedule_arguments(subparsers):
|
||||
"""
|
||||
Add schedule subcommand and arguments to parser.
|
||||
|
||||
Args:
|
||||
subparsers: Subparsers object from argparse
|
||||
"""
|
||||
schedule_parser = subparsers.add_parser(
|
||||
'schedule',
|
||||
help='Schedule an agent to run continuously at regular intervals'
|
||||
)
|
||||
|
||||
schedule_parser.add_argument(
|
||||
'schedule_yaml',
|
||||
nargs='?',
|
||||
default='agents.yaml',
|
||||
help='Path to agents.yaml file (default: agents.yaml)'
|
||||
)
|
||||
|
||||
schedule_parser.add_argument(
|
||||
'--interval',
|
||||
dest='schedule_interval',
|
||||
type=str,
|
||||
help='Override schedule interval (e.g., "hourly", "*/30m", "daily")'
|
||||
)
|
||||
|
||||
schedule_parser.add_argument(
|
||||
'--max-retries',
|
||||
dest='schedule_max_retries',
|
||||
type=int,
|
||||
help='Override maximum retry attempts (default: from YAML or 3)'
|
||||
)
|
||||
|
||||
schedule_parser.add_argument(
|
||||
'--verbose', '-v',
|
||||
action='store_true',
|
||||
help='Enable verbose logging'
|
||||
)
|
||||
|
||||
schedule_parser.set_defaults(func=AgentSchedulerHandler.handle_schedule_command)
|
||||
|
||||
|
||||
def setup_scheduler_cli(subparsers):
|
||||
"""
|
||||
Setup scheduler CLI commands.
|
||||
|
||||
Args:
|
||||
subparsers: Subparsers object from argparse
|
||||
"""
|
||||
AgentSchedulerHandler.add_schedule_arguments(subparsers)
|
||||
@@ -236,7 +236,7 @@ class PraisonAI:
|
||||
# Check for subcommands (start, list, stop, logs, restart)
|
||||
subcommand = unknown_args[0] if unknown_args and not unknown_args[0].startswith('-') else None
|
||||
|
||||
if subcommand in ['start', 'list', 'stop', 'logs', 'restart', 'delete', 'describe', 'save']:
|
||||
if subcommand in ['start', 'list', 'stop', 'logs', 'restart', 'delete', 'describe', 'save', 'stop-all', 'stats']:
|
||||
exit_code = AgentSchedulerHandler.handle_daemon_command(subcommand, args, unknown_args[1:] if len(unknown_args) > 1 else [])
|
||||
else:
|
||||
# Legacy mode: direct scheduling (foreground)
|
||||
|
||||
@@ -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==2.3.69 gunicorn markdown\n")
|
||||
file.write("RUN pip install flask praisonai==2.3.71 gunicorn markdown\n")
|
||||
file.write("EXPOSE 8080\n")
|
||||
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
|
||||
|
||||
|
||||
@@ -100,16 +100,16 @@ class AgentScheduler:
|
||||
self.is_running = True
|
||||
self._stop_event.clear()
|
||||
|
||||
logger.info(f"Starting agent scheduler: {getattr(self.agent, 'name', 'Agent')}")
|
||||
logger.info(f"Task: {self.task}")
|
||||
logger.info(f"Schedule: {schedule_expr} ({interval}s interval)")
|
||||
logger.debug(f"Starting agent scheduler: {getattr(self.agent, 'name', 'Agent')}")
|
||||
logger.debug(f"Task: {self.task}")
|
||||
logger.debug(f"Schedule: {schedule_expr} ({interval}s interval)")
|
||||
self.is_running = True
|
||||
self._stop_event.clear()
|
||||
self._start_time = datetime.now()
|
||||
|
||||
# Run immediately if requested
|
||||
if run_immediately:
|
||||
logger.info("Running agent immediately before starting schedule...")
|
||||
logger.debug("Running agent immediately before starting schedule...")
|
||||
self._execute_with_retry(max_retries)
|
||||
|
||||
self._thread = threading.Thread(
|
||||
@@ -119,7 +119,7 @@ class AgentScheduler:
|
||||
)
|
||||
self._thread.start()
|
||||
|
||||
logger.info("Agent scheduler started successfully")
|
||||
logger.debug("Agent scheduler started successfully")
|
||||
if self.timeout:
|
||||
logger.info(f"Timeout per execution: {self.timeout}s")
|
||||
if self.max_cost:
|
||||
@@ -139,18 +139,18 @@ class AgentScheduler:
|
||||
True if stopped successfully
|
||||
"""
|
||||
if not self.is_running:
|
||||
logger.info("Scheduler is not running")
|
||||
logger.debug("Scheduler is not running")
|
||||
return True
|
||||
|
||||
logger.info("Stopping agent scheduler...")
|
||||
logger.debug("Stopping agent scheduler...")
|
||||
self._stop_event.set()
|
||||
|
||||
if self._thread and self._thread.is_alive():
|
||||
self._thread.join(timeout=10)
|
||||
|
||||
self.is_running = False
|
||||
logger.info("Agent scheduler stopped")
|
||||
logger.info(f"Execution stats - Total: {self._execution_count}, Success: {self._success_count}, Failed: {self._failure_count}")
|
||||
logger.debug("Agent scheduler stopped")
|
||||
logger.debug(f"Execution stats - Total: {self._execution_count}, Success: {self._success_count}, Failed: {self._failure_count}")
|
||||
return True
|
||||
|
||||
def get_stats(self) -> Dict[str, Any]:
|
||||
@@ -219,15 +219,15 @@ class AgentScheduler:
|
||||
self.stop()
|
||||
break
|
||||
|
||||
logger.info(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Starting scheduled agent execution")
|
||||
logger.debug(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Starting scheduled agent execution")
|
||||
|
||||
self._execute_with_retry(max_retries)
|
||||
|
||||
# Wait for next scheduled time
|
||||
logger.info(f"Next execution in {interval} seconds ({interval/3600:.1f} hours)")
|
||||
logger.debug(f"Next execution in {interval} seconds ({interval/3600:.1f} hours)")
|
||||
if self.max_cost:
|
||||
remaining = self.max_cost - self._total_cost
|
||||
logger.info(f"Budget remaining: ${remaining:.4f}")
|
||||
logger.debug(f"Budget remaining: ${remaining:.4f}")
|
||||
self._stop_event.wait(interval)
|
||||
|
||||
def _execute_with_retry(self, max_retries: int):
|
||||
@@ -238,7 +238,7 @@ class AgentScheduler:
|
||||
|
||||
for attempt in range(max_retries):
|
||||
try:
|
||||
logger.info(f"Attempt {attempt + 1}/{max_retries}")
|
||||
logger.debug(f"Attempt {attempt + 1}/{max_retries}")
|
||||
|
||||
# Execute with timeout if specified
|
||||
if self.timeout:
|
||||
@@ -260,8 +260,8 @@ class AgentScheduler:
|
||||
else:
|
||||
result = self._executor.execute(self.task)
|
||||
|
||||
logger.info(f"Agent execution successful on attempt {attempt + 1}")
|
||||
logger.info(f"Result: {result}")
|
||||
logger.debug(f"Agent execution successful on attempt {attempt + 1}")
|
||||
logger.debug(f"Result: {result}")
|
||||
|
||||
# Always print result to stdout (even in non-verbose mode)
|
||||
print(f"\n✅ Agent Response:\n{result}\n")
|
||||
@@ -269,7 +269,7 @@ class AgentScheduler:
|
||||
# Estimate cost (rough: ~$0.0001 per execution for gpt-4o-mini)
|
||||
estimated_cost = 0.0001 # Base cost estimate
|
||||
self._total_cost += estimated_cost
|
||||
logger.info(f"Estimated cost this run: ${estimated_cost:.4f}, Total: ${self._total_cost:.4f}")
|
||||
logger.debug(f"Estimated cost this run: ${estimated_cost:.4f}, Total: ${self._total_cost:.4f}")
|
||||
|
||||
self._success_count += 1
|
||||
success = True
|
||||
@@ -290,7 +290,7 @@ class AgentScheduler:
|
||||
|
||||
if attempt < max_retries - 1:
|
||||
wait_time = 30 * (attempt + 1) # Exponential backoff
|
||||
logger.info(f"Waiting {wait_time}s before retry...")
|
||||
logger.debug(f"Waiting {wait_time}s before retry...")
|
||||
time.sleep(wait_time)
|
||||
|
||||
if not success:
|
||||
@@ -310,10 +310,10 @@ class AgentScheduler:
|
||||
Returns:
|
||||
Agent execution result
|
||||
"""
|
||||
logger.info("Executing agent once")
|
||||
logger.debug("Executing agent once")
|
||||
try:
|
||||
result = self._executor.execute(self.task)
|
||||
logger.info(f"One-time execution successful: {result}")
|
||||
logger.debug(f"One-time execution successful: {result}")
|
||||
|
||||
if self.on_success:
|
||||
try:
|
||||
|
||||
@@ -52,12 +52,6 @@ class DaemonManager:
|
||||
|
||||
# Open log file
|
||||
with open(log_file, 'a') as log:
|
||||
log.write(f"\n{'='*60}\n")
|
||||
log.write(f"Starting daemon: {name}\n")
|
||||
log.write(f"Task: {task}\n")
|
||||
log.write(f"Interval: {interval}\n")
|
||||
log.write(f"Time: {datetime.now().isoformat()}\n")
|
||||
log.write(f"{'='*60}\n\n")
|
||||
log.flush()
|
||||
|
||||
# Start process as daemon
|
||||
|
||||
@@ -1 +1 @@
|
||||
__version__ = "2.3.69"
|
||||
__version__ = "2.3.71"
|
||||
@@ -0,0 +1,288 @@
|
||||
"""
|
||||
Tests for new scheduler features: stop-all, stats, and clean logs.
|
||||
"""
|
||||
import pytest
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
|
||||
|
||||
class TestStopAllCommand:
|
||||
"""Test stop-all command functionality."""
|
||||
|
||||
def test_stop_all_with_multiple_schedulers(self):
|
||||
"""Test stopping all schedulers at once."""
|
||||
from praisonai.cli.features.agent_scheduler import AgentSchedulerHandler
|
||||
from praisonai.scheduler.state_manager import SchedulerStateManager
|
||||
from praisonai.scheduler.daemon_manager import DaemonManager
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
state_dir = Path(tmpdir) / "schedulers"
|
||||
log_dir = Path(tmpdir) / "logs"
|
||||
|
||||
state_manager = SchedulerStateManager(state_dir)
|
||||
daemon_manager = DaemonManager(log_dir)
|
||||
|
||||
# Create mock states for 3 schedulers
|
||||
for i in range(1, 4):
|
||||
state_manager.save_state(f"test-{i}", {
|
||||
"name": f"test-{i}",
|
||||
"pid": 1000 + i,
|
||||
"task": f"Task {i}",
|
||||
"interval": "*/10s",
|
||||
"status": "running"
|
||||
})
|
||||
|
||||
# Mock daemon_manager.stop_daemon to return True
|
||||
with patch.object(daemon_manager, 'stop_daemon', return_value=True):
|
||||
result = AgentSchedulerHandler._handle_stop_all(state_manager, daemon_manager)
|
||||
|
||||
assert result == 0
|
||||
# All states should be deleted
|
||||
assert len(state_manager.list_all()) == 0
|
||||
|
||||
def test_stop_all_with_no_schedulers(self):
|
||||
"""Test stop-all when no schedulers are running."""
|
||||
from praisonai.cli.features.agent_scheduler import AgentSchedulerHandler
|
||||
from praisonai.scheduler.state_manager import SchedulerStateManager
|
||||
from praisonai.scheduler.daemon_manager import DaemonManager
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
state_dir = Path(tmpdir) / "schedulers"
|
||||
log_dir = Path(tmpdir) / "logs"
|
||||
|
||||
state_manager = SchedulerStateManager(state_dir)
|
||||
daemon_manager = DaemonManager(log_dir)
|
||||
|
||||
result = AgentSchedulerHandler._handle_stop_all(state_manager, daemon_manager)
|
||||
|
||||
assert result == 0
|
||||
|
||||
def test_stop_all_with_failures(self):
|
||||
"""Test stop-all when some schedulers fail to stop."""
|
||||
from praisonai.cli.features.agent_scheduler import AgentSchedulerHandler
|
||||
from praisonai.scheduler.state_manager import SchedulerStateManager
|
||||
from praisonai.scheduler.daemon_manager import DaemonManager
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
state_dir = Path(tmpdir) / "schedulers"
|
||||
log_dir = Path(tmpdir) / "logs"
|
||||
|
||||
state_manager = SchedulerStateManager(state_dir)
|
||||
daemon_manager = DaemonManager(log_dir)
|
||||
|
||||
# Create mock states
|
||||
state_manager.save_state("test-1", {
|
||||
"name": "test-1",
|
||||
"pid": 1001,
|
||||
"task": "Task 1",
|
||||
"interval": "*/10s",
|
||||
"status": "running"
|
||||
})
|
||||
state_manager.save_state("test-2", {
|
||||
"name": "test-2",
|
||||
"pid": 1002,
|
||||
"task": "Task 2",
|
||||
"interval": "*/10s",
|
||||
"status": "running"
|
||||
})
|
||||
|
||||
# Mock stop_daemon to fail for second scheduler
|
||||
def mock_stop(pid):
|
||||
return pid == 1001
|
||||
|
||||
with patch.object(daemon_manager, 'stop_daemon', side_effect=mock_stop):
|
||||
result = AgentSchedulerHandler._handle_stop_all(state_manager, daemon_manager)
|
||||
|
||||
assert result == 1 # Should return 1 when failures occur
|
||||
|
||||
|
||||
class TestStatsCommand:
|
||||
"""Test stats command functionality."""
|
||||
|
||||
def test_stats_aggregate(self):
|
||||
"""Test aggregate stats for all schedulers."""
|
||||
from praisonai.cli.features.agent_scheduler import AgentSchedulerHandler
|
||||
from praisonai.scheduler.state_manager import SchedulerStateManager
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
state_dir = Path(tmpdir) / "schedulers"
|
||||
state_manager = SchedulerStateManager(state_dir)
|
||||
|
||||
# Create mock states with execution data
|
||||
state_manager.save_state("test-1", {
|
||||
"name": "test-1",
|
||||
"pid": 1001,
|
||||
"task": "Task 1",
|
||||
"interval": "*/10s",
|
||||
"status": "running",
|
||||
"executions": 5,
|
||||
"cost": 0.0005
|
||||
})
|
||||
state_manager.save_state("test-2", {
|
||||
"name": "test-2",
|
||||
"pid": 1002,
|
||||
"task": "Task 2",
|
||||
"interval": "*/20s",
|
||||
"status": "stopped",
|
||||
"executions": 3,
|
||||
"cost": 0.0003
|
||||
})
|
||||
|
||||
result = AgentSchedulerHandler._handle_stats(state_manager)
|
||||
|
||||
assert result == 0
|
||||
|
||||
def test_stats_with_no_schedulers(self):
|
||||
"""Test stats when no schedulers exist."""
|
||||
from praisonai.cli.features.agent_scheduler import AgentSchedulerHandler
|
||||
from praisonai.scheduler.state_manager import SchedulerStateManager
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
state_dir = Path(tmpdir) / "schedulers"
|
||||
state_manager = SchedulerStateManager(state_dir)
|
||||
|
||||
result = AgentSchedulerHandler._handle_stats(state_manager)
|
||||
|
||||
assert result == 0
|
||||
|
||||
def test_stats_individual_scheduler(self):
|
||||
"""Test stats command as alias for describe."""
|
||||
from praisonai.cli.features.agent_scheduler import AgentSchedulerHandler
|
||||
from praisonai.scheduler.state_manager import SchedulerStateManager
|
||||
from praisonai.scheduler.daemon_manager import DaemonManager
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
state_dir = Path(tmpdir) / "schedulers"
|
||||
log_dir = Path(tmpdir) / "logs"
|
||||
|
||||
state_manager = SchedulerStateManager(state_dir)
|
||||
daemon_manager = DaemonManager(log_dir)
|
||||
|
||||
# Create mock state
|
||||
state_manager.save_state("test-1", {
|
||||
"name": "test-1",
|
||||
"pid": 1001,
|
||||
"task": "Task 1",
|
||||
"interval": "*/10s",
|
||||
"status": "running",
|
||||
"executions": 5,
|
||||
"cost": 0.0005,
|
||||
"started_at": "2025-12-22T10:00:00"
|
||||
})
|
||||
|
||||
result = AgentSchedulerHandler._handle_stats(
|
||||
state_manager,
|
||||
unknown_args=["test-1"],
|
||||
daemon_manager=daemon_manager
|
||||
)
|
||||
|
||||
assert result == 0
|
||||
|
||||
|
||||
class TestCleanLogs:
|
||||
"""Test clean log output (DEBUG level)."""
|
||||
|
||||
def test_scheduler_uses_debug_logging(self):
|
||||
"""Test that scheduler uses DEBUG level for internal logs."""
|
||||
from praisonai.scheduler.agent_scheduler import AgentScheduler
|
||||
from praisonaiagents import Agent
|
||||
import logging
|
||||
|
||||
# Create a test handler to capture log records
|
||||
test_handler = logging.Handler()
|
||||
test_handler.setLevel(logging.DEBUG)
|
||||
records = []
|
||||
|
||||
def capture_record(record):
|
||||
records.append(record)
|
||||
return True
|
||||
|
||||
test_handler.emit = lambda record: records.append(record)
|
||||
|
||||
# Get the scheduler logger
|
||||
logger = logging.getLogger('praisonai.scheduler')
|
||||
logger.addHandler(test_handler)
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
try:
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
role="Tester",
|
||||
goal="Test logging",
|
||||
instructions="Test"
|
||||
)
|
||||
|
||||
scheduler = AgentScheduler(
|
||||
agent=agent,
|
||||
task="Test task",
|
||||
timeout=5,
|
||||
max_cost=1.0
|
||||
)
|
||||
|
||||
# Start and immediately stop
|
||||
scheduler.start(schedule_expr="*/10s", max_retries=1, run_immediately=False)
|
||||
scheduler.stop()
|
||||
|
||||
# Check that debug messages were logged
|
||||
debug_messages = [r for r in records if r.levelno == logging.DEBUG]
|
||||
assert len(debug_messages) > 0, "Should have DEBUG level logs"
|
||||
|
||||
finally:
|
||||
logger.removeHandler(test_handler)
|
||||
|
||||
|
||||
class TestStateUpdatePersistence:
|
||||
"""Test that execution stats are persisted to state files."""
|
||||
|
||||
def test_state_update_after_execution(self):
|
||||
"""Test that _update_state_if_daemon updates the state file."""
|
||||
from praisonai.scheduler.agent_scheduler import AgentScheduler
|
||||
from praisonai.scheduler.state_manager import SchedulerStateManager
|
||||
from praisonaiagents import Agent
|
||||
import os
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
state_dir = Path(tmpdir) / "schedulers"
|
||||
state_dir.mkdir(parents=True)
|
||||
|
||||
# Create a mock state file
|
||||
state_manager = SchedulerStateManager(state_dir)
|
||||
pid = os.getpid()
|
||||
|
||||
state_manager.save_state("test-scheduler", {
|
||||
"name": "test-scheduler",
|
||||
"pid": pid,
|
||||
"task": "Test",
|
||||
"interval": "*/10s",
|
||||
"status": "running",
|
||||
"executions": 0,
|
||||
"cost": 0.0
|
||||
})
|
||||
|
||||
agent = Agent(
|
||||
name="Test Agent",
|
||||
role="Tester",
|
||||
goal="Test",
|
||||
instructions="Test"
|
||||
)
|
||||
|
||||
scheduler = AgentScheduler(
|
||||
agent=agent,
|
||||
task="Test task",
|
||||
timeout=5,
|
||||
max_cost=1.0
|
||||
)
|
||||
|
||||
# Simulate execution
|
||||
scheduler._execution_count = 5
|
||||
scheduler._total_cost = 0.0005
|
||||
|
||||
# Call update method
|
||||
scheduler._update_state_if_daemon()
|
||||
|
||||
# Check if state was updated
|
||||
state = state_manager.load_state("test-scheduler")
|
||||
if state: # State might be updated
|
||||
assert state.get('executions', 0) >= 0
|
||||
assert state.get('cost', 0.0) >= 0.0
|
||||
Reference in new issue
Block a user