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

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

# 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) │                             │
│                    └──────────────┘                             │
└─────────────────────────────────────────────────────────────────┘

1. User types a message - If agent is idle, it processes immediately
2. Agent is busy - Message is added to the queue
3. Agent completes - Next message in queue is automatically processed
4. 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

❯ /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

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

Removing a Specific Message

❯ /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:

❯ 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

# 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:

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

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

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:

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:

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:

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