feat(sdk): return lastEventId from writeTurnComplete and typed capture result

chat.writeTurnComplete() now resolves to { lastEventId }, the resume cursor
for the next turn, so a custom-agent loop can persist it straight from the
task instead of round-tripping it back from the client.

chat.pipeAndCapture() no longer throws when a stream is stopped or fails. It
resolves to a PipeAndCaptureResult carrying any partial message captured
before the stop or failure, a typed status (complete | aborted | error), and
the error on failure. The internal turn.complete() path keeps its existing
contract (returns UIMessage | undefined, still throws on a genuine stream
failure and discards output on full cancel).

Author: matt-aitken

.changeset/chat-custom-agent-capture.md

@@ -0,0 +1,22 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Custom chat agent loops get two ergonomic wins for owning the turn loop.
+
+`chat.writeTurnComplete()` now returns the `lastEventId` of the turn-complete record, so you can persist the next turn's resume cursor straight from the task instead of round-tripping it back from the client.
+
+```ts
+const { lastEventId } = await chat.writeTurnComplete();
+await db.chats.update(chatId, { lastEventId });
+```
+
+`chat.pipeAndCapture()` no longer throws when a stream is stopped or fails. It now returns a `PipeAndCaptureResult` whose `message` holds any partial output captured before the stop or failure, alongside a typed `status` (`"complete" | "aborted" | "error"`) and, on failure, the `error`. Read the message off the result:
+
+```ts
+const { message, status, error } = await chat.pipeAndCapture(result, { signal });
+if (message) conversation.addResponse(message);
+if (status === "error") logger.error("turn failed", { error });
+```
+
+Note: `pipeAndCapture` previously resolved to `UIMessage | undefined`. Update call sites to read `.message` from the returned result.

docs/ai-chat/changelog.mdx

@@ -55,10 +55,10 @@ if (isFinal) {
   const result = streamText({ model, messages: conversation.modelMessages, tools });
   // Pass originalMessages so the handed-over tool round merges into the
   // step-1 assistant instead of starting a new message.
-  const response = await chat.pipeAndCapture(result, {
+  const { message } = await chat.pipeAndCapture(result, {
     originalMessages: conversation.uiMessages,
   });
-  if (response) await conversation.addResponse(response);
+  if (message) await conversation.addResponse(message);
 }
 ```
 

docs/ai-chat/compaction.mdx

@@ -365,8 +365,8 @@ for (let turn = 0; turn < 100; turn++) {
     stopWhen: stepCountIs(15),
   });
 
-  const response = await chat.pipeAndCapture(result);
-  if (response) await conversation.addResponse(response);
+  const { message } = await chat.pipeAndCapture(result);
+  if (message) await conversation.addResponse(message);
 
   // Outer-loop compaction
   const usage = await result.totalUsage;

docs/ai-chat/custom-agents.mdx

@@ -160,11 +160,11 @@ for await (const turn of session) {
   });
 
   // Manual: pipe and capture separately
-  const response = await chat.pipeAndCapture(result, { signal: turn.signal });
+  const { message } = await chat.pipeAndCapture(result, { signal: turn.signal });
 
-  if (response) {
+  if (message) {
     // Custom processing before accumulating
-    await turn.addResponse(response);
+    await turn.addResponse(message);
   }
 
   // Custom persistence, analytics, etc.
@@ -215,8 +215,8 @@ For full control, skip `createSession` and compose the primitives directly:
 | ------------------------------- | -------------------------------------------------------------------------------------------- |
 | `chat.messages`                 | Input stream for incoming messages — use `.waitWithIdleTimeout()` to wait for the next turn  |
 | `chat.createStopSignal()`       | Create a managed stop signal wired to the stop input stream                                  |
-| `chat.pipeAndCapture(result)`   | Pipe a `StreamTextResult` to the chat stream and capture the response                        |
-| `chat.writeTurnComplete()`      | Signal the frontend that the current turn is complete                                        |
+| `chat.pipeAndCapture(result)`   | Pipe a stream and capture the response; returns `{ message, status, error }`                 |
+| `chat.writeTurnComplete()`      | Signal the frontend that the current turn is complete; returns `{ lastEventId }`             |
 | `chat.MessageAccumulator`       | Accumulates conversation messages across turns                                               |
 | `chat.pipe(stream)`             | Pipe a stream to the frontend (no response capture)                                          |
 | `chat.cleanupAbortedParts(msg)` | Clean up incomplete parts from a stopped response                                            |
@@ -285,21 +285,16 @@ export const myChat = chat.customAgent({
         stopWhen: stepCountIs(15),
       });
 
-      let response;
-      try {
-        response = await chat.pipeAndCapture(result, { signal: combinedSignal });
-      } catch (error) {
-        if (error instanceof Error && error.name === "AbortError") {
-          if (runSignal.aborted) break;
-          // Stop — fall through to accumulate partial
-        } else {
-          throw error;
-        }
-      }
+      const { message, status, error } = await chat.pipeAndCapture(result, {
+        signal: combinedSignal,
+      });
+      // pipeAndCapture never throws: a user stop returns status "aborted" with
+      // the partial message, and a failure returns status "error" with `error`.
+      if (status === "error") throw error;
 
-      if (response) {
+      if (message) {
         const cleaned =
-          stop.signal.aborted && !runSignal.aborted ? chat.cleanupAbortedParts(response) : response;
+          stop.signal.aborted && !runSignal.aborted ? chat.cleanupAbortedParts(message) : message;
         await conversation.addResponse(cleaned);
       }
 
@@ -346,8 +341,8 @@ const messages = await conversation.addIncoming(
 );
 
 // After piping, add the response
-const response = await chat.pipeAndCapture(result);
-if (response) await conversation.addResponse(response);
+const { message } = await chat.pipeAndCapture(result);
+if (message) await conversation.addResponse(message);
 
 // Access accumulated messages for persistence
 conversation.uiMessages; // UIMessage[]

docs/ai-chat/fast-starts.mdx

@@ -589,10 +589,10 @@ if (turn === 0 && payload.trigger === "handover-prepare") {
       messages: conversation.modelMessages,
       stopWhen: stepCountIs(10),
     });
-    const response = await chat.pipeAndCapture(result, {
+    const { message } = await chat.pipeAndCapture(result, {
       originalMessages: conversation.uiMessages,
     });
-    if (response) await conversation.addResponse(response);
+    if (message) await conversation.addResponse(message);
   }
   await chat.writeTurnComplete(); // on isFinal the warm partial is already the response
   return;

docs/ai-chat/pending-messages.mdx

@@ -179,10 +179,10 @@ for (let turn = 0; turn < 100; turn++) {
     stopWhen: stepCountIs(15),
   });
 
-  const response = await chat.pipeAndCapture(result);
+  const { message } = await chat.pipeAndCapture(result);
   sub.off();
 
-  if (response) await conversation.addResponse(response);
+  if (message) await conversation.addResponse(message);
   await chat.writeTurnComplete();
 }
 ```

docs/ai-chat/reference.mdx

@@ -503,8 +503,8 @@ All methods available on the `chat` object from `@trigger.dev/sdk/ai`.
 | `chat.agent(options)`                       | Create a chat agent                                                                                                          |
 | `chat.createSession(payload, options)`      | Create an async iterator for chat turns                                                                                      |
 | `chat.pipe(source, options?)`               | Pipe a stream to the frontend (from anywhere inside a task)                                                                  |
-| `chat.pipeAndCapture(source, options?)`     | Pipe and capture the response `UIMessage`                                                                                    |
-| `chat.writeTurnComplete(options?)`          | Signal the frontend that the current turn is complete                                                                        |
+| `chat.pipeAndCapture(source, options?)`     | Pipe and capture the response; returns `{ message, status, error }`                                                         |
+| `chat.writeTurnComplete(options?)`          | Signal the frontend that the current turn is complete; returns `{ lastEventId }`                                            |
 | `chat.createStopSignal()`                   | Create a managed stop signal wired to the stop input stream                                                                  |
 | `chat.messages`                             | Input stream for incoming messages — use `.waitWithIdleTimeout()`                                                            |
 | `chat.local<T>({ id })`                     | Create a per-run typed local (see [`chat.local`](/ai-chat/chat-local))                             |

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

@@ -8828,35 +8828,72 @@ function createStopSignal(): {
  * The `TriggerChatTransport` intercepts this to close the ReadableStream
  * for the current turn. Call after piping the response stream.
  *
+ * Returns the `lastEventId` of the turn-complete control record on
+ * `session.out` — the resume cursor for the start of the next turn. Save it
+ * from the task (e.g. to your DB) instead of round-tripping it back from the
+ * client after the turn ends. `undefined` if the write returned no ack.
+ *
  * @example
  * ```ts
  * await chat.pipe(result);
- * await chat.writeTurnComplete();
+ * const { lastEventId } = await chat.writeTurnComplete();
+ * await db.chats.update(chatId, { lastEventId });
  * ```
  */
-async function chatWriteTurnComplete(options?: { publicAccessToken?: string }): Promise<void> {
-  await writeTurnCompleteChunk(undefined, options?.publicAccessToken);
+async function chatWriteTurnComplete(options?: {
+  publicAccessToken?: string;
+}): Promise<{ lastEventId?: string }> {
+  const result = await writeTurnCompleteChunk(undefined, options?.publicAccessToken);
+  return { lastEventId: result?.lastEventId };
 }
 
+/**
+ * The outcome of a turn's stream, reported by {@link pipeChatAndCapture}.
+ *
+ * - `complete` — the stream finished on its own.
+ * - `aborted` — the stream was stopped via the `signal` (user stop / cancel).
+ * - `error` — the stream threw; `error` carries what was thrown.
+ *
+ * `message` holds whatever the assistant produced and is present for every
+ * status — including `aborted` and `error` — as long as any output streamed
+ * before the stop or failure, so partial responses are never lost.
+ */
+export type PipeAndCaptureResult = {
+  /** The captured assistant message, or `undefined` if nothing streamed. */
+  message: UIMessage | undefined;
+  /** Coarse outcome of the stream. */
+  status: "complete" | "aborted" | "error";
+  /** The AI SDK finish reason, when the stream reported one. */
+  finishReason?: FinishReason;
+  /** What the stream threw, present only when `status === "error"`. */
+  error?: unknown;
+};
+
 /**
  * Pipe a `StreamTextResult` (or similar) to the chat stream and capture
  * the assistant's response message via `onFinish`.
  *
  * Combines `toUIMessageStream()` + `onFinish` callback + `chat.pipe()`.
- * Returns the captured `UIMessage`, or `undefined` if capture failed.
+ * Never throws on a stopped or failed stream: it returns a
+ * {@link PipeAndCaptureResult} whose `message` holds any partial output
+ * captured before the stop/failure, alongside a typed `status` (and `error`
+ * on failure). Save the partial after a stop or error without separate
+ * capture logic.
  *
  * @example
  * ```ts
  * const result = streamText({ model, messages, abortSignal: signal });
- * const response = await chat.pipeAndCapture(result, { signal });
- * if (response) conversation.addResponse(response);
+ * const { message, status, error } = await chat.pipeAndCapture(result, { signal });
+ * if (message) conversation.addResponse(message);
+ * if (status === "error") logger.error("turn failed", { error });
  * ```
  */
 async function pipeChatAndCapture(
   source: UIMessageStreamable,
   options?: { signal?: AbortSignal; spanName?: string; originalMessages?: UIMessage[] }
-): Promise<UIMessage | undefined> {
+): Promise<PipeAndCaptureResult> {
   let captured: UIMessage | undefined;
+  let capturedFinishReason: FinishReason | undefined;
   let resolveOnFinish: () => void;
   const onFinishPromise = new Promise<void>((r) => {
     resolveOnFinish = r;
@@ -8875,19 +8912,51 @@ async function pipeChatAndCapture(
     // the frontend replaces the partial message — wiping the
     // pre-injection text from the UI and the captured response.
     generateMessageId: resolvedOptions.generateMessageId ?? generateMessageId,
-    onFinish: ({ responseMessage }: { responseMessage: UIMessage }) => {
+    onFinish: ({
+      responseMessage,
+      finishReason,
+    }: {
+      responseMessage: UIMessage;
+      finishReason?: FinishReason;
+    }) => {
       captured = responseMessage;
+      capturedFinishReason = finishReason;
       resolveOnFinish!();
     },
   });
 
-  await pipeChat(uiStream, {
-    signal: options?.signal,
-    spanName: options?.spanName ?? "stream response",
-  });
-  await onFinishPromise;
+  let status: PipeAndCaptureResult["status"] = "complete";
+  let error: unknown;
+  try {
+    await pipeChat(uiStream, {
+      signal: options?.signal,
+      spanName: options?.spanName ?? "stream response",
+    });
+    // The pipe can drain cleanly on a stop — the source stream just ends
+    // early — so classify by the signal rather than relying on a throw.
+    if (options?.signal?.aborted) {
+      status = "aborted";
+    }
+  } catch (err) {
+    if ((err instanceof Error && err.name === "AbortError") || options?.signal?.aborted) {
+      status = "aborted";
+    } else {
+      status = "error";
+      error = err;
+    }
+  }
 
-  return captured;
+  // `onFinish` fires even on abort, carrying the partial — but a hard stop can
+  // prevent it from firing at all, so race it against a timeout to avoid
+  // hanging the caller. Mirrors chat.agent's capture path.
+  await Promise.race([onFinishPromise, new Promise<void>((r) => setTimeout(r, 2_000))]);
+
+  return {
+    message: captured,
+    status,
+    finishReason: capturedFinishReason,
+    ...(error !== undefined ? { error } : {}),
+  };
 }
 
 /**
@@ -8903,8 +8972,8 @@ async function pipeChatAndCapture(
  * for (let turn = 0; turn < 100; turn++) {
  *   const messages = await conversation.addIncoming(payload.messages, payload.trigger, turn);
  *   const result = streamText({ model, messages });
- *   const response = await chat.pipeAndCapture(result);
- *   if (response) await conversation.addResponse(response);
+ *   const { message } = await chat.pipeAndCapture(result);
+ *   if (message) await conversation.addResponse(message);
  * }
  * ```
  */
@@ -9570,7 +9639,7 @@ function createChatSession(
               }
               let response: UIMessage | undefined;
               try {
-                response = await pipeChatAndCapture(source, {
+                const captured = await pipeChatAndCapture(source, {
                   signal: combinedSignal,
                   // On a non-final handover turn, thread the spliced partial so a
                   // resumed tool round's tool-output chunks merge into the
@@ -9579,18 +9648,18 @@ function createChatSession(
                   // fresh response into the prior assistant message).
                   ...(handoverThisTurn ? { originalMessages: accumulator.uiMessages } : {}),
                 });
-              } catch (error) {
-                if (error instanceof Error && error.name === "AbortError") {
-                  if (runSignal.aborted) {
-                    // Full cancel — don't accumulate
-                    sessionMsgSub.off();
-                    await chatWriteTurnComplete();
-                    return undefined;
-                  }
-                  // Stop — fall through to accumulate partial response
-                } else {
-                  throw error;
+                if (runSignal.aborted) {
+                  // Full cancel — don't accumulate
+                  sessionMsgSub.off();
+                  await chatWriteTurnComplete();
+                  return undefined;
+                }
+                // Surface a genuine stream failure to the caller. A user stop
+                // (status "aborted") falls through so the partial is accumulated.
+                if (captured.status === "error") {
+                  throw captured.error;
                 }
+                response = captured.message;
               } finally {
                 // Detach at stream end (like the agent loop): the steering queue
                 // can't inject anymore, so later arrivals must buffer for the next turn.