Queue System

Queue management for PraisonAI TUI

The TUI queue holds multiple agent tasks with priority, cancel, and retry controls.

from praisonaiagents import Agent

agent = Agent(name="tui-assistant", instructions="Manage queued TUI jobs.")
agent.start("Run these three research tasks in order.")

The user submits several prompts; the queue runs them sequentially without blocking the input pane.

Queue System

The PraisonAI Queue System enables managing multiple agent runs with priority ordering, concurrency limits, and persistence.

Overview

The queue system provides:

• Priority-based FIFO - URGENT > HIGH > NORMAL > LOW ordering
• Concurrency limits - Global, per-agent, and per-workspace limits
• Cancel/retry - Full lifecycle management
• Persistence - SQLite-backed crash recovery
• Streaming - Token-by-token output with backpressure

Quick Start

1. Submit runs with priorities

   import asyncio
   from praisonai.cli.features.queue import QueueManager, QueueConfig, RunPriority
   
   async def main():
       manager = QueueManager(config=QueueConfig(max_concurrent_global=4))
       await manager.start()
       await manager.submit("Analyze the codebase", agent_name="Analyst", priority=RunPriority.HIGH)
       await manager.stop()
   
   asyncio.run(main())
   
2. Manage via CLI

   praisonai queue ls
   praisonai queue cancel abc123
   

Python Usage

Basic Queue Operations

import asyncio
from praisonai.cli.features.queue import (
    QueueManager,
    QueueConfig,
    RunPriority,
)

async def main():
    # Configure the queue
    config = QueueConfig(
        max_concurrent_global=4,
        max_concurrent_per_agent=2,
        max_queue_size=100,
        enable_persistence=True,
    )
    
    # Create manager with callbacks
    async def on_output(run_id: str, chunk: str):
        print(chunk, end="", flush=True)
    
    async def on_complete(run_id: str, run):
        print(f"\nRun {run_id} completed: {run.state.value}")
    
    manager = QueueManager(
        config=config,
        on_output=on_output,
        on_complete=on_complete,
    )
    
    # Start the manager
    await manager.start()
    
    # Submit runs with different priorities
    run1 = await manager.submit(
        input_content="Analyze the codebase",
        agent_name="Analyst",
        priority=RunPriority.HIGH,
    )
    
    run2 = await manager.submit(
        input_content="Write documentation",
        agent_name="Writer",
        priority=RunPriority.NORMAL,
    )
    
    # Wait for completion
    await asyncio.sleep(30)
    
    # Stop the manager
    await manager.stop()

asyncio.run(main())

Queue Configuration

from praisonai.cli.features.queue import QueueConfig, RunPriority

config = QueueConfig(
    # Concurrency limits
    max_concurrent_global=4,      # Max runs across all agents
    max_concurrent_per_agent=2,   # Max runs per agent type
    max_concurrent_per_workspace=4,  # Max runs per workspace
    
    # Queue limits
    max_queue_size=100,           # Max queued runs
    default_priority=RunPriority.NORMAL,
    
    # Persistence
    enable_persistence=True,
    db_path=".praison/queue.db",
    
    # Autosave
    autosave_interval=30.0,       # Seconds between autosaves
)

Run States

from praisonai.cli.features.queue import RunState

# Available states
RunState.QUEUED      # Waiting in queue
RunState.RUNNING     # Currently executing
RunState.PAUSED      # Paused (can resume)
RunState.SUCCEEDED   # Completed successfully
RunState.FAILED      # Failed with error
RunState.CANCELLED   # Cancelled by user

# Check state properties
state = RunState.RUNNING
state.is_terminal()  # False - still active
state.is_active()    # True - not finished

Priority Levels

from praisonai.cli.features.queue import RunPriority

# Priority ordering (highest to lowest)
RunPriority.URGENT   # 3 - Process immediately
RunPriority.HIGH     # 2 - High priority
RunPriority.NORMAL   # 1 - Default priority
RunPriority.LOW      # 0 - Background tasks

# Parse from string
priority = RunPriority.from_string("high")  # RunPriority.HIGH

Cancel and Retry

# Cancel a running task
success = await manager.cancel(run_id)

# Retry a failed task
new_run_id = await manager.retry(run_id)

# Check if run can be retried
run = manager.get_run(run_id)
if run.can_retry():
    await manager.retry(run_id)

Queue Statistics

# Get statistics
stats = manager.get_stats()

print(f"Queued: {stats.queued_count}")
print(f"Running: {stats.running_count}")
print(f"Succeeded: {stats.succeeded_count}")
print(f"Failed: {stats.failed_count}")
print(f"Total: {stats.total_runs}")
print(f"Avg wait: {stats.avg_wait_seconds:.1f}s")
print(f"Avg duration: {stats.avg_duration_seconds:.1f}s")

Data Model

QueuedRun

from praisonai.cli.features.queue import QueuedRun

run = QueuedRun(
    run_id="abc123",           # Unique identifier
    agent_name="Assistant",    # Agent to execute
    input_content="Hello",     # Input prompt
    state=RunState.QUEUED,     # Current state
    priority=RunPriority.NORMAL,
    
    # Attribution
    session_id="session123",
    trace_id="trace456",
    workspace="/path/to/workspace",
    
    # Retry configuration
    retry_count=0,
    max_retries=3,
    
    # Custom configuration
    config={"model": "gpt-4"},
)

# Properties
run.duration_seconds  # Time from start to end
run.wait_seconds      # Time from created to started
run.can_retry()       # Check if retryable

# Serialization
data = run.to_dict()
run2 = QueuedRun.from_dict(data)

Persistence

The queue system uses SQLite for persistence:

.praison/
├── queue.db           # Queue state, runs, messages
└── sessions/
    └── {session_id}/
        └── state.json # Session state

Crash Recovery

# Start with recovery enabled
await manager.start(recover=True)

# This will:
# 1. Load pending runs from database
# 2. Mark interrupted runs as failed
# 3. Re-queue recoverable runs

Manual Persistence Operations

from praisonai.cli.features.queue import QueuePersistence

persistence = QueuePersistence(db_path=".praison/queue.db")
persistence.initialize()

# Save/load runs
persistence.save_run(run)
loaded = persistence.load_run(run_id)

# List runs
runs = persistence.list_runs(state=RunState.QUEUED, limit=10)

# Cleanup old runs
deleted = persistence.cleanup_old_runs(days=30)

persistence.close()

CLI Usage

See TUI Commands for complete CLI reference.

# List queue
praisonai queue ls

# Cancel a run
praisonai queue cancel abc123

# Retry a failed run
praisonai queue retry abc123

# Show statistics
praisonai queue stats

# Clear queue
praisonai queue clear --force

Best Practices

Inspect queue stats regularly

Run praisonai queue stats to spot stuck or failing jobs before users report delays.

Retry failed runs deliberately

Use queue retry with the job id after fixing the root cause — avoid blind retries in a loop.

Clear with force only in dev

queue clear --force wipes pending work — confirm no production jobs remain first.

Size the queue for peak load

Long-running agent tasks should enqueue rather than block the TUI event loop.

Related

TUI Commands

Queue CLI reference

TUI Overview

Terminal user interface