feat(sdk): onEvent observability callback on the chat transport (#4187)

## Summary

`sendMessage` from `useChat` gives no feedback about whether a message
actually reached the backend, and the `fetch` override is wire-level: it
requires knowing endpoint semantics, cannot attribute requests to
messages, and misses the headStart first-turn POST entirely. This adds a
typed `onEvent` observability callback to `TriggerChatTransport` /
`useTriggerChatTransport` so send-success metrics, time-to-first-token,
and "sent but never answered" watchdogs become a few lines of client
code.

## Example

```ts
const transport = useTriggerChatTransport({
  task: "my-chat",
  accessToken: ({ chatId }) => mintChatAccessToken(chatId),
  onEvent: (event) => {
    switch (event.type) {
      case "message-sent":
        // Durably acknowledged by the session's input stream, not just "request accepted".
        metrics.increment("chat.message_sent", { source: event.source });
        metrics.timing("chat.send_duration_ms", event.durationMs);
        break;
      case "message-send-failed":
        metrics.increment("chat.message_send_failed", { status: event.status });
        break;
      case "first-chunk":
        metrics.timing("chat.ttft_ms", event.sinceSendMs ?? 0);
        break;
      case "turn-completed":
        metrics.timing("chat.turn_duration_ms", event.sinceSendMs ?? 0);
        break;
    }
  },
});
```

## Design

One callback, one discriminated union (`ChatTransportEvent`):

- `message-sent` / `message-send-failed`: terminal send outcomes with
`messageId`, a `source` discriminator (submit, regenerate, steer,
action, stop, head-start), `durationMs`, `bodyBytes`, the append's
idempotency key (`partId`, also stored on the server-side record), and
error + HTTP status on failure. `message-sent` means the append was
durably acknowledged, after any internal token-refresh retries.
- `stream-connected` (with a `resumed` flag and the cursor it connected
from), `first-chunk` (chunk type plus `sinceSendMs` for
time-to-first-token), `turn-completed` (`sinceSendMs` full-turn latency
and the agent's committed input cursor), and `stream-error` follow the
response side, so a send can be paired with the answer that should
follow it. `messageId` on response events is client-side attribution
from the last turn-producing send on that chat.

Emissions sit at the transport's existing choke points, covering every
send path uniformly (including steering and headStart, which the fetch
override cannot observe). Exceptions thrown by the callback are
swallowed: observability can never break the chat. The React hook keeps
the callback live across renders instead of freezing the first-render
closure.

## Verification

Unit tests drive the transport directly with the `fetch` override as the
network stub (send success/failure per source, stream lifecycle, resumed
flag, field enrichment, callback exceptions swallowed). Verified
end-to-end against a realistic metrics setup in the ai-chat reference
app (counters, send-duration and TTFT histograms, and both watchdogs
built purely on these events): a healthy two-turn chat produces exactly
the expected event sequence and TTFT values; an oversized append records
`message_send_failed` with status 413; and killing the worker after a
durable send fires both `sent_but_no_stream` and `sent_but_unanswered`,
reproducing and detecting the "message disappeared" failure mode that
motivated this feature.

Author: ericallam

.changeset/chat-transport-onevent.md

@@ -0,0 +1,12 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+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
+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);
+},
+```

docs/ai-chat/frontend.mdx

@@ -574,3 +574,49 @@ 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
+// 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) => {
+  if (event.type === "message-sent" && event.source === "submit-message") {
+    pending.set(event.chatId, setTimeout(() => {
+      // 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") {
+    clearTimeout(pending.get(event.chatId));
+    pending.delete(event.chatId);
+  }
+};
+```
+
+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

@@ -630,10 +630,30 @@ 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`, `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
 
 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.**

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.