From f6de9a5820733676a412c0ca1a8c44793c26d59a Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Tue, 7 Jul 2026 21:50:13 +0100
Subject: [PATCH 1/9] feat(sdk): onEvent observability callback on the chat
 transport

Typed lifecycle events (message-sent, message-send-failed,
stream-connected, first-chunk, turn-completed, stream-error) emitted
from the transport's send and stream paths, including headStart and
steering sends the fetch override cannot see. Send events are terminal
and durable-ack semantics; callback exceptions are swallowed. The React
hook keeps the callback live across renders.
---
 .changeset/chat-transport-onevent.md          |  17 ++
 docs/ai-chat/frontend.mdx                     |  42 ++++
 docs/ai-chat/reference.mdx                    |  18 ++
 packages/trigger-sdk/src/v3/chat-react.ts     |   7 +-
 packages/trigger-sdk/src/v3/chat.ts           | 195 +++++++++++++++--
 .../test/chat-transport-events.test.ts        | 201 ++++++++++++++++++
 6 files changed, 463 insertions(+), 17 deletions(-)
 create mode 100644 .changeset/chat-transport-onevent.md
 create mode 100644 packages/trigger-sdk/test/chat-transport-events.test.ts

diff --git a/.changeset/chat-transport-onevent.md b/.changeset/chat-transport-onevent.md
new file mode 100644
index 00000000000..96e5ac737ae
--- /dev/null
+++ b/.changeset/chat-transport-onevent.md
@@ -0,0 +1,17 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Add an `onEvent` observability callback to `TriggerChatTransport` / `useTriggerChatTransport` that emits typed lifecycle events: `message-sent` and `message-send-failed` (durable send outcomes with source and duration), `stream-connected`, `first-chunk`, `turn-completed`, and `stream-error`. Together these make send-success metrics, time-to-first-token, and "sent but never answered" watchdogs a few lines of client code.
+
+```ts
+const transport = useTriggerChatTransport({
+  task: "my-chat",
+  accessToken,
+  onEvent: (event) => {
+    if (event.type === "message-sent") {
+      metrics.timing("chat.send_duration_ms", event.durationMs);
+    }
+  },
+});
+```
diff --git a/docs/ai-chat/frontend.mdx b/docs/ai-chat/frontend.mdx
index a1cc93a09b6..d074e2a17fb 100644
--- a/docs/ai-chat/frontend.mdx
+++ b/docs/ai-chat/frontend.mdx
@@ -574,3 +574,45 @@ baseURL: ({ endpoint }) =>
 ```
 
 For per-request control beyond URL routing (header injection, custom retries, tracing), pass a `fetch` override. See [Trusted edge signals](/ai-chat/patterns/trusted-edge-signals) for a full proxy walkthrough.
+
+## Monitoring message delivery
+
+`sendMessage` from `useChat` gives no feedback about whether the message actually reached the backend. The transport's `onEvent` callback closes that gap with typed lifecycle events (see the [event catalog](/ai-chat/reference#transport-events)) so you can record real metrics:
+
+```ts
+const transport = useTriggerChatTransport({
+  task: "my-chat",
+  accessToken: ({ chatId }) => mintChatAccessToken(chatId),
+  onEvent: (event) => {
+    switch (event.type) {
+      case "message-sent":
+        metrics.increment("chat.message_sent");
+        metrics.timing("chat.send_duration_ms", event.durationMs);
+        break;
+      case "message-send-failed":
+        metrics.increment("chat.message_send_failed", { status: event.status });
+        break;
+    }
+  },
+});
+```
+
+A `message-sent` event means the message is durably written to the session's input stream (the stream the agent consumes from), so it's a true "sent successfully" signal. Because send and response events share the same callback, "sent but never answered" becomes a small client-side watchdog:
+
+```ts
+const pending = new Map<string, ReturnType<typeof setTimeout>>();
+
+onEvent: (event) => {
+  if (event.type === "message-sent" && event.source === "submit-message") {
+    pending.set(event.chatId, setTimeout(() => {
+      metrics.increment("chat.sent_but_unanswered", { chatId: event.chatId });
+    }, 30_000));
+  }
+  if (event.type === "first-chunk" || event.type === "turn-completed" || event.type === "stream-error") {
+    clearTimeout(pending.get(event.chatId));
+    pending.delete(event.chatId);
+  }
+};
+```
+
+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.
diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx
index 192684d4163..2b2710d818d 100644
--- a/docs/ai-chat/reference.mdx
+++ b/docs/ai-chat/reference.mdx
@@ -630,10 +630,28 @@ Options for the frontend transport constructor and `useTriggerChatTransport` hoo
 | `clientData`           | Typed by `clientDataSchema`                                          | —                           | Default client data merged into per-turn `metadata` and threaded through `startSession`'s params (so the first run's `payload.metadata` matches per-turn `metadata`). Live-updated when the option value changes. |
 | `sessions`             | `Record<string, ChatSession>`                                        | —                           | Restore sessions from storage. See [ChatSession](#chatsession).             |
 | `onSessionChange`      | `(chatId, session \| null) => void`                                  | —                           | Fires when session state changes. `session` is the full `ChatSession` or `null` when the run ends. |
+| `onEvent`              | `(event: ChatTransportEvent) => void`                                | —                           | Observability callback for transport lifecycle events (send outcomes, stream connects, first chunk, turn completion). See [Transport events](#transport-events). Exceptions thrown by the callback are swallowed. |
 | `multiTab`             | `boolean`                                                            | `false`                     | Enable multi-tab claim coordination via `BroadcastChannel`. See [Frontend → multi-tab](/ai-chat/frontend#multi-tab-coordination). |
 | `watch`                | `boolean`                                                            | `false`                     | Read-only watcher mode — keep the SSE subscription open across `trigger:turn-complete` so a viewer sees turns 2, 3, … through one long-lived stream. |
 | `headStart`            | `string`                                                             | —                           | URL of a [`chat.headStart`](/ai-chat/fast-starts#head-start) route handler. When set, the FIRST message of a brand-new chat POSTs to this URL so step 1's LLM call runs in your warm process while the agent run boots in parallel. Subsequent turns bypass it. |
 
+### Transport events
+
+The `onEvent` callback receives a `ChatTransportEvent` (exported from `@trigger.dev/sdk/chat` and `@trigger.dev/sdk/chat/react`) for each lifecycle moment the transport observes. All events carry `chatId` and `timestamp`.
+
+| 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. |
+
+`source` identifies the send path: `"submit-message"`, `"regenerate-message"`, `"steer"` (`sendPendingMessage`), `"action"` (`sendAction`), `"stop"` (`stopGeneration`), or `"head-start"`.
+
+See [Monitoring message delivery](/ai-chat/frontend#monitoring-message-delivery) for the metrics and watchdog patterns these enable.
+
 ### `accessToken` callback
 
 The transport invokes `accessToken` whenever it needs a *fresh* session-scoped PAT — initial use after no PAT is cached, or after a 401/403 from any session-PAT-authed request. The callback's job is to **return a token, not to start a run.**
diff --git a/packages/trigger-sdk/src/v3/chat-react.ts b/packages/trigger-sdk/src/v3/chat-react.ts
index fb81a46d8b7..b6823d6dc21 100644
--- a/packages/trigger-sdk/src/v3/chat-react.ts
+++ b/packages/trigger-sdk/src/v3/chat-react.ts
@@ -50,6 +50,7 @@ export type UseTriggerChatTransportOptions<TTask extends AnyTask = AnyTask> = Om
 };
 
 export type { InferChatUIMessage };
+export type { ChatTransportEvent, ChatTransportSendSource } from "./chat.js";
 
 /**
  * React hook that creates and memoizes a `TriggerChatTransport` instance.
@@ -90,11 +91,15 @@ export function useTriggerChatTransport<TTask extends AnyTask = AnyTask>(
   }
 
   // Keep callbacks up to date without recreating the transport.
-  const { onSessionChange, clientData } = options;
+  const { onSessionChange, clientData, onEvent } = options;
   useEffect(() => {
     ref.current?.setOnSessionChange(onSessionChange);
   }, [onSessionChange]);
 
+  useEffect(() => {
+    ref.current?.setOnEvent(onEvent);
+  }, [onEvent]);
+
   // Keep `clientData` up to date so the transport's per-turn merge and
   // `startSession` callback both see the latest value without
   // reconstructing the transport.
diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts
index d7a71ee8421..a6059177ef2 100644
--- a/packages/trigger-sdk/src/v3/chat.ts
+++ b/packages/trigger-sdk/src/v3/chat.ts
@@ -75,6 +75,55 @@ export type ChatFetchOverride = (
   ctx: ChatTransportEndpointContext
 ) => Promise<Response>;
 
+/**
+ * Which transport path produced a send-lifecycle event: `useChat` sends,
+ * steering (`sendPendingMessage`), actions, stops, or the first-turn
+ * `headStart` POST.
+ */
+export type ChatTransportSendSource =
+  | "submit-message"
+  | "regenerate-message"
+  | "steer"
+  | "action"
+  | "stop"
+  | "head-start";
+
+/**
+ * Lifecycle events emitted through the transport's `onEvent` callback.
+ *
+ * Send events are terminal: `message-sent` fires when the append is
+ * durably acknowledged (2xx from the session input stream, after any
+ * internal token-refresh retries), `message-send-failed` when the send
+ * definitively failed. Response events follow the output stream:
+ * `stream-connected` when the SSE subscription opens, `first-chunk` on
+ * the first response chunk of a turn, `turn-completed` when the agent's
+ * turn-complete control record arrives, and `stream-error` when the
+ * stream fails unrecoverably.
+ */
+export type ChatTransportEvent =
+  | {
+      type: "message-sent";
+      chatId: string;
+      timestamp: number;
+      messageId?: string;
+      source: ChatTransportSendSource;
+      durationMs: number;
+    }
+  | {
+      type: "message-send-failed";
+      chatId: string;
+      timestamp: number;
+      messageId?: string;
+      source: ChatTransportSendSource;
+      error: Error;
+      status?: number;
+      durationMs: 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 };
+
 /**
  * Detect 401/403 from realtime/input-stream calls without relying on `instanceof`
  * (Vitest can load duplicate `@trigger.dev/core` copies, which breaks subclass checks).
@@ -318,6 +367,23 @@ export type TriggerChatTransportOptions<TClientData = unknown> = {
    */
   fetch?: ChatFetchOverride;
 
+  /**
+   * Observability callback for transport lifecycle events (sends,
+   * stream connects, first chunk, turn completion). See
+   * {@link ChatTransportEvent} for the event catalog and semantics.
+   * Exceptions thrown by the callback are swallowed.
+   *
+   * @example Record send metrics:
+   * ```ts
+   * onEvent: (event) => {
+   *   if (event.type === "message-sent") {
+   *     metrics.timing("chat.send", event.durationMs);
+   *   }
+   * },
+   * ```
+   */
+  onEvent?: (event: ChatTransportEvent) => void;
+
   /** Additional headers included in every API request. */
   headers?: Record<string, string>;
 
@@ -449,6 +515,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
   private _onSessionChange:
     | ((chatId: string, session: ChatSessionPersistedState | null) => void)
     | undefined;
+  private _onEvent: ((event: ChatTransportEvent) => void) | undefined;
 
   private sessions: Map<string, ChatSessionState> = new Map();
   private activeStreams: Map<string, AbortController> = new Map();
@@ -471,6 +538,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     this.streamTimeoutSeconds = options.streamTimeoutSeconds ?? DEFAULT_STREAM_TIMEOUT_SECONDS;
     this.defaultMetadata = options.clientData;
     this._onSessionChange = options.onSessionChange;
+    this._onEvent = options.onEvent;
     this.watchMode = options.watch ?? false;
     this.headStart = options.headStart;
 
@@ -628,7 +696,9 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       );
     };
 
-    await this.callWithAuthRetry(chatId, state, sendChatMessage);
+    await this.sendWithEvents(chatId, trigger, messageId ?? messages.at(-1)?.id, () =>
+      this.callWithAuthRetry(chatId, state, sendChatMessage)
+    );
 
     // Cancel any in-flight stream for this chat — the new turn supersedes it.
     const activeStream = this.activeStreams.get(chatId);
@@ -677,19 +747,31 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       metadata: args.metadata,
     };
 
-    const response = await fetch(this.headStart, {
-      method: "POST",
-      headers: {
-        "Content-Type": "application/json",
-        ...this.extraHeaders,
-      },
-      body: JSON.stringify(wirePayload),
-      signal: args.abortSignal,
-    });
+    let response!: Response;
+    await this.sendWithEvents(
+      args.chatId,
+      "head-start",
+      args.messageId ?? args.messages.at(-1)?.id,
+      async () => {
+        response = await fetch(this.headStart!, {
+          method: "POST",
+          headers: {
+            "Content-Type": "application/json",
+            ...this.extraHeaders,
+          },
+          body: JSON.stringify(wirePayload),
+          signal: args.abortSignal,
+        });
 
-    if (!response.ok) {
-      throw new Error(`chat.handover endpoint returned ${response.status} ${response.statusText}`);
-    }
+        if (!response.ok) {
+          const err = new Error(
+            `chat.handover endpoint returned ${response.status} ${response.statusText}`
+          ) as Error & { status: number };
+          err.status = response.status;
+          throw err;
+        }
+      }
+    );
     if (!response.body) {
       throw new Error("chat.handover endpoint returned no response body");
     }
@@ -814,7 +896,9 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     };
 
     try {
-      await this.callWithAuthRetry(chatId, state, send);
+      await this.sendWithEvents(chatId, "steer", message.id, () =>
+        this.callWithAuthRetry(chatId, state, send)
+      );
       return true;
     } catch {
       return false;
@@ -846,6 +930,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       : abortController.signal;
 
     return this.subscribeToSessionStream(state, abortSignal, options.chatId, {
+      resumed: true,
       sendStopOnAbort: !!options.abortSignal,
       // Reconnect-on-reload opts into the server's settled-peek shortcut
       // so the SSE doesn't hang for 60s when no turn is in flight. Active
@@ -875,7 +960,9 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     };
 
     try {
-      await this.callWithAuthRetry(chatId, state, send);
+      await this.sendWithEvents(chatId, "stop", undefined, () =>
+        this.callWithAuthRetry(chatId, state, send)
+      );
     } catch {
       return false;
     }
@@ -928,7 +1015,9 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       await this.appendInputChunk(chatId, token, body, partId);
     };
 
-    await this.callWithAuthRetry(chatId, state, send);
+    await this.sendWithEvents(chatId, "action", undefined, () =>
+      this.callWithAuthRetry(chatId, state, send)
+    );
 
     // Supersede any in-flight reader before subscribing — same as
     // `sendMessages`. Two concurrent readers both write `state.lastEventId`
@@ -973,6 +1062,55 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     this._onSessionChange = callback;
   }
 
+  /** Update the `onEvent` callback without recreating the transport. */
+  setOnEvent(callback: ((event: ChatTransportEvent) => void) | undefined): void {
+    this._onEvent = callback;
+  }
+
+  private emitEvent(event: ChatTransportEvent): void {
+    const cb = this._onEvent;
+    if (!cb) return;
+    try {
+      cb(event);
+    } catch {
+      // observability must never break the chat
+    }
+  }
+
+  /** Run a send op, emitting the terminal message-sent / message-send-failed event. */
+  private async sendWithEvents(
+    chatId: string,
+    source: ChatTransportSendSource,
+    messageId: string | undefined,
+    op: () => Promise<void>
+  ): Promise<void> {
+    const startedAt = Date.now();
+    try {
+      await op();
+      this.emitEvent({
+        type: "message-sent",
+        chatId,
+        timestamp: Date.now(),
+        messageId,
+        source,
+        durationMs: Date.now() - startedAt,
+      });
+    } 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,
+      });
+      throw error;
+    }
+  }
+
   /**
    * Update the transport's `clientData`. Used by `useTriggerChatTransport`
    * to keep the latest value reachable from inside `startSession` and
@@ -1229,6 +1367,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     options?: {
       sendStopOnAbort?: boolean;
       peekSettled?: boolean;
+      resumed?: boolean;
     }
   ): ReadableStream<UIMessageChunk> {
     const internalAbort = new AbortController();
@@ -1403,6 +1542,14 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             }
           }
 
+          this.emitEvent({
+            type: "stream-connected",
+            chatId,
+            timestamp: Date.now(),
+            resumed: options?.resumed ?? false,
+          });
+          let sawFirstChunk = false;
+
           while (true) {
             let value: {
               id: string;
@@ -1482,6 +1629,12 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
               if (refreshedToken) {
                 state.publicAccessToken = refreshedToken;
               }
+              this.emitEvent({
+                type: "turn-completed",
+                chatId,
+                timestamp: Date.now(),
+                lastEventId: state.lastEventId,
+              });
               state.isStreaming = false;
               this.notifySessionChange(chatId, state);
               this.coordinator?.release(chatId);
@@ -1504,6 +1657,10 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             // unwrapped from the S2 record envelope (the parser does the
             // JSON unwrap). Drop empty/malformed payloads defensively.
             if (value.chunk == null) continue;
+            if (!sawFirstChunk) {
+              sawFirstChunk = true;
+              this.emitEvent({ type: "first-chunk", chatId, timestamp: Date.now() });
+            }
             controller.enqueue(value.chunk as UIMessageChunk);
           }
         } catch (error) {
@@ -1515,6 +1672,12 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             }
             return;
           }
+          this.emitEvent({
+            type: "stream-error",
+            chatId,
+            timestamp: Date.now(),
+            error: error instanceof Error ? error : new Error(String(error)),
+          });
           controller.error(error);
         } finally {
           teardownWakeListeners();
diff --git a/packages/trigger-sdk/test/chat-transport-events.test.ts b/packages/trigger-sdk/test/chat-transport-events.test.ts
new file mode 100644
index 00000000000..22d94c55cf8
--- /dev/null
+++ b/packages/trigger-sdk/test/chat-transport-events.test.ts
@@ -0,0 +1,201 @@
+import { describe, expect, it } from "vitest";
+import type { UIMessage } from "ai";
+import {
+  TriggerChatTransport,
+  type ChatTransportEvent,
+  type TriggerChatTransportOptions,
+} from "../src/v3/chat.js";
+
+// ── Helpers ────────────────────────────────────────────────────────────
+
+function user(text: string, id: string): UIMessage {
+  return { id, role: "user", parts: [{ type: "text", text }] };
+}
+
+function jsonOk(): Response {
+  return new Response("{}", { status: 200 });
+}
+
+/** Build a `text/event-stream` Response from raw SSE text (v1 wire). */
+function sseResponse(frames: string): Response {
+  return new Response(frames, {
+    status: 200,
+    headers: { "Content-Type": "text/event-stream" },
+  });
+}
+
+/** SSE body: one data chunk then a legacy turn-complete control chunk. */
+const SSE_ONE_TURN = [
+  `id: 1`,
+  `data: {"type":"text-delta","id":"t1","delta":"hello"}`,
+  ``,
+  `id: 2`,
+  `data: {"type":"trigger:turn-complete"}`,
+  ``,
+  ``,
+].join("\n");
+
+async function readAll(stream: ReadableStream<unknown>): Promise<unknown[]> {
+  const out: unknown[] = [];
+  const reader = stream.getReader();
+  while (true) {
+    const next = await reader.read();
+    if (next.done) return out;
+    out.push(next.value);
+  }
+}
+
+function makeTransport(overrides: Partial<TriggerChatTransportOptions> = {}) {
+  const events: ChatTransportEvent[] = [];
+  const transport = new TriggerChatTransport({
+    task: "test-task",
+    accessToken: async () => "tok_test",
+    sessions: { c1: { publicAccessToken: "tok_test", isStreaming: false } },
+    onEvent: (event) => events.push(event),
+    fetch: async (_url, _init, ctx) =>
+      ctx.endpoint === "in" ? jsonOk() : sseResponse(SSE_ONE_TURN),
+    ...overrides,
+  });
+  return { transport, events };
+}
+
+// ── Tests ──────────────────────────────────────────────────────────────
+
+describe("transport send events", () => {
+  it("emits message-sent for a successful submit and the full stream lifecycle", async () => {
+    const { transport, events } = makeTransport();
+
+    const stream = await transport.sendMessages({
+      trigger: "submit-message",
+      chatId: "c1",
+      messageId: undefined,
+      messages: [user("hi", "u-1")],
+      abortSignal: undefined,
+    });
+    await readAll(stream);
+
+    const types = events.map((e) => e.type);
+    expect(types).toEqual(["message-sent", "stream-connected", "first-chunk", "turn-completed"]);
+
+    const sent = events[0] as Extract<ChatTransportEvent, { type: "message-sent" }>;
+    expect(sent.chatId).toBe("c1");
+    expect(sent.messageId).toBe("u-1");
+    expect(sent.source).toBe("submit-message");
+    expect(sent.durationMs).toBeGreaterThanOrEqual(0);
+    expect(sent.timestamp).toBeGreaterThan(0);
+
+    const connected = events[1] as Extract<ChatTransportEvent, { type: "stream-connected" }>;
+    expect(connected.resumed).toBe(false);
+  });
+
+  it("emits message-send-failed with the HTTP status when the append fails", async () => {
+    const { transport, events } = makeTransport({
+      fetch: async () => new Response("too large", { status: 413 }),
+    });
+
+    await expect(
+      transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "c1",
+        messageId: undefined,
+        messages: [user("hi", "u-1")],
+        abortSignal: undefined,
+      })
+    ).rejects.toThrow();
+
+    expect(events).toHaveLength(1);
+    const failed = events[0] as Extract<ChatTransportEvent, { type: "message-send-failed" }>;
+    expect(failed.type).toBe("message-send-failed");
+    expect(failed.source).toBe("submit-message");
+    expect(failed.messageId).toBe("u-1");
+    expect(failed.status).toBe(413);
+    expect(failed.error).toBeInstanceOf(Error);
+  });
+
+  it("emits steer events from sendPendingMessage without changing its boolean result", async () => {
+    const { transport, events } = makeTransport();
+    const ok = await transport.sendPendingMessage("c1", user("steer", "u-2"));
+    expect(ok).toBe(true);
+    expect(events[0]).toMatchObject({ type: "message-sent", source: "steer", messageId: "u-2" });
+
+    const failing = makeTransport({ fetch: async () => new Response("nope", { status: 500 }) });
+    const notOk = await failing.transport.sendPendingMessage("c1", user("steer", "u-3"));
+    expect(notOk).toBe(false);
+    expect(failing.events[0]).toMatchObject({
+      type: "message-send-failed",
+      source: "steer",
+      status: 500,
+    });
+  });
+
+  it("emits action and stop send events", async () => {
+    const { transport, events } = makeTransport();
+
+    const stream = await transport.sendAction("c1", { type: "undo" });
+    await readAll(stream);
+    expect(events[0]).toMatchObject({ type: "message-sent", source: "action" });
+
+    events.length = 0;
+    const stopped = await transport.stopGeneration("c1");
+    expect(stopped).toBe(true);
+    expect(events[0]).toMatchObject({ type: "message-sent", source: "stop" });
+  });
+
+  it("swallows exceptions thrown by the onEvent callback", async () => {
+    const { transport } = makeTransport({
+      onEvent: () => {
+        throw new Error("observer exploded");
+      },
+    });
+
+    const stream = await transport.sendMessages({
+      trigger: "submit-message",
+      chatId: "c1",
+      messageId: undefined,
+      messages: [user("hi", "u-1")],
+      abortSignal: undefined,
+    });
+    const chunks = await readAll(stream);
+    expect(chunks.length).toBeGreaterThan(0);
+  });
+});
+
+describe("transport stream events", () => {
+  it("marks reconnectToStream subscriptions as resumed", async () => {
+    const { transport, events } = makeTransport({
+      sessions: {
+        c1: { publicAccessToken: "tok_test", isStreaming: true, lastEventId: "1" },
+      },
+    });
+
+    const stream = await transport.reconnectToStream({ chatId: "c1" });
+    expect(stream).not.toBeNull();
+    await readAll(stream!);
+
+    const connected = events.find((e) => e.type === "stream-connected") as Extract<
+      ChatTransportEvent,
+      { type: "stream-connected" }
+    >;
+    expect(connected.resumed).toBe(true);
+    expect(events.some((e) => e.type === "turn-completed")).toBe(true);
+  });
+
+  it("emits stream-error when the output stream fails unrecoverably", async () => {
+    const { transport, events } = makeTransport({
+      fetch: async (_url, _init, ctx) =>
+        ctx.endpoint === "in" ? jsonOk() : new Response("gone", { status: 400 }),
+    });
+
+    const stream = await transport.sendMessages({
+      trigger: "submit-message",
+      chatId: "c1",
+      messageId: undefined,
+      messages: [user("hi", "u-1")],
+      abortSignal: undefined,
+    });
+    await expect(readAll(stream)).rejects.toThrow();
+
+    expect(events.some((e) => e.type === "stream-error")).toBe(true);
+    expect(events.some((e) => e.type === "stream-connected")).toBe(false);
+  });
+});

From 54fa7907d026a5acb6eb00131760075208dd8f64 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Tue, 7 Jul 2026 22:09:00 +0100
Subject: [PATCH 2/9] 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.
---
 docs/ai-chat/frontend.mdx                     |   2 +-
 docs/ai-chat/reference.mdx                    |  14 +-
 packages/trigger-sdk/src/v3/chat.ts           | 142 +++++++++++++-----
 .../test/chat-transport-events.test.ts        |  14 ++
 4 files changed, 130 insertions(+), 42 deletions(-)

diff --git a/docs/ai-chat/frontend.mdx b/docs/ai-chat/frontend.mdx
index d074e2a17fb..ba380e053c4 100644
--- a/docs/ai-chat/frontend.mdx
+++ b/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.
diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx
index 2b2710d818d..340422050fb 100644
--- a/docs/ai-chat/reference.mdx
+++ b/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
diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts
index a6059177ef2..66d6074faa2 100644
--- a/packages/trigger-sdk/src/v3/chat.ts
+++ b/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,10 +793,14 @@ 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",
@@ -759,7 +808,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
             "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,19 +1128,24 @@ 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;
@@ -1101,16 +1153,23 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
         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 {
diff --git a/packages/trigger-sdk/test/chat-transport-events.test.ts b/packages/trigger-sdk/test/chat-transport-events.test.ts
index 22d94c55cf8..b9e78193888 100644
--- a/packages/trigger-sdk/test/chat-transport-events.test.ts
+++ b/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 () => {

From 27fcb0d59fa413978729a2c968e3eccd85a940e2 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Tue, 7 Jul 2026 22:42:18 +0100
Subject: [PATCH 3/9] chore: show the enriched events in the changeset example

---
 .changeset/chat-transport-onevent.md | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/.changeset/chat-transport-onevent.md b/.changeset/chat-transport-onevent.md
index 96e5ac737ae..59ba10cea10 100644
--- a/.changeset/chat-transport-onevent.md
+++ b/.changeset/chat-transport-onevent.md
@@ -2,7 +2,7 @@
 "@trigger.dev/sdk": patch
 ---
 
-Add an `onEvent` observability callback to `TriggerChatTransport` / `useTriggerChatTransport` that emits typed lifecycle events: `message-sent` and `message-send-failed` (durable send outcomes with source and duration), `stream-connected`, `first-chunk`, `turn-completed`, and `stream-error`. Together these make send-success metrics, time-to-first-token, and "sent but never answered" watchdogs a few lines of client code.
+Add an `onEvent` observability callback to `TriggerChatTransport` / `useTriggerChatTransport` that emits typed lifecycle events: `message-sent` and `message-send-failed` (durable send outcomes with source, duration, payload size, and the append's idempotency key), `stream-connected`, `first-chunk` and `turn-completed` (with built-in time-to-first-token and turn latency via `sinceSendMs`), and `stream-error`. Send-success metrics, TTFT, and "sent but never answered" watchdogs become a few lines of client code.
 
 ```ts
 const transport = useTriggerChatTransport({
@@ -12,6 +12,12 @@ const transport = useTriggerChatTransport({
     if (event.type === "message-sent") {
       metrics.timing("chat.send_duration_ms", event.durationMs);
     }
+    if (event.type === "first-chunk") {
+      metrics.timing("chat.ttft_ms", event.sinceSendMs ?? 0);
+    }
+    if (event.type === "message-send-failed") {
+      metrics.increment("chat.send_failed", { status: event.status });
+    }
   },
 });
 ```

From 259fb01f68eba2406736d0e1ca58683a0dd7d92a Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Tue, 7 Jul 2026 22:45:49 +0100
Subject: [PATCH 4/9] chore: tighten the changeset

---
 .changeset/chat-transport-onevent.md | 21 +++++----------------
 1 file changed, 5 insertions(+), 16 deletions(-)

diff --git a/.changeset/chat-transport-onevent.md b/.changeset/chat-transport-onevent.md
index 59ba10cea10..b5fffc34357 100644
--- a/.changeset/chat-transport-onevent.md
+++ b/.changeset/chat-transport-onevent.md
@@ -2,22 +2,11 @@
 "@trigger.dev/sdk": patch
 ---
 
-Add an `onEvent` observability callback to `TriggerChatTransport` / `useTriggerChatTransport` that emits typed lifecycle events: `message-sent` and `message-send-failed` (durable send outcomes with source, duration, payload size, and the append's idempotency key), `stream-connected`, `first-chunk` and `turn-completed` (with built-in time-to-first-token and turn latency via `sinceSendMs`), and `stream-error`. Send-success metrics, TTFT, and "sent but never answered" watchdogs become a few lines of client code.
+Add an `onEvent` callback to `TriggerChatTransport` / `useTriggerChatTransport` that emits typed lifecycle events for sends, stream connects, first chunk, and turn completion. Send-success metrics, time-to-first-token, and "sent but never answered" watchdogs become a few lines of client code.
 
 ```ts
-const transport = useTriggerChatTransport({
-  task: "my-chat",
-  accessToken,
-  onEvent: (event) => {
-    if (event.type === "message-sent") {
-      metrics.timing("chat.send_duration_ms", event.durationMs);
-    }
-    if (event.type === "first-chunk") {
-      metrics.timing("chat.ttft_ms", event.sinceSendMs ?? 0);
-    }
-    if (event.type === "message-send-failed") {
-      metrics.increment("chat.send_failed", { status: event.status });
-    }
-  },
-});
+onEvent: (event) => {
+  if (event.type === "message-sent") metrics.timing("chat.send_ms", event.durationMs);
+  if (event.type === "first-chunk") metrics.timing("chat.ttft_ms", event.sinceSendMs ?? 0);
+},
 ```

From 55ec228871b348452bee3094a56db0d7e88a27a7 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Tue, 7 Jul 2026 23:16:10 +0100
Subject: [PATCH 5/9] docs(ai-chat): drop the unbounded chatId metric tag from
 the watchdog example

---
 docs/ai-chat/frontend.mdx | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/docs/ai-chat/frontend.mdx b/docs/ai-chat/frontend.mdx
index ba380e053c4..2361bce1053 100644
--- a/docs/ai-chat/frontend.mdx
+++ b/docs/ai-chat/frontend.mdx
@@ -605,7 +605,9 @@ const pending = new Map<string, ReturnType<typeof setTimeout>>();
 onEvent: (event) => {
   if (event.type === "message-sent" && event.source === "submit-message") {
     pending.set(event.chatId, setTimeout(() => {
-      metrics.increment("chat.sent_but_unanswered", { chatId: event.chatId });
+      // Log the chatId; don't tag the metric with it (unbounded cardinality).
+      metrics.increment("chat.sent_but_unanswered");
+      console.warn("sent but unanswered", event.chatId);
     }, 30_000));
   }
   if (event.type === "first-chunk" || event.type === "turn-completed" || event.type === "stream-error") {

From 78deb6c186e2e93c9f730b894e0b7b7f1f121a72 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Wed, 8 Jul 2026 09:28:52 +0100
Subject: [PATCH 6/9] docs(ai-chat): note the watchdog map scope and
 intentional chat keying

---
 docs/ai-chat/frontend.mdx | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/docs/ai-chat/frontend.mdx b/docs/ai-chat/frontend.mdx
index 2361bce1053..0d7b068fb72 100644
--- a/docs/ai-chat/frontend.mdx
+++ b/docs/ai-chat/frontend.mdx
@@ -600,6 +600,8 @@ const transport = useTriggerChatTransport({
 A `message-sent` event means the message is durably written to the session's input stream (the stream the agent consumes from), so it's a true "sent successfully" signal. Because send and response events share the same callback, "sent but never answered" becomes a small client-side watchdog:
 
 ```ts
+// Module scope (or a ref) so re-renders don't recreate it. Keyed by chatId
+// on purpose: a new send on the same chat supersedes the in-flight turn.
 const pending = new Map<string, ReturnType<typeof setTimeout>>();
 
 onEvent: (event) => {

From 43d4a9b22afff068a6e5c32fa0250f7faf937542 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Wed, 8 Jul 2026 09:45:40 +0100
Subject: [PATCH 7/9] fix(sdk): re-arm the first-chunk event per turn on
 watch-mode streams

---
 packages/trigger-sdk/src/v3/chat.ts           |  2 ++
 .../test/chat-transport-events.test.ts        | 31 +++++++++++++++++++
 2 files changed, 33 insertions(+)

diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts
index 66d6074faa2..a05b6e9b772 100644
--- a/packages/trigger-sdk/src/v3/chat.ts
+++ b/packages/trigger-sdk/src/v3/chat.ts
@@ -1705,6 +1705,8 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
                 lastEventId: state.lastEventId,
               });
 
+              // Re-arm per-turn events for watch mode's next turn.
+              sawFirstChunk = false;
               if (this.watchMode) continue;
 
               internalAbort.abort();
diff --git a/packages/trigger-sdk/test/chat-transport-events.test.ts b/packages/trigger-sdk/test/chat-transport-events.test.ts
index b9e78193888..37d1c527cd1 100644
--- a/packages/trigger-sdk/test/chat-transport-events.test.ts
+++ b/packages/trigger-sdk/test/chat-transport-events.test.ts
@@ -194,6 +194,37 @@ describe("transport stream events", () => {
     expect(events.some((e) => e.type === "turn-completed")).toBe(true);
   });
 
+  it("re-arms first-chunk per turn on a watch-mode stream", async () => {
+    const TWO_TURNS = [
+      `id: 1`,
+      `data: {"type":"text-delta","id":"t1","delta":"turn one"}`,
+      ``,
+      `id: 2`,
+      `data: {"type":"trigger:turn-complete"}`,
+      ``,
+      `id: 3`,
+      `data: {"type":"text-delta","id":"t2","delta":"turn two"}`,
+      ``,
+      `id: 4`,
+      `data: {"type":"trigger:turn-complete"}`,
+      ``,
+      ``,
+    ].join("\n");
+
+    const { transport, events } = makeTransport({
+      watch: true,
+      sessions: { c1: { publicAccessToken: "tok_test", isStreaming: true } },
+      fetch: async (_url, _init, ctx) =>
+        ctx.endpoint === "in" ? jsonOk() : sseResponse(TWO_TURNS),
+    });
+
+    const stream = await transport.reconnectToStream({ chatId: "c1" });
+    await readAll(stream!);
+
+    expect(events.filter((e) => e.type === "first-chunk")).toHaveLength(2);
+    expect(events.filter((e) => e.type === "turn-completed")).toHaveLength(2);
+  });
+
   it("emits stream-error when the output stream fails unrecoverably", async () => {
     const { transport, events } = makeTransport({
       fetch: async (_url, _init, ctx) =>

From 2434ee0f61921da6e67bb6a40af60b572f55a574 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Wed, 8 Jul 2026 09:53:32 +0100
Subject: [PATCH 8/9] feat(sdk): emit stream lifecycle events on the headStart
 first-turn path

The handover response stream now emits stream-connected, first-chunk,
and turn-completed like the session subscribe path, so headStart turns
get time-to-first-token and turn latency instead of an event gap on the
exact path that exists to improve first-turn TTFT.
---
 packages/trigger-sdk/src/v3/chat.ts           | 28 +++++++++++
 .../test/chat-transport-events.test.ts        | 47 +++++++++++++++++++
 2 files changed, 75 insertions(+)

diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts
index a05b6e9b772..951b3a71be3 100644
--- a/packages/trigger-sdk/src/v3/chat.ts
+++ b/packages/trigger-sdk/src/v3/chat.ts
@@ -879,6 +879,17 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
         notifyChange(chatId, state);
       }
     };
+    const emit = (event: ChatTransportEvent) => this.emitEvent(event);
+    const attribution = () => this.turnAttribution(chatId);
+    let sawFirstChunk = false;
+
+    emit({
+      type: "stream-connected",
+      chatId,
+      timestamp: Date.now(),
+      resumed: false,
+      messageId: this.lastTurnSends.get(chatId)?.messageId,
+    });
 
     return response.body
       .pipeThrough(new TextDecoderStream())
@@ -890,6 +901,13 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
               const type = (chunk as { type?: unknown }).type;
               if (type === TRIGGER_TURN_COMPLETE) {
                 clearStreaming();
+                emit({
+                  type: "turn-completed",
+                  chatId,
+                  timestamp: Date.now(),
+                  lastEventId: sessions.get(chatId)?.lastEventId,
+                  ...attribution(),
+                });
                 return; // drop — not a real UIMessageChunk
               }
               if (type === TRIGGER_SESSION_STATE) {
@@ -900,6 +918,16 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
                 return; // drop
               }
             }
+            if (!sawFirstChunk) {
+              sawFirstChunk = true;
+              emit({
+                type: "first-chunk",
+                chatId,
+                timestamp: Date.now(),
+                chunkType: (chunk as { type?: string } | undefined)?.type,
+                ...attribution(),
+              });
+            }
             controller.enqueue(chunk);
           },
           flush() {
diff --git a/packages/trigger-sdk/test/chat-transport-events.test.ts b/packages/trigger-sdk/test/chat-transport-events.test.ts
index 37d1c527cd1..57b029b442b 100644
--- a/packages/trigger-sdk/test/chat-transport-events.test.ts
+++ b/packages/trigger-sdk/test/chat-transport-events.test.ts
@@ -225,6 +225,53 @@ describe("transport stream events", () => {
     expect(events.filter((e) => e.type === "turn-completed")).toHaveLength(2);
   });
 
+  it("emits the full lifecycle on the headStart first-turn path", async () => {
+    const handoverSse = [
+      `data: {"type":"start","messageId":"a-1"}`,
+      ``,
+      `data: {"type":"text-delta","id":"t1","delta":"warm"}`,
+      ``,
+      `data: {"type":"trigger:turn-complete"}`,
+      ``,
+      ``,
+    ].join("\n");
+    const originalFetch = globalThis.fetch;
+    globalThis.fetch = (async () =>
+      new Response(handoverSse, {
+        status: 200,
+        headers: {
+          "Content-Type": "text/event-stream",
+          "X-Trigger-Chat-Access-Token": "tok_handover",
+        },
+      })) as typeof fetch;
+
+    try {
+      const { transport, events } = makeTransport({
+        headStart: "/api/chat",
+        sessions: {},
+      });
+
+      const stream = await transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "c-hs",
+        messageId: undefined,
+        messages: [user("hi", "u-hs")],
+        abortSignal: undefined,
+      });
+      await readAll(stream);
+
+      const types = events.map((e) => e.type);
+      expect(types).toEqual(["message-sent", "stream-connected", "first-chunk", "turn-completed"]);
+      expect(events[0]).toMatchObject({ source: "head-start", messageId: "u-hs" });
+      const firstChunk = events[2] as Extract<ChatTransportEvent, { type: "first-chunk" }>;
+      expect(firstChunk.chunkType).toBe("start");
+      expect(firstChunk.messageId).toBe("u-hs");
+      expect(firstChunk.sinceSendMs).toBeGreaterThanOrEqual(0);
+    } finally {
+      globalThis.fetch = originalFetch;
+    }
+  });
+
   it("emits stream-error when the output stream fails unrecoverably", async () => {
     const { transport, events } = makeTransport({
       fetch: async (_url, _init, ctx) =>

From a91725096971fedc944346f74cfe5ed9898e9cb0 Mon Sep 17 00:00:00 2001
From: Eric Allam <eallam@icloud.com>
Date: Wed, 8 Jul 2026 10:16:23 +0100
Subject: [PATCH 9/9] fix(sdk): emit stream-error on the headStart response
 path

---
 packages/trigger-sdk/src/v3/chat.ts           | 40 ++++++++++++++++++-
 .../test/chat-transport-events.test.ts        | 37 +++++++++++++++++
 2 files changed, 76 insertions(+), 1 deletion(-)

diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts
index 951b3a71be3..f625155189c 100644
--- a/packages/trigger-sdk/src/v3/chat.ts
+++ b/packages/trigger-sdk/src/v3/chat.ts
@@ -891,7 +891,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       messageId: this.lastTurnSends.get(chatId)?.messageId,
     });
 
-    return response.body
+    const piped = response.body
       .pipeThrough(new TextDecoderStream())
       .pipeThrough(parseUIMessageSseTransform())
       .pipeThrough(
@@ -935,6 +935,44 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
           },
         })
       );
+
+    // Pump wrapper: a TransformStream can't observe upstream errors, so
+    // read here to emit stream-error (aborts close cleanly, matching the
+    // session subscribe path).
+    const pipedReader = piped.getReader();
+    return new ReadableStream<UIMessageChunk>({
+      async pull(controller) {
+        try {
+          const next = await pipedReader.read();
+          if (next.done) {
+            controller.close();
+            return;
+          }
+          controller.enqueue(next.value);
+        } catch (error) {
+          if (error instanceof Error && error.name === "AbortError") {
+            try {
+              controller.close();
+            } catch {
+              /* already closed */
+            }
+            return;
+          }
+          const errorStatus = (error as { status?: unknown }).status;
+          emit({
+            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);
+        }
+      },
+      cancel(reason) {
+        return pipedReader.cancel(reason);
+      },
+    });
   }
 
   /**
diff --git a/packages/trigger-sdk/test/chat-transport-events.test.ts b/packages/trigger-sdk/test/chat-transport-events.test.ts
index 57b029b442b..ded1895dfd8 100644
--- a/packages/trigger-sdk/test/chat-transport-events.test.ts
+++ b/packages/trigger-sdk/test/chat-transport-events.test.ts
@@ -272,6 +272,43 @@ describe("transport stream events", () => {
     }
   });
 
+  it("emits stream-error when the headStart response body fails mid-read", async () => {
+    const encoder = new TextEncoder();
+    const failingBody = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(encoder.encode(`data: {"type":"text-delta","id":"t1","delta":"w"}\n\n`));
+        controller.error(new Error("network drop"));
+      },
+    });
+    const originalFetch = globalThis.fetch;
+    globalThis.fetch = (async () =>
+      new Response(failingBody, {
+        status: 200,
+        headers: {
+          "Content-Type": "text/event-stream",
+          "X-Trigger-Chat-Access-Token": "tok_handover",
+        },
+      })) as typeof fetch;
+
+    try {
+      const { transport, events } = makeTransport({ headStart: "/api/chat", sessions: {} });
+      const stream = await transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "c-hs-err",
+        messageId: undefined,
+        messages: [user("hi", "u-hs")],
+        abortSignal: undefined,
+      });
+      await expect(readAll(stream)).rejects.toThrow("network drop");
+
+      const streamError = events.find((e) => e.type === "stream-error");
+      expect(streamError).toBeDefined();
+      expect(events.some((e) => e.type === "turn-completed")).toBe(false);
+    } finally {
+      globalThis.fetch = originalFetch;
+    }
+  });
+
   it("emits stream-error when the output stream fails unrecoverably", async () => {
     const { transport, events } = makeTransport({
       fetch: async (_url, _init, ctx) =>