Managing Message Collisions with the Command Queue

I have often run into the problem where a user sends four messages in quick succession, and the agent tries to process all of them at once. This usually ends in a disaster of file locks, messy logs, hitting API rate limits, and competing for the same session data. It is a common headache when building any agent that talks to humans on fast-moving channels.

To fix this, we use a lane-aware FIFO queue that serializes runs. This ensures that while we can still handle many users at once, a single session never trips over itself.

What You’ll Need

• An agent using the gateway reply pipeline (WhatsApp, Telegram, Slack, Discord, etc.).
• Access to your config.json5 file to set global or per-channel defaults.

Quick Start

If you want to get the queue running with the recommended settings, follow these steps:

1. Choose your mode: I recommend collect as it is the safest for most use cases.
2. Update your configuration: Add the messages.queue block to your config file.
3. Set concurrency: Adjust agents.defaults.maxConcurrent to control how many sessions run in parallel.
4. Verify the setup: Send multiple messages to your agent and check if it processes them in order rather than all at once.

{
  messages: {
    queue: {
      mode: "collect",
      debounceMs: 1000,
      cap: 20,
      drop: "summarize",
      byChannel: { discord: "collect" },
    },
  },
  agents: {
    defaults: {
      maxConcurrent: 4
    }
  }
}

How the Queue Works

The system uses “lanes” to manage traffic. When runEmbeddedPiAgent starts, it enqueues the run by its session key. This creates a specific lane for that user, guaranteeing only one active run per session.

These session runs then feed into a global lane (usually called main). The parallelism of this global lane is controlled by agents.defaults.maxConcurrent. This structure prevents a single user from breaking their session state while allowing you to help multiple users simultaneously.

Queue Modes

You can change how the queue handles new messages while a run is already active. Here are the four primary modes:

• collect (Default): This gathers all incoming messages during a run and processes them together in one single followup turn.
• steer: This injects the new message into the current run immediately, cancelling pending tool calls.
• followup: This simply queues the message for the very next turn after the current one finishes.
• steer-backlog: This steers the current run and still keeps the message for a followup turn.

I suggest sticking with collect or steer if you want to avoid duplicate-looking responses on streaming interfaces.

Per-Session Overrides

You don’t have to restart your agent to change how a specific session behaves. You can send commands directly in the chat:

• /queue collect debounce:2s cap:25 drop:summarize
• /queue default (restores config settings)

Configuration Options

When you use followup, collect, or steer-backlog, you have four main settings to tweak:

1. debounceMs: How long to wait for silence before starting a followup (defaults to 1000).
2. cap: The maximum number of messages allowed in the queue for one session (defaults to 20).
3. drop: What to do when the cap is hit (old, new, or summarize).
4. byChannel: Specific overrides for different platforms like Discord or Slack.

If you choose summarize for the drop policy, the agent creates a bulleted list of the ignored messages and feeds it to the next successful run as a synthetic prompt.

Troubleshooting

• Commands seem stuck: Enable verbose logs. Look for lines that say “queued for …ms” to see if the queue is actually draining.
• Checking queue depth: If you suspect a bottleneck, enable verbose logs and monitor the queue timing lines to see how long tasks are waiting.

If you have specific questions about your lane configuration, ask the AI Setup Assistant.

What’s Next

• Configuring Agent Concurrency
• Handling Gateway Channels