Pending Messages

import { chat } from "@trigger.dev/sdk/ai";
import { stepCountIs, tool } from "ai";
import { anthropic } from "@ai-sdk/anthropic";
import { z } from "zod";
import { setTimeout } from "node:timers/promises";

export const myChat = chat.agent({
  id: "my-chat",
  pendingMessages: {
    // Only inject when there are completed steps (tool calls happened)
    shouldInject: ({ steps }) => steps.length > 0,
  },
  run: async ({ messages, signal, streamText }) => {
    return streamText({
      model: anthropic("claude-sonnet-4-5"),
      messages,
      tools: {
        inspectDocument: tool({
          description: "Inspect a document before summarizing it.",
          inputSchema: z.object({ topic: z.string() }),
          execute: async ({ topic }, { abortSignal }) => {
            // Leave time to send a steering message in this example.
            await setTimeout(10_000, undefined, { signal: abortSignal });
            return { topic, findings: "The document describes a chat application." };
          },
        }),
      },
      abortSignal: signal,
      stopWhen: stepCountIs(15),
    });
  },
});

pendingMessages: {
  // Always inject
  shouldInject: () => true,

  // Only inject after tool calls
  shouldInject: ({ steps }) => steps.length > 0,

  // Only inject if there's one message
  shouldInject: ({ messages }) => messages.length === 1,
},

pendingMessages: {
  shouldInject: ({ steps }) => steps.length > 0,
  prepare: ({ messages }) => [{
    role: "user",
    content: messages.length === 1
      ? messages[0].parts[0]?.text ?? ""
      : `The user sent ${messages.length} messages:\n${
          messages.map((m, i) => `${i + 1}. ${m.parts[0]?.text}`).join("\n")
        }`,
  }],
},

const session = chat.createSession(payload, {
  signal,
  idleTimeoutInSeconds: 60,
  pendingMessages: {
    shouldInject: () => true,
  },
});

for await (const turn of session) {
  const result = streamText({
    model: anthropic("claude-sonnet-4-5"),
    messages: turn.messages,
    abortSignal: turn.signal,
    prepareStep: turn.prepareStep(), // Handles injection + compaction
    stopWhen: stepCountIs(15),
  });

  await turn.complete(result);
}

const conversation = new chat.MessageAccumulator({
  pendingMessages: {
    shouldInject: () => true,
    prepare: ({ messages }) => [{
      role: "user",
      content: `[Steering]: ${messages.map(m => m.parts[0]?.text).join(", ")}`,
    }],
  },
});

for (let turn = 0; turn < 100; turn++) {
  // The wire payload carries at most one new message per turn.
  const messages = await conversation.addIncoming(
    payload.message ? [payload.message] : [],
    payload.trigger,
    turn
  );

  // Listen for steering messages during streaming
  const sub = chat.messages.on(async (msg) => {
    if (msg.message) await conversation.steerAsync(msg.message);
  });

  const result = streamText({
    model: anthropic("claude-sonnet-4-5"),
    messages,
    prepareStep: conversation.prepareStep(), // Handles injection + compaction
    stopWhen: stepCountIs(15),
  });

  const { message, status, error } = await chat.pipeAndCapture(result);
  sub.off();

  // Keep any partial output, then branch on the outcome: pipeAndCapture no
  // longer throws, so rethrow a real failure and don't complete a failed turn.
  if (message) await conversation.addResponse(message);
  if (status === "error") throw error;

  await chat.writeTurnComplete();
}

import { useChat } from "@ai-sdk/react";
import { useTriggerChatTransport, usePendingMessages } from "@trigger.dev/sdk/chat/react";

function Chat({ chatId }: { chatId: string }) {
  const transport = useTriggerChatTransport({
    task: "my-chat",
    accessToken: ({ chatId }) => mintChatAccessToken(chatId),
    startSession: ({ chatId, clientData }) =>
      startChatSession({ chatId, clientData }),
  });

  const { messages, setMessages, sendMessage, stop, status } = useChat({
    id: chatId,
    transport,
  });

  const pending = usePendingMessages({
    transport,
    chatId,
    status,
    messages,
    setMessages,
    sendMessage,
    metadata: { model: "gpt-4o" },
  });

  return (
    <div>
      {/* Render messages */}
      {messages.map((msg) => (
        <div key={msg.id}>
          {msg.role === "assistant" ? (
            msg.parts.map((part, i) =>
              pending.isInjectionPoint(part) ? (
                // Render injected messages inline at the injection point
                <div key={i}>
                  {pending.getInjectedMessages(part).map((m) => (
                    <div key={m.id} className="injected-message">{m.text}</div>
                  ))}
                </div>
              ) : (
                <Part key={i} part={part} />
              )
            )
          ) : (
            <UserMessage msg={msg} />
          )}
        </div>
      ))}

      {/* Render pending messages */}
      {pending.pending.map((msg) => (
        <div key={msg.id}>
          <span>{msg.text}</span>
          <span>{msg.mode === "steering" ? "Steering" : "Queued"}</span>
          {msg.mode === "queued" && status === "streaming" && (
            <button onClick={() => pending.promoteToSteering(msg.id)}>
              Steer instead
            </button>
          )}
        </div>
      ))}

      {/* Send form */}
      <form onSubmit={(e) => {
        e.preventDefault();
        pending.steer(input); // Steers during streaming, sends normally when ready
        setInput("");
      }}>
        <input value={input} onChange={(e) => setInput(e.target.value)} />
        <button type="submit">Send</button>
        {status === "streaming" && (
          <button type="button" onClick={() => { pending.queue(input); setInput(""); }}>
            Queue
          </button>
        )}
      </form>
    </div>
  );
}

const sent = await transport.sendPendingMessage(chatId, {
  id: crypto.randomUUID(),
  role: "user",
  parts: [{ type: "text", text: "and compare to vercel" }],
}, { model: "gpt-4o" });