> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Queue System
> Queue management for PraisonAI TUI
The TUI queue holds multiple agent tasks with priority, cancel, and retry controls.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
U[Input] --> A[Agent]
A --> O[Output]
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
class A agent
class U,O tool
```
# 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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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())
```
```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
praisonai queue ls
praisonai queue cancel abc123
```
## Python Usage
### Basic Queue Operations
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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](/docs/features/tui/commands) for complete CLI reference.
```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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
Run `praisonai queue stats` to spot stuck or failing jobs before users report delays.
Use `queue retry` with the job id after fixing the root cause — avoid blind retries in a loop.
`queue clear --force` wipes pending work — confirm no production jobs remain first.
Long-running agent tasks should enqueue rather than block the TUI event loop.
## Related
Queue CLI reference
Terminal user interface