diff --git a/.changeset/chat-transport-onevent.md b/.changeset/chat-transport-onevent.md new file mode 100644 index 00000000000..b5fffc34357 --- /dev/null +++ b/.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); +}, +``` diff --git a/docs/ai-chat/frontend.mdx b/docs/ai-chat/frontend.mdx index a1cc93a09b6..0d7b068fb72 100644 --- a/docs/ai-chat/frontend.mdx +++ b/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>(); + +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. diff --git a/docs/ai-chat/reference.mdx b/docs/ai-chat/reference.mdx index 192684d4163..340422050fb 100644 --- a/docs/ai-chat/reference.mdx +++ b/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` | — | 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.** 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 = 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( } // 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..f625155189c 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"; @@ -75,6 +80,89 @@ export type ChatFetchOverride = ( ctx: ChatTransportEndpointContext ) => Promise; +/** + * 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; + /** 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"; + chatId: string; + timestamp: number; + messageId?: string; + source: ChatTransportSendSource; + 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: "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` * (Vitest can load duplicate `@trigger.dev/core` copies, which breaks subclass checks). @@ -318,6 +406,23 @@ export type TriggerChatTransportOptions = { */ 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; @@ -449,10 +554,14 @@ export class TriggerChatTransport implements ChatTransport { private _onSessionChange: | ((chatId: string, session: ChatSessionPersistedState | null) => void) | undefined; + private _onEvent: ((event: ChatTransportEvent) => void) | undefined; private sessions: Map = new Map(); private activeStreams: Map = new Map(); private pendingStarts: Map> = new Map(); + // Last turn-producing send per chat — attribution source for the + // response-side events' `messageId` / `sinceSendMs`. + private lastTurnSends: Map = new Map(); constructor(options: TriggerChatTransportOptions) { this.taskId = options.task; @@ -471,6 +580,7 @@ export class TriggerChatTransport implements ChatTransport { 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; @@ -619,16 +729,21 @@ export class TriggerChatTransport implements ChatTransport { // 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.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. const activeStream = this.activeStreams.get(chatId); @@ -677,19 +792,35 @@ export class TriggerChatTransport implements ChatTransport { metadata: args.metadata, }; - const response = await fetch(this.headStart, { - method: "POST", - headers: { - "Content-Type": "application/json", - ...this.extraHeaders, + let response!: Response; + const serializedBody = JSON.stringify(wirePayload); + await this.sendWithEvents( + args.chatId, + "head-start", + { + messageId: args.messageId ?? args.messages.at(-1)?.id, + bodyBytes: byteLength(serializedBody), }, - body: JSON.stringify(wirePayload), - signal: args.abortSignal, - }); + async () => { + response = await fetch(this.headStart!, { + method: "POST", + headers: { + "Content-Type": "application/json", + ...this.extraHeaders, + }, + body: serializedBody, + 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"); } @@ -748,8 +879,19 @@ export class TriggerChatTransport implements ChatTransport { notifyChange(chatId, state); } }; + const emit = (event: ChatTransportEvent) => this.emitEvent(event); + const attribution = () => this.turnAttribution(chatId); + let sawFirstChunk = false; - return response.body + emit({ + type: "stream-connected", + chatId, + timestamp: Date.now(), + resumed: false, + messageId: this.lastTurnSends.get(chatId)?.messageId, + }); + + const piped = response.body .pipeThrough(new TextDecoderStream()) .pipeThrough(parseUIMessageSseTransform()) .pipeThrough( @@ -759,6 +901,13 @@ export class TriggerChatTransport implements ChatTransport { 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) { @@ -769,6 +918,16 @@ export class TriggerChatTransport implements ChatTransport { 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() { @@ -776,6 +935,44 @@ export class TriggerChatTransport implements ChatTransport { }, }) ); + + // 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({ + 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); + }, + }); } /** @@ -804,17 +1001,18 @@ export class TriggerChatTransport implements ChatTransport { }; 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.callWithAuthRetry(chatId, state, send); + await this.sendWithEvents( + chatId, + "steer", + { messageId: message.id, partId, bodyBytes: byteLength(serializedBody) }, + () => this.callWithAuthRetry(chatId, state, send) + ); return true; } catch { return false; @@ -846,6 +1044,7 @@ export class TriggerChatTransport implements ChatTransport { : 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 @@ -865,17 +1064,18 @@ export class TriggerChatTransport implements ChatTransport { 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.callWithAuthRetry(chatId, state, send); + await this.sendWithEvents( + chatId, + "stop", + { partId, bodyBytes: byteLength(serializedBody) }, + () => this.callWithAuthRetry(chatId, state, send) + ); } catch { return false; } @@ -928,7 +1128,9 @@ export class TriggerChatTransport implements ChatTransport { await this.appendInputChunk(chatId, token, body, partId); }; - await this.callWithAuthRetry(chatId, state, send); + await this.sendWithEvents(chatId, "action", { partId, bodyBytes: byteLength(body) }, () => + this.callWithAuthRetry(chatId, state, send) + ); // Supersede any in-flight reader before subscribing — same as // `sendMessages`. Two concurrent readers both write `state.lastEventId` @@ -973,6 +1175,67 @@ export class TriggerChatTransport implements ChatTransport { 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, + extras: { messageId?: string; partId?: string; bodyBytes?: number }, + op: () => Promise + ): Promise { + 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(), + 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(), + 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 @@ -1229,6 +1492,7 @@ export class TriggerChatTransport implements ChatTransport { options?: { sendStopOnAbort?: boolean; peekSettled?: boolean; + resumed?: boolean; } ): ReadableStream { const internalAbort = new AbortController(); @@ -1403,6 +1667,16 @@ export class TriggerChatTransport implements ChatTransport { } } + this.emitEvent({ + type: "stream-connected", + chatId, + timestamp: Date.now(), + resumed: options?.resumed ?? false, + lastEventId: state.lastEventId, + messageId: this.lastTurnSends.get(chatId)?.messageId, + }); + let sawFirstChunk = false; + while (true) { let value: { id: string; @@ -1482,6 +1756,14 @@ export class TriggerChatTransport implements ChatTransport { if (refreshedToken) { state.publicAccessToken = refreshedToken; } + this.emitEvent({ + type: "turn-completed", + 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); this.coordinator?.release(chatId); @@ -1489,6 +1771,8 @@ export class TriggerChatTransport implements ChatTransport { lastEventId: state.lastEventId, }); + // Re-arm per-turn events for watch mode's next turn. + sawFirstChunk = false; if (this.watchMode) continue; internalAbort.abort(); @@ -1504,6 +1788,17 @@ export class TriggerChatTransport implements ChatTransport { // 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(), + chunkType: (value.chunk as { type?: string }).type, + lastEventId: value.id || undefined, + ...this.turnAttribution(chatId), + }); + } controller.enqueue(value.chunk as UIMessageChunk); } } catch (error) { @@ -1515,6 +1810,14 @@ export class TriggerChatTransport implements ChatTransport { } 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 { 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..ded1895dfd8 --- /dev/null +++ b/packages/trigger-sdk/test/chat-transport-events.test.ts @@ -0,0 +1,330 @@ +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): Promise { + 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 = {}) { + 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; + 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); + expect(sent.partId).toMatch(/[0-9a-f-]{36}/); + expect(sent.bodyBytes).toBeGreaterThan(0); + + const connected = events[1] as Extract; + expect(connected.resumed).toBe(false); + expect(connected.messageId).toBe("u-1"); + + const firstChunk = events[2] as Extract; + 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; + 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 () => { + 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; + 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("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 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; + 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 headStart response body fails mid-read", async () => { + const encoder = new TextEncoder(); + const failingBody = new ReadableStream({ + 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) => + 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); + }); +});