feat: add onTurnStart hook, lastEventId support, and stream resume deduplication

Author: ericallam

packages/core/src/v3/realtimeStreams/manager.ts

@@ -6,6 +6,7 @@ import {
   RealtimeStreamInstance,
   RealtimeStreamOperationOptions,
   RealtimeStreamsManager,
+  StreamWriteResult,
 } from "./types.js";
 
 export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
@@ -16,7 +17,7 @@ export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
   ) {}
   // Track active streams - using a Set allows multiple streams for the same key to coexist
   private activeStreams = new Set<{
-    wait: () => Promise<void>;
+    wait: () => Promise<StreamWriteResult>;
     abortController: AbortController;
   }>();
 

packages/core/src/v3/realtimeStreams/noopManager.ts

@@ -15,7 +15,7 @@ export class NoopRealtimeStreamsManager implements RealtimeStreamsManager {
     options?: RealtimeStreamOperationOptions
   ): RealtimeStreamInstance<T> {
     return {
-      wait: () => Promise.resolve(),
+      wait: () => Promise.resolve({}),
       get stream(): AsyncIterableStream<T> {
         return createAsyncIterableStreamFromAsyncIterable(source);
       },

packages/core/src/v3/realtimeStreams/streamInstance.ts

@@ -3,7 +3,7 @@ import { AsyncIterableStream } from "../streams/asyncIterableStream.js";
 import { AnyZodFetchOptions } from "../zodfetch.js";
 import { StreamsWriterV1 } from "./streamsWriterV1.js";
 import { StreamsWriterV2 } from "./streamsWriterV2.js";
-import { StreamsWriter } from "./types.js";
+import { StreamsWriter, StreamWriteResult } from "./types.js";
 
 export type StreamInstanceOptions<T> = {
   apiClient: ApiClient;
@@ -63,8 +63,9 @@ export class StreamInstance<T> implements StreamsWriter {
     return streamWriter;
   }
 
-  public async wait(): Promise<void> {
-    return this.streamPromise.then((writer) => writer.wait());
+  public async wait(): Promise<StreamWriteResult> {
+    const writer = await this.streamPromise;
+    return writer.wait();
   }
 
   public get stream(): AsyncIterableStream<T> {

packages/core/src/v3/realtimeStreams/streamsWriterV1.ts

@@ -2,7 +2,7 @@ import { request as httpsRequest } from "node:https";
 import { request as httpRequest } from "node:http";
 import { URL } from "node:url";
 import { randomBytes } from "node:crypto";
-import { StreamsWriter } from "./types.js";
+import { StreamsWriter, StreamWriteResult } from "./types.js";
 
 export type StreamsWriterV1Options<T> = {
   baseUrl: string;
@@ -258,8 +258,9 @@ export class StreamsWriterV1<T> implements StreamsWriter {
     await this.makeRequest(0);
   }
 
-  public async wait(): Promise<void> {
-    return this.streamPromise;
+  public async wait(): Promise<StreamWriteResult> {
+    await this.streamPromise;
+    return {};
   }
 
   public [Symbol.asyncIterator]() {

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -1,5 +1,5 @@
 import { S2, AppendRecord, BatchTransform } from "@s2-dev/streamstore";
-import { StreamsWriter } from "./types.js";
+import { StreamsWriter, StreamWriteResult } from "./types.js";
 import { nanoid } from "nanoid";
 
 export type StreamsWriterV2Options<T = any> = {
@@ -54,6 +54,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
   private readonly maxInflightBytes: number;
   private aborted = false;
   private sessionWritable: WritableStream<any> | null = null;
+  private lastSeqNum: number | undefined;
 
   constructor(private options: StreamsWriterV2Options<T>) {
     this.debug = options.debug ?? false;
@@ -169,9 +170,9 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       const lastAcked = session.lastAckedPosition();
 
       if (lastAcked?.end) {
-        const recordsWritten = lastAcked.end.seqNum;
+        this.lastSeqNum = lastAcked.end.seqNum;
         this.log(
-          `[S2MetadataStream] Written ${recordsWritten} records, ending at seqNum=${lastAcked.end.seqNum}`
+          `[S2MetadataStream] Written ${this.lastSeqNum} records, ending at seqNum=${this.lastSeqNum}`
         );
       }
     } catch (error) {
@@ -184,8 +185,9 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
     }
   }
 
-  public async wait(): Promise<void> {
+  public async wait(): Promise<StreamWriteResult> {
     await this.streamPromise;
+    return { lastEventId: this.lastSeqNum?.toString() };
   }
 
   public [Symbol.asyncIterator]() {

packages/core/src/v3/realtimeStreams/types.ts

@@ -26,13 +26,17 @@ export interface RealtimeStreamsManager {
   ): Promise<void>;
 }
 
+export type StreamWriteResult = {
+  lastEventId?: string;
+};
+
 export interface RealtimeStreamInstance<T> {
-  wait(): Promise<void>;
+  wait(): Promise<StreamWriteResult>;
   get stream(): AsyncIterableStream<T>;
 }
 
 export interface StreamsWriter {
-  wait(): Promise<void>;
+  wait(): Promise<StreamWriteResult>;
 }
 
 export type RealtimeDefinedStream<TPart> = {
@@ -93,7 +97,7 @@ export type PipeStreamResult<T> = {
    * to the realtime stream. Use this to wait for the stream to complete before
    * finishing your task.
    */
-  waitUntilComplete: () => Promise<void>;
+  waitUntilComplete: () => Promise<StreamWriteResult>;
 };
 
 /**

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

@@ -4,6 +4,7 @@ import {
   isSchemaZodEsque,
   SemanticInternalAttributes,
   Task,
+  taskContext,
   type inferSchemaIn,
   type PipeStreamOptions,
   type TaskIdentifier,
@@ -12,7 +13,8 @@ import {
   type TaskWithSchema,
 } from "@trigger.dev/core/v3";
 import type { ModelMessage, UIMessage } from "ai";
-import { convertToModelMessages, dynamicTool, jsonSchema, JSONSchema7, Schema, Tool, ToolCallOptions, zodSchema } from "ai";
+import type { StreamWriteResult } from "@trigger.dev/core/v3";
+import { convertToModelMessages, dynamicTool, generateId as generateMessageId, jsonSchema, JSONSchema7, Schema, Tool, ToolCallOptions, zodSchema } from "ai";
 import { type Attributes, trace } from "@opentelemetry/api";
 import { auth } from "./auth.js";
 import { metadata } from "./metadata.js";
@@ -153,7 +155,7 @@ export const ai = {
 function createChatAccessToken<TTask extends AnyTask>(
   taskId: TaskIdentifier<TTask>
 ): Promise<string> {
-  return auth.createTriggerPublicToken(taskId as string, { multipleUse: true });
+  return auth.createTriggerPublicToken(taskId as string, { expirationTime: "24h" });
 }
 
 // ---------------------------------------------------------------------------
@@ -389,6 +391,28 @@ export type ChatStartEvent = {
   messages: ModelMessage[];
   /** Custom data from the frontend (passed via `metadata` on `sendMessage()` or the transport). */
   clientData: unknown;
+  /** The Trigger.dev run ID for this conversation. */
+  runId: string;
+  /** A scoped access token for this chat run. Persist this for frontend reconnection. */
+  chatAccessToken: string;
+};
+
+/**
+ * Event passed to the `onTurnStart` callback.
+ */
+export type TurnStartEvent = {
+  /** The unique identifier for the chat session. */
+  chatId: string;
+  /** The accumulated model-ready messages (all turns so far, including new user message). */
+  messages: ModelMessage[];
+  /** The accumulated UI messages (all turns so far, including new user message). */
+  uiMessages: UIMessage[];
+  /** The turn number (0-indexed). */
+  turn: number;
+  /** The Trigger.dev run ID for this conversation. */
+  runId: string;
+  /** A scoped access token for this chat run. */
+  chatAccessToken: string;
 };
 
 /**
@@ -418,6 +442,12 @@ export type TurnCompleteEvent = {
   responseMessage: UIMessage | undefined;
   /** The turn number (0-indexed). */
   turn: number;
+  /** The Trigger.dev run ID for this conversation. */
+  runId: string;
+  /** A fresh scoped access token for this chat run (renewed each turn). Persist this for frontend reconnection. */
+  chatAccessToken: string;
+  /** The last event ID from the stream writer. Use this with `resume: true` to avoid replaying events after refresh. */
+  lastEventId?: string;
 };
 
 export type ChatTaskOptions<TIdentifier extends string> = Omit<
@@ -449,6 +479,22 @@ export type ChatTaskOptions<TIdentifier extends string> = Omit<
    */
   onChatStart?: (event: ChatStartEvent) => Promise<void> | void;
 
+  /**
+   * Called at the start of every turn, after message accumulation and `onChatStart` (turn 0),
+   * but before the `run` function executes.
+   *
+   * Use this to persist messages before streaming begins, so a mid-stream page refresh
+   * still shows the user's message.
+   *
+   * @example
+   * ```ts
+   * onTurnStart: async ({ chatId, uiMessages }) => {
+   *   await db.chat.update({ where: { id: chatId }, data: { messages: uiMessages } });
+   * }
+   * ```
+   */
+  onTurnStart?: (event: TurnStartEvent) => Promise<void> | void;
+
   /**
    * Called after each turn completes (after the response is captured, before waiting
    * for the next message). Also fires on the final turn.
@@ -492,6 +538,17 @@ export type ChatTaskOptions<TIdentifier extends string> = Omit<
    * @default 30
    */
   warmTimeoutInSeconds?: number;
+
+  /**
+   * How long the `chatAccessToken` (scoped to this run) remains valid.
+   * A fresh token is minted after each turn, so this only needs to cover
+   * the gap between turns.
+   *
+   * Accepts a duration string (e.g. `"1h"`, `"30m"`, `"2h"`).
+   *
+   * @default "1h"
+   */
+  chatAccessTokenTTL?: string;
 };
 
 /**
@@ -527,10 +584,12 @@ function chatTask<TIdentifier extends string>(
   const {
     run: userRun,
     onChatStart,
+    onTurnStart,
     onTurnComplete,
     maxTurns = 100,
     turnTimeout = "1h",
     warmTimeoutInSeconds = 30,
+    chatAccessTokenTTL = "1h",
     ...restOptions
   } = options;
 
@@ -653,6 +712,24 @@ function chatTask<TIdentifier extends string>(
                 turnNewUIMessages.push(...uiMessages);
               }
 
+              // Mint a scoped public access token once per turn, reused for
+              // onChatStart, onTurnStart, onTurnComplete, and the turn-complete chunk.
+              const currentRunId = taskContext.ctx?.run.id ?? "";
+              let turnAccessToken = "";
+              if (currentRunId) {
+                try {
+                  turnAccessToken = await auth.createPublicToken({
+                    scopes: {
+                      read: { runs: currentRunId },
+                      write: { inputStreams: currentRunId },
+                    },
+                    expirationTime: chatAccessTokenTTL,
+                  });
+                } catch {
+                  // Token creation failed
+                }
+              }
+
               // Fire onChatStart on the first turn
               if (turn === 0 && onChatStart) {
                 await tracer.startActiveSpan(
@@ -662,6 +739,32 @@ function chatTask<TIdentifier extends string>(
                       chatId: currentWirePayload.chatId,
                       messages: accumulatedMessages,
                       clientData: wireMetadata,
+                      runId: currentRunId,
+                      chatAccessToken: turnAccessToken,
+                    });
+                  },
+                  {
+                    attributes: {
+                      [SemanticInternalAttributes.STYLE_ICON]: "task-hook-onStart",
+                      [SemanticInternalAttributes.COLLAPSED]: true,
+                    },
+                  }
+                );
+              }
+
+              // Fire onTurnStart before running user code — persist messages
+              // so a mid-stream page refresh still shows the user's message.
+              if (onTurnStart) {
+                await tracer.startActiveSpan(
+                  "onTurnStart()",
+                  async () => {
+                    await onTurnStart({
+                      chatId: currentWirePayload.chatId,
+                      messages: accumulatedMessages,
+                      uiMessages: accumulatedUIMessages,
+                      turn,
+                      runId: currentRunId,
+                      chatAccessToken: turnAccessToken,
                     });
                   },
                   {
@@ -715,6 +818,12 @@ function chatTask<TIdentifier extends string>(
               // The onFinish callback fires even on abort/stop, so partial responses
               // from stopped generation are captured correctly.
               if (capturedResponseMessage) {
+                // Ensure the response message has an ID (the stream's onFinish
+                // may produce a message with an empty ID since IDs are normally
+                // assigned by the frontend's useChat).
+                if (!capturedResponseMessage.id) {
+                  capturedResponseMessage = { ...capturedResponseMessage, id: generateMessageId() };
+                }
                 accumulatedUIMessages.push(capturedResponseMessage);
                 turnNewUIMessages.push(capturedResponseMessage);
                 try {
@@ -734,6 +843,13 @@ function chatTask<TIdentifier extends string>(
 
               if (runSignal.aborted) return "exit";
 
+              // Write turn-complete control chunk so frontend closes its stream.
+              // Capture the lastEventId from the stream writer for resume support.
+              const turnCompleteResult = await writeTurnCompleteChunk(
+                currentWirePayload.chatId,
+                turnAccessToken
+              );
+
               // Fire onTurnComplete after response capture
               if (onTurnComplete) {
                 await tracer.startActiveSpan(
@@ -747,6 +863,9 @@ function chatTask<TIdentifier extends string>(
                       newUIMessages: turnNewUIMessages,
                       responseMessage: capturedResponseMessage,
                       turn,
+                      runId: currentRunId,
+                      chatAccessToken: turnAccessToken,
+                      lastEventId: turnCompleteResult.lastEventId,
                     });
                   },
                   {
@@ -758,9 +877,6 @@ function chatTask<TIdentifier extends string>(
                 );
               }
 
-              // Write turn-complete control chunk so frontend closes its stream
-              await writeTurnCompleteChunk(currentWirePayload.chatId);
-
               // If messages arrived during streaming, use the first one immediately
               if (pendingMessages.length > 0) {
                 currentWirePayload = pendingMessages[0]!;
@@ -927,15 +1043,18 @@ export const chat = {
  * The frontend transport intercepts this to close the ReadableStream for the current turn.
  * @internal
  */
-async function writeTurnCompleteChunk(chatId?: string): Promise<void> {
+async function writeTurnCompleteChunk(chatId?: string, publicAccessToken?: string): Promise<StreamWriteResult> {
   const { waitUntilComplete } = streams.writer(CHAT_STREAM_KEY, {
     spanName: "turn complete",
     collapsed: true,
     execute: ({ write }) => {
-      write({ type: "__trigger_turn_complete" });
+      write({
+        type: "__trigger_turn_complete",
+        ...(publicAccessToken ? { publicAccessToken } : {}),
+      });
     },
   });
-  await waitUntilComplete();
+  return await waitUntilComplete();
 }
 
 /**