AI Agents

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.

WorkflowAgent'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. prepareStep also allows you to modify the model choice and existing messages mid-turn, see AI SDK's prepareStep callback for more details.

When to use this

Message queueing is useful when:

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

If you need basic multi-turn conversations where messages arrive between turns, see Chat Session Modeling. This guide covers the more advanced case of injecting messages during turns.

The prepareStep callback

The prepareStep callback runs before each step in the agent loop. Use WorkflowAgent's exported types rather than redeclaring its normalized provider-prompt contract:

import type {
  PrepareStepInfo,
  PrepareStepResult,
} from "@ai-sdk/workflow";

const prepareStep = (
  { messages }: PrepareStepInfo
): PrepareStepResult => ({ messages });

PrepareStepInfo.messages is a normalized LanguageModelV4Prompt, not the application-level ModelMessage[] accepted by WorkflowAgent.stream().

Queueing messages during and between turns

Use one async Hook consumer and one FIFO. prepareStep atomically drains messages that arrived during a model turn; messages that arrive after the final model step become input to the next turn. Each Hook payload therefore has exactly one ownership path.

workflows/chat/index.ts
import { WorkflowAgent, type ModelCallStreamPart } from "@ai-sdk/workflow";
import { getWritable, getWorkflowMetadata } from "workflow";
import { chatMessageHook } from "./hooks/chat-message";
import { flightBookingTools, FLIGHT_ASSISTANT_PROMPT } from "./steps/tools";
import type { ModelMessage } from "ai";

export async function chat(initialMessages: ModelMessage[]) {
  "use workflow";

  const { workflowRunId: runId } = getWorkflowMetadata();
  const writable = getWritable<ModelCallStreamPart>();
  let messages: ModelMessage[] = [...initialMessages];
  const messageQueue: Array<{ role: "user"; content: string }> = [];
  let stopped = false;
  let notifyMessage: (() => void) | undefined;

  const agent = new WorkflowAgent({
    model: "spacexai/grok-4.6",
    instructions: FLIGHT_ASSISTANT_PROMPT,
    tools: flightBookingTools,
  });

  const hook = chatMessageHook.create({ token: runId });
  // This is the only code path that consumes Hook payloads.
  const consumeMessages = (async () => {
    for await (const { message } of hook) {
      if (message === "/done") {
        stopped = true;
        notifyMessage?.();
        break;
      }
      messageQueue.push({ role: "user", content: message });
      notifyMessage?.();
      notifyMessage = undefined;
    }
  })();

  const waitForMessage = async () => {
    while (messageQueue.length === 0 && !stopped) {
      await new Promise<void>((resolve) => {
        notifyMessage = resolve;
      });
    }
  };

  while (!stopped) {
    const result = await agent.stream({
      messages,
      writable,
      preventClose: true,
      sendFinish: false,
      prepareStep: ({ messages: currentMessages }) => {
        const queued = messageQueue.splice(0); // Atomic drain
        if (queued.length === 0) return {};
        return {
          messages: [
            ...currentMessages,
            ...queued.map(({ role, content }) => ({
              role,
              content: [{ type: "text" as const, text: content }],
            })),
          ],
        };
      },
    });
    messages = result.messages;

    if (stopped) break;
    await waitForMessage();
    if (stopped) break;

    // Anything not consumed by prepareStep arrived after the final model step.
    messages = [...messages, ...messageQueue.splice(0)];
  }

  await consumeMessages;
  return { messages };
}

Messages sent via chatMessageHook.resume() accumulate until either prepareStep or the between-turn branch drains the FIFO. Send /done to stop the consumer and let the workflow return.

PreviousPatterns for Defining ToolsNextChat Session Modeling