feat(sdk): enrich transport events with correlation and timing fields

Send events gain partId (the append idempotency key, also on the
server-side record) and bodyBytes. Response events gain client-side
attribution: messageId of the last turn-producing send plus sinceSendMs,
so time-to-first-token and full-turn latency need no consumer
bookkeeping. first-chunk carries chunkType and its record id,
turn-completed carries the agent's committed input cursor, and
stream-error carries the HTTP status when one exists.

Author: ericallam

docs/ai-chat/frontend.mdx

@@ -615,4 +615,4 @@ onEvent: (event) => {
 };
 ```
 
-Time to first token is the delta between a `message-sent` and the following `first-chunk` on the same chat. Exceptions thrown inside `onEvent` are swallowed, so a failing metrics pipeline can never break the chat.
+Time to first token is `first-chunk`'s `sinceSendMs` (the transport tracks the last turn-producing send per chat, so no bookkeeping is needed), and `turn-completed`'s `sinceSendMs` is the full turn latency. Exceptions thrown inside `onEvent` are swallowed, so a failing metrics pipeline can never break the chat.

docs/ai-chat/reference.mdx

@@ -641,15 +641,17 @@ The `onEvent` callback receives a `ChatTransportEvent` (exported from `@trigger.
 
 | Event | Extra fields | Fires when |
 | --- | --- | --- |
-| `message-sent` | `messageId?`, `source`, `durationMs` | A send was durably acknowledged — a 2xx from the session input stream append (or the `headStart` POST), after any internal token-refresh retries. This means the message is durably written to the stream the agent consumes from, not merely "request accepted". |
-| `message-send-failed` | `messageId?`, `source`, `error`, `status?`, `durationMs` | A send definitively failed after internal retries. Fires in addition to `useChat`'s `onError`. |
-| `stream-connected` | `resumed` | The SSE subscription to the session's output stream started delivering. `resumed: true` when reconnecting from a stored cursor (page reload) rather than following a fresh send. |
-| `first-chunk` | — | The first response chunk of a turn arrived. Pair with `message-sent` for time-to-first-token. |
-| `turn-completed` | `lastEventId?` | The agent's turn-complete control record arrived — the "finished answering" signal. |
-| `stream-error` | `error` | The output stream failed unrecoverably. |
+| `message-sent` | `messageId?`, `source`, `durationMs`, `partId?`, `bodyBytes?` | A send was durably acknowledged — a 2xx from the session input stream append (or the `headStart` POST), after any internal token-refresh retries. This means the message is durably written to the stream the agent consumes from, not merely "request accepted". `partId` is the append's idempotency key, also stored on the server-side record. |
+| `message-send-failed` | `messageId?`, `source`, `error`, `status?`, `durationMs`, `partId?`, `bodyBytes?` | A send definitively failed after internal retries. Fires in addition to `useChat`'s `onError`. |
+| `stream-connected` | `resumed`, `lastEventId?`, `messageId?` | The SSE subscription to the session's output stream started delivering. `resumed: true` when reconnecting from a stored cursor (page reload) rather than following a fresh send. `lastEventId` is the cursor it connected from. |
+| `first-chunk` | `chunkType?`, `lastEventId?`, `messageId?`, `sinceSendMs?` | The first response chunk of a turn arrived. `sinceSendMs` is the delta from the last turn-producing send — time to first token without any bookkeeping. |
+| `turn-completed` | `lastEventId?`, `sessionInEventId?`, `messageId?`, `sinceSendMs?` | The agent's turn-complete control record arrived — the "finished answering" signal. `sinceSendMs` is the full turn latency; `sessionInEventId` is the agent's committed input-stream cursor. |
+| `stream-error` | `error`, `status?` | The output stream failed unrecoverably. |
 
 `source` identifies the send path: `"submit-message"`, `"regenerate-message"`, `"steer"` (`sendPendingMessage`), `"action"` (`sendAction`), `"stop"` (`stopGeneration`), or `"head-start"`.
 
+`messageId` on the response-side events is client-side attribution: the id from the most recent turn-producing send (`submit-message`, `regenerate-message`, or `head-start`) on that chat.
+
 See [Monitoring message delivery](/ai-chat/frontend#monitoring-message-delivery) for the metrics and watchdog patterns these enable.
 
 ### `accessToken` callback

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

@@ -28,9 +28,14 @@ import {
   controlSubtype,
   headerValue,
   PUBLIC_ACCESS_TOKEN_HEADER,
+  SESSION_IN_EVENT_ID_HEADER,
   SSEStreamSubscription,
   TRIGGER_CONTROL_SUBTYPE,
 } from "@trigger.dev/core/v3";
+
+function byteLength(body: string): number {
+  return new TextEncoder().encode(body).byteLength;
+}
 import { ChatTabCoordinator } from "./chat-tab-coordinator.js";
 import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js";
 import { slimSubmitMessageForWire } from "./ai-shared.js";
@@ -108,6 +113,10 @@ export type ChatTransportEvent =
       messageId?: string;
       source: ChatTransportSendSource;
       durationMs: number;
+      /** The append's idempotency key — also stored on the server-side record. */
+      partId?: string;
+      /** Serialized request body size in bytes. */
+      bodyBytes?: number;
     }
   | {
       type: "message-send-failed";
@@ -118,11 +127,41 @@ export type ChatTransportEvent =
       error: Error;
       status?: number;
       durationMs: number;
+      partId?: string;
+      bodyBytes?: number;
+    }
+  | {
+      type: "stream-connected";
+      chatId: string;
+      timestamp: number;
+      resumed: boolean;
+      /** The resume cursor the subscription connected from, if any. */
+      lastEventId?: string;
+      /** The last turn-producing send on this chat (client-side attribution). */
+      messageId?: string;
+    }
+  | { type: "stream-error"; chatId: string; timestamp: number; error: Error; status?: number }
+  | {
+      type: "first-chunk";
+      chatId: string;
+      timestamp: number;
+      chunkType?: string;
+      lastEventId?: string;
+      messageId?: string;
+      /** Milliseconds since the last turn-producing send on this chat (time to first token). */
+      sinceSendMs?: number;
     }
-  | { type: "stream-connected"; chatId: string; timestamp: number; resumed: boolean }
-  | { type: "stream-error"; chatId: string; timestamp: number; error: Error }
-  | { type: "first-chunk"; chatId: string; timestamp: number }
-  | { type: "turn-completed"; chatId: string; timestamp: number; lastEventId?: string };
+  | {
+      type: "turn-completed";
+      chatId: string;
+      timestamp: number;
+      lastEventId?: string;
+      /** The agent's committed input-stream cursor from the turn-complete record. */
+      sessionInEventId?: string;
+      messageId?: string;
+      /** Milliseconds since the last turn-producing send on this chat (full turn latency). */
+      sinceSendMs?: number;
+    };
 
 /**
  * Detect 401/403 from realtime/input-stream calls without relying on `instanceof`
@@ -520,6 +559,9 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
   private sessions: Map<string, ChatSessionState> = new Map();
   private activeStreams: Map<string, AbortController> = new Map();
   private pendingStarts: Map<string, Promise<ChatSessionState>> = new Map();
+  // Last turn-producing send per chat — attribution source for the
+  // response-side events' `messageId` / `sinceSendMs`.
+  private lastTurnSends: Map<string, { messageId?: string; at: number }> = new Map();
 
   constructor(options: TriggerChatTransportOptions) {
     this.taskId = options.task;
@@ -687,17 +729,20 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     // Generated outside the closure so auth-retries reuse the same part id
     // and the server-side dedupe sees one logical append.
     const partId = crypto.randomUUID();
+    const serializedBody = this.serializeInputChunk({ kind: "message", payload: wirePayload });
     const sendChatMessage = async (token: string) => {
-      await this.appendInputChunk(
-        chatId,
-        token,
-        this.serializeInputChunk({ kind: "message", payload: wirePayload }),
-        partId
-      );
+      await this.appendInputChunk(chatId, token, serializedBody, partId);
     };
 
-    await this.sendWithEvents(chatId, trigger, messageId ?? messages.at(-1)?.id, () =>
-      this.callWithAuthRetry(chatId, state, sendChatMessage)
+    await this.sendWithEvents(
+      chatId,
+      trigger,
+      {
+        messageId: messageId ?? messages.at(-1)?.id,
+        partId,
+        bodyBytes: byteLength(serializedBody),
+      },
+      () => this.callWithAuthRetry(chatId, state, sendChatMessage)
     );
 
     // Cancel any in-flight stream for this chat — the new turn supersedes it.
@@ -748,18 +793,22 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     };
 
     let response!: Response;
+    const serializedBody = JSON.stringify(wirePayload);
     await this.sendWithEvents(
       args.chatId,
       "head-start",
-      args.messageId ?? args.messages.at(-1)?.id,
+      {
+        messageId: args.messageId ?? args.messages.at(-1)?.id,
+        bodyBytes: byteLength(serializedBody),
+      },
       async () => {
         response = await fetch(this.headStart!, {
           method: "POST",
           headers: {
             "Content-Type": "application/json",
             ...this.extraHeaders,
           },
-          body: JSON.stringify(wirePayload),
+          body: serializedBody,
           signal: args.abortSignal,
         });
 
@@ -886,18 +935,17 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     };
 
     const partId = crypto.randomUUID();
+    const serializedBody = this.serializeInputChunk({ kind: "message", payload: wirePayload });
     const send = async (token: string) => {
-      await this.appendInputChunk(
-        chatId,
-        token,
-        this.serializeInputChunk({ kind: "message", payload: wirePayload }),
-        partId
-      );
+      await this.appendInputChunk(chatId, token, serializedBody, partId);
     };
 
     try {
-      await this.sendWithEvents(chatId, "steer", message.id, () =>
-        this.callWithAuthRetry(chatId, state, send)
+      await this.sendWithEvents(
+        chatId,
+        "steer",
+        { messageId: message.id, partId, bodyBytes: byteLength(serializedBody) },
+        () => this.callWithAuthRetry(chatId, state, send)
       );
       return true;
     } catch {
@@ -950,18 +998,17 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     if (!state) return false;
 
     const partId = crypto.randomUUID();
+    const serializedBody = this.serializeInputChunk({ kind: "stop" });
     const send = async (token: string) => {
-      await this.appendInputChunk(
-        chatId,
-        token,
-        this.serializeInputChunk({ kind: "stop" }),
-        partId
-      );
+      await this.appendInputChunk(chatId, token, serializedBody, partId);
     };
 
     try {
-      await this.sendWithEvents(chatId, "stop", undefined, () =>
-        this.callWithAuthRetry(chatId, state, send)
+      await this.sendWithEvents(
+        chatId,
+        "stop",
+        { partId, bodyBytes: byteLength(serializedBody) },
+        () => this.callWithAuthRetry(chatId, state, send)
       );
     } catch {
       return false;
@@ -1015,7 +1062,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       await this.appendInputChunk(chatId, token, body, partId);
     };
 
-    await this.sendWithEvents(chatId, "action", undefined, () =>
+    await this.sendWithEvents(chatId, "action", { partId, bodyBytes: byteLength(body) }, () =>
       this.callWithAuthRetry(chatId, state, send)
     );
 
@@ -1081,36 +1128,48 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
   private async sendWithEvents(
     chatId: string,
     source: ChatTransportSendSource,
-    messageId: string | undefined,
+    extras: { messageId?: string; partId?: string; bodyBytes?: number },
     op: () => Promise<void>
   ): Promise<void> {
     const startedAt = Date.now();
     try {
       await op();
+      const turnProducing =
+        source === "submit-message" || source === "regenerate-message" || source === "head-start";
+      if (turnProducing) {
+        this.lastTurnSends.set(chatId, { messageId: extras.messageId, at: Date.now() });
+      }
       this.emitEvent({
         type: "message-sent",
         chatId,
         timestamp: Date.now(),
-        messageId,
         source,
         durationMs: Date.now() - startedAt,
+        ...extras,
       });
     } catch (error) {
       const status = (error as { status?: unknown }).status;
       this.emitEvent({
         type: "message-send-failed",
         chatId,
         timestamp: Date.now(),
-        messageId,
         source,
         error: error instanceof Error ? error : new Error(String(error)),
         status: typeof status === "number" ? status : undefined,
         durationMs: Date.now() - startedAt,
+        ...extras,
       });
       throw error;
     }
   }
 
+  /** Attribution fields for response-side events. */
+  private turnAttribution(chatId: string): { messageId?: string; sinceSendMs?: number } {
+    const lastSend = this.lastTurnSends.get(chatId);
+    if (!lastSend) return {};
+    return { messageId: lastSend.messageId, sinceSendMs: Date.now() - lastSend.at };
+  }
+
   /**
    * Update the transport's `clientData`. Used by `useTriggerChatTransport`
    * to keep the latest value reachable from inside `startSession` and
@@ -1547,6 +1606,8 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             chatId,
             timestamp: Date.now(),
             resumed: options?.resumed ?? false,
+            lastEventId: state.lastEventId,
+            messageId: this.lastTurnSends.get(chatId)?.messageId,
           });
           let sawFirstChunk = false;
 
@@ -1634,6 +1695,8 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
                 chatId,
                 timestamp: Date.now(),
                 lastEventId: state.lastEventId,
+                sessionInEventId: headerValue(value.headers, SESSION_IN_EVENT_ID_HEADER),
+                ...this.turnAttribution(chatId),
               });
               state.isStreaming = false;
               this.notifySessionChange(chatId, state);
@@ -1659,7 +1722,14 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             if (value.chunk == null) continue;
             if (!sawFirstChunk) {
               sawFirstChunk = true;
-              this.emitEvent({ type: "first-chunk", chatId, timestamp: Date.now() });
+              this.emitEvent({
+                type: "first-chunk",
+                chatId,
+                timestamp: Date.now(),
+                chunkType: (value.chunk as { type?: string }).type,
+                lastEventId: value.id || undefined,
+                ...this.turnAttribution(chatId),
+              });
             }
             controller.enqueue(value.chunk as UIMessageChunk);
           }
@@ -1672,11 +1742,13 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             }
             return;
           }
+          const errorStatus = (error as { status?: unknown }).status;
           this.emitEvent({
             type: "stream-error",
             chatId,
             timestamp: Date.now(),
             error: error instanceof Error ? error : new Error(String(error)),
+            status: typeof errorStatus === "number" ? errorStatus : undefined,
           });
           controller.error(error);
         } finally {

packages/trigger-sdk/test/chat-transport-events.test.ts

@@ -83,9 +83,23 @@ describe("transport send events", () => {
     expect(sent.source).toBe("submit-message");
     expect(sent.durationMs).toBeGreaterThanOrEqual(0);
     expect(sent.timestamp).toBeGreaterThan(0);
+    expect(sent.partId).toMatch(/[0-9a-f-]{36}/);
+    expect(sent.bodyBytes).toBeGreaterThan(0);
 
     const connected = events[1] as Extract<ChatTransportEvent, { type: "stream-connected" }>;
     expect(connected.resumed).toBe(false);
+    expect(connected.messageId).toBe("u-1");
+
+    const firstChunk = events[2] as Extract<ChatTransportEvent, { type: "first-chunk" }>;
+    expect(firstChunk.chunkType).toBe("text-delta");
+    expect(firstChunk.lastEventId).toBe("1");
+    expect(firstChunk.messageId).toBe("u-1");
+    expect(firstChunk.sinceSendMs).toBeGreaterThanOrEqual(0);
+
+    const turnCompleted = events[3] as Extract<ChatTransportEvent, { type: "turn-completed" }>;
+    expect(turnCompleted.messageId).toBe("u-1");
+    expect(turnCompleted.sinceSendMs).toBeGreaterThanOrEqual(0);
+    expect(turnCompleted.lastEventId).toBe("2");
   });
 
   it("emits message-send-failed with the HTTP status when the append fails", async () => {