feat(chat): add preload support, dynamic tools, and preload-specific timeouts

- Add onPreload hook and preloaded field to all lifecycle events
- Add transport.preload(chatId) for eagerly starting runs before first message
- Add preloadWarmTimeoutInSeconds and preloadTimeout task options
- Add preload:true run tag and chat.preloaded span attributes
- Add UserTool model for per-user dynamic tools loaded from DB
- Load dynamic tools in onPreload/onChatStart via chat.local
- Build dynamicTool() instances in run and spread into streamText tools
- Reference project: preload on new chat, dynamic company-info and user-preferences tools

Author: ericallam

packages/trigger-sdk/src/v3/ai.ts

@@ -300,7 +300,7 @@ const chatStream = streams.define<UIMessageChunk>({ id: _CHAT_STREAM_KEY });
 type ChatTaskWirePayload<TMessage extends UIMessage = UIMessage, TMetadata = unknown> = {
   messages: TMessage[];
   chatId: string;
-  trigger: "submit-message" | "regenerate-message";
+  trigger: "submit-message" | "regenerate-message" | "preload";
   messageId?: string;
   metadata?: TMetadata;
   /** Whether this run is continuing an existing chat whose previous run ended. */
@@ -330,8 +330,9 @@ export type ChatTaskPayload<TClientData = unknown> = {
    * The trigger type:
    * - `"submit-message"`: A new user message
    * - `"regenerate-message"`: Regenerate the last assistant response
+   * - `"preload"`: Run was preloaded before the first message (only on turn 0)
    */
-  trigger: "submit-message" | "regenerate-message";
+  trigger: "submit-message" | "regenerate-message" | "preload";
 
   /** The ID of the message to regenerate (only for `"regenerate-message"`) */
   messageId?: string;
@@ -343,6 +344,8 @@ export type ChatTaskPayload<TClientData = unknown> = {
   continuation: boolean;
   /** The run ID of the previous run (only set when `continuation` is true). */
   previousRunId?: string;
+  /** Whether this run was preloaded before the first message. */
+  preloaded: boolean;
 };
 
 /**
@@ -510,6 +513,20 @@ async function pipeChat(
  * emits a control chunk and suspends via `messagesInput.wait()`. The frontend
  * transport resumes the same run by sending the next message via input streams.
  */
+/**
+ * Event passed to the `onPreload` callback.
+ */
+export type PreloadEvent<TClientData = unknown> = {
+  /** The unique identifier for the chat session. */
+  chatId: string;
+  /** The Trigger.dev run ID for this conversation. */
+  runId: string;
+  /** A scoped access token for this chat run. */
+  chatAccessToken: string;
+  /** Custom data from the frontend. */
+  clientData?: TClientData;
+};
+
 /**
  * Event passed to the `onChatStart` callback.
  */
@@ -528,6 +545,8 @@ export type ChatStartEvent<TClientData = unknown> = {
   continuation: boolean;
   /** The run ID of the previous run (only set when `continuation` is true). */
   previousRunId?: string;
+  /** Whether this run was preloaded before the first message. */
+  preloaded: boolean;
 };
 
 /**
@@ -552,6 +571,8 @@ export type TurnStartEvent<TClientData = unknown> = {
   continuation: boolean;
   /** The run ID of the previous run (only set when `continuation` is true). */
   previousRunId?: string;
+  /** Whether this run was preloaded before the first message. */
+  preloaded: boolean;
 };
 
 /**
@@ -601,6 +622,8 @@ export type TurnCompleteEvent<TClientData = unknown> = {
   continuation: boolean;
   /** The run ID of the previous run (only set when `continuation` is true). */
   previousRunId?: string;
+  /** Whether this run was preloaded before the first message. */
+  preloaded: boolean;
 };
 
 export type ChatTaskOptions<
@@ -638,6 +661,22 @@ export type ChatTaskOptions<
    */
   run: (payload: ChatTaskRunPayload<inferSchemaOut<TClientDataSchema>>) => Promise<unknown>;
 
+  /**
+   * Called when a preloaded run starts, before the first message arrives.
+   *
+   * Use this to initialize state, create DB records, and load context early —
+   * so everything is ready when the user's first message comes through.
+   *
+   * @example
+   * ```ts
+   * onPreload: async ({ chatId, clientData }) => {
+   *   await db.chat.create({ data: { id: chatId } });
+   *   userContext.init(await loadUser(clientData.userId));
+   * }
+   * ```
+   */
+  onPreload?: (event: PreloadEvent<inferSchemaOut<TClientDataSchema>>) => Promise<void> | void;
+
   /**
    * Called on the first turn (turn 0) of a new run, before the `run` function executes.
    *
@@ -722,6 +761,26 @@ export type ChatTaskOptions<
    * @default "1h"
    */
   chatAccessTokenTTL?: string;
+
+  /**
+   * How long (in seconds) to keep the run warm after `onPreload` fires,
+   * waiting for the first message before suspending.
+   *
+   * Only applies to preloaded runs (triggered via `transport.preload()`).
+   *
+   * @default Same as `warmTimeoutInSeconds`
+   */
+  preloadWarmTimeoutInSeconds?: number;
+
+  /**
+   * How long to wait (suspended) for the first message after a preloaded run starts.
+   * If no message arrives within this time, the run ends.
+   *
+   * Only applies to preloaded runs.
+   *
+   * @default Same as `turnTimeout`
+   */
+  preloadTimeout?: string;
 };
 
 /**
@@ -760,13 +819,16 @@ function chatTask<
   const {
     run: userRun,
     clientDataSchema,
+    onPreload,
     onChatStart,
     onTurnStart,
     onTurnComplete,
     maxTurns = 100,
     turnTimeout = "1h",
     warmTimeoutInSeconds = 30,
     chatAccessTokenTTL = "1h",
+    preloadWarmTimeoutInSeconds,
+    preloadTimeout,
     ...restOptions
   } = options;
 
@@ -786,6 +848,7 @@ function chatTask<
       let currentWirePayload = payload;
       const continuation = payload.continuation ?? false;
       const previousRunId = payload.previousRunId;
+      const preloaded = payload.trigger === "preload";
 
       // Accumulated model messages across turns. Turn 1 initialises from the
       // full history the frontend sends; subsequent turns append only the new
@@ -806,6 +869,96 @@ function chatTask<
       });
 
       try {
+        // Handle preloaded runs — fire onPreload, then wait for the first real message
+        if (preloaded) {
+          if (activeSpan) {
+            activeSpan.setAttribute("chat.preloaded", true);
+          }
+
+          const currentRunId = taskContext.ctx?.run.id ?? "";
+          let preloadAccessToken = "";
+          if (currentRunId) {
+            try {
+              preloadAccessToken = await auth.createPublicToken({
+                scopes: {
+                  read: { runs: currentRunId },
+                  write: { inputStreams: currentRunId },
+                },
+                expirationTime: chatAccessTokenTTL,
+              });
+            } catch {
+              // Token creation failed
+            }
+          }
+
+          // Parse client data for the preload hook
+          const preloadClientData = (parseClientData
+            ? await parseClientData(payload.metadata)
+            : payload.metadata) as inferSchemaOut<TClientDataSchema>;
+
+          // Fire onPreload hook
+          if (onPreload) {
+            await tracer.startActiveSpan(
+              "onPreload()",
+              async () => {
+                await onPreload({
+                  chatId: payload.chatId,
+                  runId: currentRunId,
+                  chatAccessToken: preloadAccessToken,
+                  clientData: preloadClientData,
+                });
+              },
+              {
+                attributes: {
+                  [SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart",
+                  [SemanticInternalAttributes.COLLAPSED]: true,
+                  "chat.id": payload.chatId,
+                  "chat.preloaded": true,
+                },
+              }
+            );
+          }
+
+          // Wait for the first real message — use preload-specific timeouts if configured
+          const effectivePreloadWarmTimeout =
+            (metadata.get(WARM_TIMEOUT_METADATA_KEY) as number | undefined)
+            ?? preloadWarmTimeoutInSeconds
+            ?? warmTimeoutInSeconds;
+
+          let firstMessage: ChatTaskWirePayload | undefined;
+
+          if (effectivePreloadWarmTimeout > 0) {
+            const warm = await messagesInput.once({
+              timeoutMs: effectivePreloadWarmTimeout * 1000,
+              spanName: "preload wait (warm)",
+            });
+
+            if (warm.ok) {
+              firstMessage = warm.output;
+            }
+          }
+
+          if (!firstMessage) {
+            const effectivePreloadTimeout =
+              (metadata.get(TURN_TIMEOUT_METADATA_KEY) as string | undefined)
+              ?? preloadTimeout
+              ?? turnTimeout;
+
+            const suspended = await messagesInput.wait({
+              timeout: effectivePreloadTimeout,
+              spanName: "preload wait (suspended)",
+            });
+
+            if (!suspended.ok) {
+              return; // Timed out waiting for first message — end run
+            }
+
+            firstMessage = suspended.output;
+          }
+
+          currentWirePayload = firstMessage;
+        }
+
         for (let turn = 0; turn < maxTurns; turn++) {
           // Extract turn-level context before entering the span
           const { metadata: wireMetadata, messages: uiMessages, ...restWire } = currentWirePayload;
@@ -947,6 +1100,7 @@ function chatTask<
                       chatAccessToken: turnAccessToken,
                       continuation,
                       previousRunId,
+                      preloaded,
                     });
                   },
                   {
@@ -956,6 +1110,7 @@ function chatTask<
                       "chat.id": currentWirePayload.chatId,
                       "chat.messages.count": accumulatedMessages.length,
                       "chat.continuation": continuation,
+                      "chat.preloaded": preloaded,
                       ...(previousRunId ? { "chat.previous_run_id": previousRunId } : {}),
                     },
                   }
@@ -978,6 +1133,7 @@ function chatTask<
                       clientData,
                       continuation,
                       previousRunId,
+                      preloaded,
                     });
                   },
                   {
@@ -989,6 +1145,7 @@ function chatTask<
                       "chat.messages.count": accumulatedMessages.length,
                       "chat.trigger": currentWirePayload.trigger,
                       "chat.continuation": continuation,
+                      "chat.preloaded": preloaded,
                       ...(previousRunId ? { "chat.previous_run_id": previousRunId } : {}),
                     },
                   }
@@ -1014,6 +1171,7 @@ function chatTask<
                   clientData,
                   continuation,
                   previousRunId,
+                  preloaded,
                   signal: combinedSignal,
                   cancelSignal,
                   stopSignal,
@@ -1119,6 +1277,7 @@ function chatTask<
                       stopped: wasStopped,
                       continuation,
                       previousRunId,
+                      preloaded,
                     });
                   },
                   {
@@ -1129,6 +1288,7 @@ function chatTask<
                       "chat.turn": turn + 1,
                       "chat.stopped": wasStopped,
                       "chat.continuation": continuation,
+                      "chat.preloaded": preloaded,
                       ...(previousRunId ? { "chat.previous_run_id": previousRunId } : {}),
                       "chat.messages.count": accumulatedMessages.length,
                       "chat.response.parts.count": capturedResponseMessage?.parts?.length ?? 0,

packages/trigger-sdk/src/v3/chat.ts

@@ -183,6 +183,7 @@ export type TriggerChatTransportOptions<TClientData = unknown> = {
     /** Priority (lower = higher priority). */
     priority?: number;
   };
+
 };
 
 /**
@@ -451,6 +452,62 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     this._onSessionChange = callback;
   }
 
+  /**
+   * Eagerly trigger a run for a chat before the first message is sent.
+   * This allows initialization (DB setup, context loading) to happen
+   * while the user is still typing, reducing first-response latency.
+   *
+   * The task's `onPreload` hook fires immediately. The run then waits
+   * for the first message via input stream. When `sendMessages` is called
+   * later, it detects the existing session and sends via input stream
+   * instead of triggering a new run.
+   *
+   * No-op if a session already exists for this chatId.
+   */
+  async preload(chatId: string): Promise<void> {
+    // Don't preload if session already exists
+    if (this.sessions.get(chatId)?.runId) return;
+
+    const payload = {
+      messages: [] as never[],
+      chatId,
+      trigger: "preload" as const,
+      metadata: this.defaultMetadata,
+    };
+
+    const currentToken = await this.resolveAccessToken();
+    const apiClient = new ApiClient(this.baseURL, currentToken);
+
+    const autoTags = [`chat:${chatId}`, "preload:true"];
+    const userTags = this.triggerOptions?.tags ?? [];
+    const tags = [...autoTags, ...userTags].slice(0, 5);
+
+    const triggerResponse = await apiClient.triggerTask(this.taskId, {
+      payload,
+      options: {
+        payloadType: "application/json",
+        tags,
+        queue: this.triggerOptions?.queue ? { name: this.triggerOptions.queue } : undefined,
+        maxAttempts: this.triggerOptions?.maxAttempts,
+        machine: this.triggerOptions?.machine,
+        priority: this.triggerOptions?.priority,
+      },
+    });
+
+    const runId = triggerResponse.id;
+    const publicAccessToken =
+      "publicAccessToken" in triggerResponse
+        ? (triggerResponse as { publicAccessToken?: string }).publicAccessToken
+        : undefined;
+
+    const newSession: ChatSessionState = {
+      runId,
+      publicAccessToken: publicAccessToken ?? currentToken,
+    };
+    this.sessions.set(chatId, newSession);
+    this.notifySessionChange(chatId, newSession);
+  }
+
   private notifySessionChange(
     chatId: string,
     session: ChatSessionState | null