Queue System Developer Guide

conv:{conversation_id}:messages    → User messages (WhatsApp)
conv:{conversation_id}:events      → System events (payments, orders)
conv:{conversation_id}:processing  → Processing lock (TTL: 2 minutes)

REDIS_URL=redis://localhost:6379  # Redis connection URL

# main.py - FastAPI endpoint
@app.post("/message")
async def handle_message(payload: ChatbotPayloadDTO):
    # Check if conversation is already processing
    if await queue_manager.is_processing(payload.conversation.id):
        # Add to message queue
        await queue_manager.add_message_to_queue(
            payload.conversation.id,
            payload
        )
        return {"status": "queued"}

    # Not processing - start new processing
    await conversation_processor.handle_new_message(
        payload.conversation.id,
        process_conversation_messages
    )
    return {"status": "processing"}

# main.py - FastAPI endpoint
@app.post("/nexus/event")
async def handle_event(event: NexusEvent):
    # Events always queue if messages are processing
    if await queue_manager.is_processing(event.conversation.id):
        await queue_manager.add_event_to_queue(
            event.conversation.id,
            event
        )
        return {"status": "queued"}

    # Process immediately if no messages
    await conversation_processor.handle_new_event(
        event.conversation.id,
        process_conversation_events
    )
    return {"status": "processing"}

# Simplified processing flow
async def process_conversation_messages(conversation_id: str):
    # Pop all queued messages
    messages = await queue_manager.pop_all_messages_from_queue(conversation_id)

    for batch in messages:
        # Create agent with queue helpers
        agent = ChatAgent()
        agent.set_queue_helpers(AgentQueueHelpers(queue_manager))

        # Process the message batch
        await agent.process_message(batch.get_payload())

class ChatAgent:
    def __init__(self):
        self.queue_helpers = None  # Automatically set by main.py

    def set_queue_helpers(self, queue_helpers: AgentQueueHelpers):
        """Called by main.py before processing each message"""
        self.queue_helpers = queue_helpers

# 1. has_new_messages() - Quick boolean check
if await self.queue_helpers.has_new_messages(conversation_id):
    # Returns True/False - doesn't tell you what the messages are
    print("User sent more messages!")

# 2. check_message_queue() - Look without taking (PEEK)
peeked = await self.queue_helpers.check_message_queue(conversation_id, count=5)
# Returns list of QueuedMessage objects
# Messages STAY in queue - you're just looking
for msg in peeked:
    content = msg.get_payload().lastMessage.content
    print(f"User said: {content}")

# 3. get_new_messages() - Take all messages (POP)
new_messages = await self.queue_helpers.get_new_messages(conversation_id)
# Returns list of QueuedMessage objects
# Messages are REMOVED from queue
# This is what you'll use 90% of the time

# 4. pop_message_from_queue() - Take just one message
single_msg = await self.queue_helpers.pop_message_from_queue(conversation_id)
# Rarely used - usually you want all messages

# Get messages from queue
messages = await self.queue_helpers.get_new_messages(conversation_id)

for queued_msg in messages:
    # Extract the actual ChatbotPayloadDTO
    payload = queued_msg.get_payload()

    # Now you can access all the usual fields
    user_text = payload.lastMessage.content
    user_phone = payload.customer.whatsappNumber
    timestamp = payload.lastMessage.timestamp

    # The payload has the same structure as your main process_message payload
    print(f"User {user_phone} said: {user_text}")

async def process_message(self, payload: ChatbotPayloadDTO):
    conversation_id = payload.conversation.id

    # Initial processing of "I want a pizza"
    intent = self.analyze_intent(payload)

    if intent == "create_order":
        # Extract initial order details
        order = self.build_order_from_message(payload)

        # CRITICAL: Check for new messages before creating order
        if self.queue_helpers:  # Always check if helpers are available
            new_messages = await self.queue_helpers.get_new_messages(conversation_id)

            if new_messages:
                # User sent "make it large" and "extra cheese" while we processed
                self.logger.info(f"Found {len(new_messages)} additional messages")

                # Merge all messages together
                all_messages = [payload] + [m.get_payload() for m in new_messages]

                # Rebuild order with ALL information
                order = self.build_order_from_messages(all_messages)

                # Let user know we saw everything
                await self.send_message(
                    f"Got it! I've included your {len(new_messages)} additional requests."
                )

        # Now create order with complete information
        await self.create_order(order)

async def handle_complex_order(self, initial_payload: ChatbotPayloadDTO):
    conversation_id = initial_payload.conversation.id
    context = {
        'messages': [initial_payload.lastMessage.content],
        'order_items': [],
        'special_requests': []
    }

    # Process initial message
    await self.process_initial_request(initial_payload, context)

    # Check for follow-up messages multiple times
    while context['status'] != 'complete':
        # Simulate some processing time (API calls, etc.)
        await self.call_some_api()

        # Check if user sent more messages
        new_messages = await self.queue_helpers.get_new_messages(conversation_id)

        if new_messages:
            for msg in new_messages:
                payload = msg.get_payload()
                user_text = payload.lastMessage.content.lower()

                # Add to our running context
                context['messages'].append(payload.lastMessage.content)

                # Check for special keywords
                if 'cancel' in user_text:
                    return await self.handle_cancellation()
                elif 'change' in user_text or 'actually' in user_text:
                    context['has_modifications'] = True
                elif any(size in user_text for size in ['small', 'medium', 'large']):
                    context['size_preference'] = self.extract_size(user_text)

            # Reprocess with new context
            await self.update_order_with_context(context)

        # Continue processing...
        context['status'] = self.check_if_complete(context)

async def execute_tool_call(self, tool_name: str, tool_args: dict, context: dict):
    conversation_id = context['conversation_id']

    # Special handling for order-related tools
    if tool_name in ['update_order_draft', 'create_order', 'send_order_form']:
        # Check for last-minute changes
        new_messages = await self.queue_helpers.get_new_messages(conversation_id)

        if new_messages:
            self.logger.info(f"Found {len(new_messages)} new messages before {tool_name}")

            # Extract any modifications from new messages
            for msg in new_messages:
                payload = msg.get_payload()
                content = payload.lastMessage.content

                # Quick parse for common modifications
                if 'no ' in content.lower():
                    # "no onions", "no ice", etc.
                    tool_args = self.apply_negation_to_order(tool_args, content)
                elif any(word in content.lower() for word in ['add', 'extra', 'more']):
                    # "add bacon", "extra cheese"
                    tool_args = self.apply_addition_to_order(tool_args, content)
                elif 'deliver' in content.lower() or 'pickup' in content.lower():
                    # Service type change
                    tool_args = self.update_service_type(tool_args, content)

            # Add note about modifications
            tool_args['customer_note'] = self.merge_customer_notes(
                tool_args.get('customer_note', ''),
                f"Modified based on {len(new_messages)} additional messages"
            )

    # Execute tool with potentially modified args
    return await self.nexus.execute_tool(tool_name, tool_args)

async def handle_rapid_messages(self, initial_payload: ChatbotPayloadDTO):
    conversation_id = initial_payload.conversation.id
    messages_buffer = [initial_payload.lastMessage.content]

    # Wait a brief moment for rapid-fire messages
    await asyncio.sleep(0.5)  # Small delay to let messages queue up

    # Get all queued messages
    new_messages = await self.queue_helpers.get_new_messages(conversation_id)

    if new_messages:
        # Add all to buffer
        for msg in new_messages:
            messages_buffer.append(msg.get_payload().lastMessage.content)

    # Combine related messages intelligently
    combined_intent = self.intelligent_message_combination(messages_buffer)

    # Example: ["I want pizza", "large", "pepperoni", "no wait", "hawaiian"]
    # Becomes: "I want a large hawaiian pizza" (recognized the correction)

    return await self.process_combined_intent(combined_intent)

def intelligent_message_combination(self, messages: List[str]) -> str:
    """Combine multiple messages into coherent intent"""
    # Handle corrections (last relevant message wins)
    if any('no wait' in m.lower() or 'actually' in m.lower() for m in messages):
        # Find the correction point and use messages after it
        correction_index = next(
            i for i, m in enumerate(messages)
            if 'no wait' in m.lower() or 'actually' in m.lower()
        )
        messages = messages[correction_index:]

    # Simple combination for now
    return ' '.join(messages)

# 1. Before ANY tool call that creates/modifies data
if tool_name in ['update_order_draft', 'create_order', 'send_order_form']:
    new_messages = await self.queue_helpers.get_new_messages(conversation_id)

# 2. After time-consuming operations
response = await self.call_menu_api()  # Takes 1-2 seconds
new_messages = await self.queue_helpers.get_new_messages(conversation_id)

# 3. Before sending confirmations or summaries
if "confirm" in response_text or "is this correct" in response_text:
    # User might have sent corrections
    new_messages = await self.queue_helpers.get_new_messages(conversation_id)

# 4. When conversation seems complete
if intent == "order_complete":
    # Last chance to catch "wait, one more thing..."
    final_messages = await self.queue_helpers.get_new_messages(conversation_id)

async def process_message(self, payload: ChatbotPayloadDTO):
    """Main entry point for message processing"""
    conversation_id = payload.conversation.id

    # Analyze what the user wants
    intent = self.extract_intent(payload)

    if intent == "order_food":
        # Start building the order
        order_context = {
            'items': self.extract_items(payload),
            'conversation_id': conversation_id
        }

        # Maybe we need to look up menu items (slow operation)
        menu_matches = await self.search_menu(order_context['items'])

        # CHECK POINT 1: After slow operation
        new_msgs = await self.queue_helpers.get_new_messages(conversation_id)
        if new_msgs:
            # User might have added "make it spicy" or "large size"
            for msg in new_msgs:
                additional_text = msg.get_payload().lastMessage.content
                order_context['modifications'] = self.extract_modifications(additional_text)

        # Prepare to create order
        order_data = self.build_order_data(menu_matches, order_context)

        # CHECK POINT 2: Before tool call
        final_msgs = await self.queue_helpers.get_new_messages(conversation_id)
        if final_msgs:
            # Last-second changes like "actually, cancel the drinks"
            order_data = self.apply_final_modifications(order_data, final_msgs)

        # Now safe to create order
        result = await self.tools.update_order_draft(order_data)

        return f"Great! I've added {len(order_data['items'])} items to your order."

# ❌ BAD: No queue check before order creation
async def create_order_bad(self, payload):
    order = self.parse_order(payload)
    return await self.tools.create_order(order)  # Misses "wait, add fries!"

# ✅ GOOD: Always check before critical actions
async def create_order_good(self, payload):
    order = self.parse_order(payload)

    # Check for last-minute additions
    new_messages = await self.queue_helpers.get_new_messages(payload.conversation.id)
    if new_messages:
        order = self.merge_order_updates(order, new_messages)

    return await self.tools.create_order(order)

# ❌ BAD: Checking after the action
async def process_order_bad(self, payload):
    result = await self.tools.create_order(order_data)  # Already created!

    # Too late - order already placed
    new_messages = await self.queue_helpers.get_new_messages(conversation_id)

# ✅ GOOD: Check before irreversible actions
async def process_order_good(self, payload):
    # Check BEFORE creating
    new_messages = await self.queue_helpers.get_new_messages(conversation_id)
    if new_messages:
        order_data = self.update_with_new_messages(order_data, new_messages)

    result = await self.tools.create_order(order_data)

# ❌ BAD: Treating each message independently
for msg in new_messages:
    content = msg.get_payload().lastMessage.content
    if "large" in content:
        size = "large"
    if "small" in content:
        size = "small"  # Oops, overwrote large!

# ✅ GOOD: Consider message relationships
messages_text = [m.get_payload().lastMessage.content for m in new_messages]
# ["make it large", "no wait", "small please"]
# Recognize "no wait" as a correction indicator
final_size = self.extract_final_intent(messages_text)  # Returns "small"

# Events are processed after all messages are done
if await self.queue_helpers.has_pending_events(conversation_id):
    events = await self.queue_helpers.pop_all_events(conversation_id)
    # Handle payment failures, order updates, etc.

# Your agent's thought process:
1. Receive: "I want pizza"
2. Think: "Let me look up pizzas..." (2 seconds pass)
3. CHECKPOINT: Check queue → Found: "large please"
4. Think: "Okay, large pizza it is"
5. CHECKPOINT: Check queue → Found: "with pepperoni"
6. Action: Create large pepperoni pizza order

# The golden rule:
if about_to_do_something_important:
    new_messages = await self.queue_helpers.get_new_messages(conversation_id)
    if new_messages:
        # Update your understanding before proceeding
        context = merge_new_information(context, new_messages)