Queueing User Messages

When using multi-turn workflows, messages typically arrive between agent turns. The workflow waits at a hook, receives a message, then starts a new turn. But sometimes you need to inject messages during an agent's turn, before tool calls complete or while the model is reasoning.

DurableAgent's prepareStep callback enables this by running before each step in the agent loop, giving you a chance to inject queued messages into the conversation.

When to Use This

Message queueing is useful when:

• Users send follow-up messages while the agent is still processing tool calls
• External systems need to inject context mid-turn (e.g., a webhook fires during processing)
• You want messages to influence the agent's next step rather than waiting for the current turn to complete

The prepareStep Callback

The prepareStep callback runs before each step in the agent loop. It receives the current state and can modify the messages sent to the model:

interface PrepareStepInfo {
  model: string | (() => Promise<LanguageModelV2>);  // Current model
  stepNumber: number;                                 // 0-indexed step count
  steps: StepResult[];                               // Previous step results
  messages: LanguageModelV2Prompt;                   // Messages to be sent
}

interface PrepareStepResult {
  model?: string | (() => Promise<LanguageModelV2>); // Override model
  messages?: LanguageModelV2Prompt;                  // Override messages
}

Injecting Queued Messages

Combine a message queue with prepareStep to inject messages that arrive during processing:

import { DurableAgent } from '@workflow/ai/agent';
import type { UIMessageChunk } from 'ai';
import { getWritable } from 'workflow';
import { chatMessageHook } from '@/ai/hooks/chat-message';

export async function chatWorkflow(threadId: string, initialMessage: string) {
  'use workflow';

  const writable = getWritable<UIMessageChunk>();
  const messageQueue: Array<{ role: 'user'; content: string }> = []; 

  const agent = new DurableAgent({
    model: 'anthropic/claude-haiku-4.5',
    system: 'You are a helpful assistant.',
    tools: { /* long-running tools */ },
  });

  // Listen for messages in background (non-blocking)
  const hook = chatMessageHook.create({ token: `thread:${threadId}` }); 
  hook.then(({ message }) => { 
    messageQueue.push({ role: 'user', content: message }); 
  }); 

  await agent.stream({
    messages: [{ role: 'user', content: initialMessage }],
    writable,
    prepareStep: ({ messages: currentMessages }) => { 
      // Inject any queued messages before the next LLM call
      if (messageQueue.length > 0) { 
        const newMessages = messageQueue.splice(0); // Drain queue
        return { 
          messages: [ 
            ...currentMessages, 
            ...newMessages.map(m => ({ 
              role: m.role, 
              content: [{ type: 'text' as const, text: m.content }], 
            })), 
          ], 
        }; 
      } 
      return {}; 
    }, 
  });
}

Messages sent via chatMessageHook.resume() accumulate in the queue and get injected before the next step, whether that's a tool call or another LLM request.

Other prepareStep Patterns

Beyond message injection, prepareStep supports other per-step modifications:

Context Management: Trim old messages to stay within context limits:

prepareStep: ({ messages }) => {
  if (messages.length > 20) {
    return {
      messages: [
        messages[0],            // Keep system message
        ...messages.slice(-10), // Keep last 10 messages
      ],
    };
  }
  return {};
}

Dynamic Model Selection: Switch models based on conversation complexity:

prepareStep: ({ stepNumber, messages }) => {
  // Use a stronger model for later steps
  if (stepNumber > 2 && messages.length > 10) {
    return { model: 'anthropic/claude-sonnet-4.5' };
  }
  return {};
}

Related Documentation

• Chat Session Modeling - Single-turn vs multi-turn patterns
• Building Durable AI Agents - Complete guide to creating durable agents
• DurableAgent API Reference - Full API documentation
• defineHook() API Reference - Hook configuration options