add stopping support and fix issue with the OpenAI responses API and stopped streams

Author: ericallam

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

@@ -216,6 +216,28 @@ export type ChatTaskRunPayload = ChatTaskPayload & ChatTaskSignals;
 const messagesInput = streams.input<ChatTaskPayload>({ id: CHAT_MESSAGES_STREAM_ID });
 const stopInput = streams.input<{ stop: true; message?: string }>({ id: CHAT_STOP_STREAM_ID });
 
+/**
+ * Strips provider-specific IDs from message parts so that partial/stopped
+ * assistant responses don't cause 404s when sent back to the provider
+ * (e.g. OpenAI Responses API message IDs).
+ * @internal
+ */
+function sanitizeMessages<TMessage extends UIMessage>(messages: TMessage[]): TMessage[] {
+  return messages.map((msg) => {
+    if (msg.role !== "assistant" || !msg.parts) return msg;
+    return {
+      ...msg,
+      parts: msg.parts.map((part: any) => {
+        // Strip provider-specific metadata (e.g. OpenAI Responses API itemId)
+        // and streaming state from assistant message parts. These cause 404s
+        // when partial/stopped responses are sent back to the provider.
+        const { providerMetadata, state, id, ...rest } = part;
+        return rest;
+      }),
+    };
+  });
+}
+
 /**
  * Tracks how many times `pipeChat` has been called in the current `chatTask` run.
  * Used to prevent double-piping when a user both calls `pipeChat()` manually
@@ -454,6 +476,7 @@ function chatTask<TIdentifier extends string>(
           try {
             const result = await userRun({
               ...currentPayload,
+              messages: sanitizeMessages(currentPayload.messages),
               signal: combinedSignal,
               cancelSignal,
               stopSignal,

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

@@ -30,6 +30,8 @@ const DEFAULT_STREAM_KEY = "chat";
 const DEFAULT_BASE_URL = "https://api.trigger.dev";
 const DEFAULT_STREAM_TIMEOUT_SECONDS = 120;
 
+
+
 /**
  * Options for creating a TriggerChatTransport.
  */
@@ -91,6 +93,8 @@ type ChatSessionState = {
   publicAccessToken: string;
   /** Last SSE event ID — used to resume the stream without replaying old events. */
   lastEventId?: string;
+  /** Set when the stream was aborted mid-turn (stop). On reconnect, skip chunks until __trigger_turn_complete. */
+  skipToTurnComplete?: boolean;
 };
 
 /**
@@ -164,14 +168,12 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
     };
 
     const session = this.sessions.get(chatId);
-
     // If we have an existing run, send the message via input stream
     // to resume the conversation in the same run.
     if (session?.runId) {
       try {
         const apiClient = new ApiClient(this.baseURL, session.publicAccessToken);
         await apiClient.sendInputStream(session.runId, CHAT_MESSAGES_STREAM_ID, payload);
-
         return this.subscribeToStream(
           session.runId,
           session.publicAccessToken,
@@ -205,7 +207,6 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       runId,
       publicAccessToken: publicAccessToken ?? currentToken,
     });
-
     return this.subscribeToStream(
       runId,
       publicAccessToken ?? currentToken,
@@ -256,7 +257,8 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       abortSignal.addEventListener(
         "abort",
         () => {
-          if (session?.runId) {
+          if (session) {
+            session.skipToTurnComplete = true;
             const api = new ApiClient(this.baseURL, session.publicAccessToken);
             api
               .sendInputStream(session.runId, CHAT_STOP_STREAM_ID, { stop: true })
@@ -283,16 +285,18 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
         try {
           const sseStream = await subscription.subscribe();
           const reader = sseStream.getReader();
+          let chunkCount = 0;
 
           try {
             while (true) {
               const { done, value } = await reader.read();
 
               if (done) {
-                // Stream closed without a control chunk — the run has
-                // ended (or was killed). Clear the session so that the
-                // next message triggers a fresh run.
-                if (chatId) {
+                // Only delete session if the stream ended naturally (not aborted by stop).
+                // When the user clicks stop, the abort closes the SSE reader which
+                // returns done=true, but the run is still alive and waiting for
+                // the next message via input streams.
+                if (chatId && !combinedSignal.aborted) {
                   this.sessions.delete(chatId);
                 }
                 controller.close();
@@ -315,11 +319,17 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
               if (value.chunk != null && typeof value.chunk === "object") {
                 const chunk = value.chunk as Record<string, unknown>;
 
-                // Intercept the turn-complete control chunk emitted by
-                // `chatTask` after the AI response stream completes. This
-                // chunk is never forwarded to the AI SDK consumer.
+                // After a stop, skip leftover chunks from the stopped turn
+                // until we see the __trigger_turn_complete marker.
+                if (session?.skipToTurnComplete) {
+                  if (chunk.type === "__trigger_turn_complete") {
+                    session.skipToTurnComplete = false;
+                    chunkCount = 0;
+                  }
+                  continue;
+                }
+
                 if (chunk.type === "__trigger_turn_complete" && chatId) {
-                  // Abort the underlying fetch to close the SSE connection
                   internalAbort.abort();
                   try {
                     controller.close();
@@ -329,6 +339,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
                   return;
                 }
 
+                chunkCount++;
                 controller.enqueue(chunk as unknown as UIMessageChunk);
               }
             }

references/ai-chat/src/components/chat.tsx

@@ -79,7 +79,7 @@ export function Chat() {
     baseURL: process.env.NEXT_PUBLIC_TRIGGER_API_URL,
   });
 
-  const { messages, sendMessage, status, error } = useChat({
+  const { messages, sendMessage, stop, status, error } = useChat({
     transport,
   });
 
@@ -152,13 +152,23 @@ export function Chat() {
           placeholder="Type a message…"
           className="flex-1 rounded-lg border border-gray-300 px-3 py-2 text-sm outline-none focus:border-blue-500 focus:ring-1 focus:ring-blue-500"
         />
-        <button
-          type="submit"
-          disabled={!input.trim() || status === "streaming"}
-          className="rounded-lg bg-blue-600 px-4 py-2 text-sm font-medium text-white hover:bg-blue-700 disabled:opacity-50"
-        >
-          Send
-        </button>
+        {status === "streaming" ? (
+          <button
+            type="button"
+            onClick={stop}
+            className="rounded-lg bg-red-600 px-4 py-2 text-sm font-medium text-white hover:bg-red-700"
+          >
+            Stop
+          </button>
+        ) : (
+          <button
+            type="submit"
+            disabled={!input.trim()}
+            className="rounded-lg bg-blue-600 px-4 py-2 text-sm font-medium text-white hover:bg-blue-700 disabled:opacity-50"
+          >
+            Send
+          </button>
+        )}
       </form>
     </div>
   );

references/ai-chat/src/trigger/chat.ts

@@ -1,5 +1,5 @@
 import { chat } from "@trigger.dev/sdk/ai";
-import { streamText, convertToModelMessages, tool } from "ai";
+import { streamText, convertToModelMessages, tool, stepCountIs } from "ai";
 import { openai } from "@ai-sdk/openai";
 import { z } from "zod";
 import os from "node:os";
@@ -64,14 +64,14 @@ declare const Deno: unknown;
 
 export const aiChat = chat.task({
   id: "ai-chat",
-  run: async ({ messages, signal }) => {
+  run: async ({ messages, stopSignal }) => {
     return streamText({
       model: openai("gpt-4o-mini"),
       system: "You are a helpful assistant. Be concise and friendly.",
       messages: await convertToModelMessages(messages),
       tools: { inspectEnvironment },
-      maxSteps: 3,
-      abortSignal: signal,
+      stopWhen: stepCountIs(10),
+      abortSignal: stopSignal,
     });
   },
 });