Message Queue

Queue messages while the AI agent is processing

Message Queue

PraisonAI CLI’s Interactive Mode supports message queuing, allowing you to type new prompts while the AI agent is still processing a previous task. Messages are queued and executed sequentially as each task completes.

This feature is inspired by Claude Code, Windsurf Cascade, Cursor, and Gemini CLI.

Overview

!Message Queue Demo

!Message Queue for Async Tasks

The message queue system provides:

Non-blocking input – Type new messages while agent is processing
FIFO processing – Messages are processed in order (First In, First Out)
Visual indicators – See queue status and pending messages
Queue management – View, clear, or remove queued messages
Thread-safe – Safe concurrent access from multiple threads

Quick Start

bash

# Start interactive mode
praisonai chat

# While the agent is processing, type more messages
# They will be queued and processed in order

How It Works

┌─────────────────────────────────────────────────────────────────┐
│                    Interactive Mode                              │
├─────────────────────────────────────────────────────────────────┤
│  ┌─────────────┐    ┌──────────────┐    ┌─────────────────┐    │
│  │ User Input  │───▶│ MessageQueue │───▶│ Agent Processing│    │
│  │             │    │   (FIFO)     │    │                 │    │
│  └─────────────┘    └──────────────┘    └─────────────────┘    │
│         │                  │                     │              │
│         │                  ▼                     │              │
│         │          ┌──────────────┐              │              │
│         └─────────▶│ Queue Display│◀─────────────┘              │
│                    │  (visual UI) │                             │
│                    └──────────────┘                             │
└─────────────────────────────────────────────────────────────────┘

User types a message – If agent is idle, it processes immediately
Agent is busy – Message is added to the queue
Agent completes – Next message in queue is automatically processed
Queue is empty – Agent returns to idle state

Queue Commands

Command	Description
`/queue`	Show all queued messages
`/queue clear`	Clear the entire queue
`/queue remove N`	Remove message at index N

Viewing the Queue

bash

❯ /queue

Queued Messages (3):
  0. ↳ Refactor the authentication module
  1. ↳ Add unit tests for the new feature
  2. ↳ Update the README with new instructions

Use /queue clear to clear, /queue remove N to remove

Clearing the Queue

bash

❯ /queue clear
✓ Cleared 3 queued message(s)

Removing a Specific Message

bash

❯ /queue remove 1
✓ Removed: Add unit tests for the new feature...

Live Status Display

While the agent is processing, a live status panel shows real-time updates:

bash

❯ Create a Python function to calculate fibonacci
╭──────────────────────────────────────╮
│ ⏳ Calling LLM...                    │
│ 📋 Queued: 2                         │
╰──────────────────────────────────────╯

The status updates as the agent progresses:

⏳ Thinking... – Initial processing
⏳ Creating agent... – Setting up the agent
⏳ Calling LLM... – Making the API call

Visual Indicators

The queue system provides visual feedback:

Indicator	Meaning
`⏳ Processing...`	Agent is currently processing a task
`📋 Queued (N)`	N messages are waiting in the queue
`↳ message`	A queued message (with truncation for long messages)
`🔧 tool_name`	Tool being executed
`💻 command`	Shell command being run

Processing States

The agent can be in one of three states:

State	Description
`IDLE`	Ready to process new messages immediately
`PROCESSING`	Currently working on a task
`WAITING_APPROVAL`	Waiting for user approval (e.g., file write)

Example Workflow

bash

# Start interactive mode
praisonai chat

# Send first task
❯ Create a Python function to calculate fibonacci numbers

# While agent is processing, queue more tasks
❯ Add docstrings to the function
❯ Create unit tests for edge cases
❯ Write a README explaining usage

# Check the queue
❯ /queue
⏳ Processing...
📋 Queued (3)

Queued Messages (3):
  0. ↳ Add docstrings to the function
  1. ↳ Create unit tests for edge cases
  2. ↳ Write a README explaining usage

# Agent will process each task in order automatically

Programmatic Usage

You can also use the message queue programmatically:

python

from praisonai.cli.features.message_queue import (
    MessageQueue,
    StateManager,
    QueueDisplay,
    ProcessingState,
    MessageQueueHandler
)

# Create a queue
queue = MessageQueue()

# Add messages
queue.add("First task")
queue.add("Second task")
queue.add("Third task")

# Check queue status
print(f"Queue count: {queue.count}")  # 3
print(f"Is empty: {queue.is_empty}")  # False

# View all messages
messages = queue.get_all()
print(messages)  # ['First task', 'Second task', 'Third task']

# Pop first message (FIFO)
first = queue.pop()
print(first)  # 'First task'

# Peek without removing
next_msg = queue.peek()
print(next_msg)  # 'Second task'

# Remove at specific index
removed = queue.remove_at(1)
print(removed)  # 'Third task'

# Clear all
queue.clear()

Using the Handler

python

from praisonai.cli.features.message_queue import MessageQueueHandler, ProcessingState

# Create handler with a processor function
def my_processor(message):
    # Process the message (e.g., send to LLM)
    return f"Processed: {message}"

handler = MessageQueueHandler(processor=my_processor)

# Submit when idle - processes immediately
handler.submit("First task")

# Simulate processing state
handler.state_manager.set_state(ProcessingState.PROCESSING)

# Submit while processing - queued
handler.submit("Second task")
handler.submit("Third task")

# Check status
status = handler.get_status()
print(status)
# {'queue_count': 2, 'state': 'processing', 'messages': ['Second task', 'Third task']}

# When processing completes, call this to process queue
handler.on_processing_complete()

Visual Display

python

from praisonai.cli.features.message_queue import (
    MessageQueue, StateManager, QueueDisplay, ProcessingState
)

queue = MessageQueue()
queue.add("Task 1")
queue.add("Task 2")

state = StateManager()
state.set_state(ProcessingState.PROCESSING)

display = QueueDisplay(queue, state_manager=state)

# Format for display
print(display.format_status())      # ⏳ Processing...
print(display.format_queue_count()) # 📋 Queued (2)
print(display.format_queue())       # ↳ Task 1\n↳ Task 2

API Reference

MessageQueue

Method	Description
`add(message)`	Add message to queue (returns False if empty)
`pop()`	Remove and return first message (FIFO)
`peek()`	View first message without removing
`clear()`	Remove all messages
`get_all()`	Get all messages as list
`remove_at(index)`	Remove message at specific index
`is_empty`	Property: True if queue is empty
`count`	Property: Number of messages in queue

StateManager

Method/Property	Description
`current_state`	Get current ProcessingState
`is_idle`	True if state is IDLE
`is_processing`	True if state is PROCESSING
`set_state(state)`	Set new state (triggers callback)

QueueDisplay

Method	Description
`format_queue()`	Format queued messages with ↳ prefix
`format_status()`	Format processing status indicator
`format_queue_count()`	Format queue count indicator

MessageQueueHandler

Method	Description
`submit(message)`	Submit message (process or queue)
`on_processing_complete()`	Called when processing finishes
`get_status()`	Get queue count, state, messages
`clear_queue()`	Clear all queued messages

Thread Safety

The message queue is thread-safe and uses threading.Lock for all operations. This ensures safe concurrent access when:

User input thread adds messages
Processing thread pops messages
Display thread reads queue status

Performance

The message queue is designed for minimal performance impact:

Lazy loading – Module only loaded when interactive mode starts
Simple data structure – Python list with O(1) append, O(n) pop(0)
No external dependencies – Uses only Python standard library
Minimal memory – Stores only message strings

Comparison with Other Tools

Feature	PraisonAI	Claude Code	Windsurf	Cursor
Queue messages	✅	✅	✅	✅
View queue	✅ `/queue`	✅	✅	✅
Clear queue	✅ `/queue clear`	✅	✅ Delete	✅
Remove specific	✅ `/queue remove N`	❌	✅ Delete	❌
Visual indicators	✅	✅	✅	✅
FIFO processing	✅	✅	✅	✅

Async Processing Classes

The message queue includes additional classes for async processing:

AsyncProcessor

Runs work functions in background threads:

python

from praisonai.cli.features.message_queue import AsyncProcessor

processor = AsyncProcessor()

def on_complete(result):
    print(f"Done: {result}")

def on_status(status):
    print(f"Status: {status}")

# Start background processing
processor.start(
    work_fn=lambda: "result",
    on_complete=on_complete,
    on_status=on_status
)

# Check if running
print(processor.is_running)  # True/False

LiveStatusDisplay

Tracks and displays real-time status:

python

from praisonai.cli.features.message_queue import LiveStatusDisplay

display = LiveStatusDisplay()

# Update status
display.update_status("Thinking...")
display.update_status("Calling LLM...")

# Track tool calls
display.add_tool_call("read_file", {"path": "main.py"})
display.add_command_execution("ls -la")

# Get formatted status
print(display.format_live_status())
# ⏳ Calling LLM...
# 🔧 read_file
# 💻 ls -la

# Clear when done
display.clear()

NonBlockingInput

Manages async user input:

python

from praisonai.cli.features.message_queue import NonBlockingInput

input_handler = NonBlockingInput()

# Submit input (from another thread)
input_handler.submit("user message")

# Check for pending input
if input_handler.has_input():
    msg = input_handler.get_input()
    print(f"Got: {msg}")

Related Features

Interactive TUI

Full interactive terminal interface

Slash Commands

All available slash commands

Cost Tracking

Monitor token usage and costs

Session Management

Save and restore sessions