feat(sdk): AI SDK custom useChat transport & chat.task harness

.changeset/ai-sdk-chat-transport.md

@@ -0,0 +1,42 @@
+---
+"@trigger.dev/sdk": minor
+---
+
+Add AI SDK chat transport integration via two new subpath exports:
+
+**`@trigger.dev/sdk/chat`** (frontend, browser-safe):
+- `TriggerChatTransport` — custom `ChatTransport` for the AI SDK's `useChat` hook that runs chat completions as durable Trigger.dev tasks
+- `createChatTransport()` — factory function
+
+```tsx
+import { useChat } from "@ai-sdk/react";
+import { TriggerChatTransport } from "@trigger.dev/sdk/chat";
+
+const { messages, sendMessage } = useChat({
+  transport: new TriggerChatTransport({
+    task: "my-chat-task",
+    accessToken,
+  }),
+});
+```
+
+**`@trigger.dev/sdk/ai`** (backend, extends existing `ai.tool`/`ai.currentToolOptions`):
+- `chatTask()` — pre-typed task wrapper with auto-pipe support
+- `pipeChat()` — pipe a `StreamTextResult` or stream to the frontend
+- `CHAT_STREAM_KEY` — the default stream key constant
+- `ChatTaskPayload` type
+
+```ts
+import { chatTask } from "@trigger.dev/sdk/ai";
+import { streamText, convertToModelMessages } from "ai";
+
+export const myChatTask = chatTask({
+  id: "my-chat-task",
+  run: async ({ messages }) => {
+    return streamText({
+      model: openai("gpt-4o"),
+      messages: convertToModelMessages(messages),
+    });
+  },
+});
+```

.claude/rules/package-installation.md

@@ -0,0 +1,22 @@
+---
+paths:
+  - "**/package.json"
+---
+
+# Installing Packages
+
+When adding a new dependency to any package.json in the monorepo:
+
+1. **Look up the latest version** on npm before adding:
+   ```bash
+   pnpm view <package-name> version
+   ```
+   If unsure which version to use (e.g. major version compatibility), confirm with the user.
+
+2. **Edit the package.json directly** — do NOT use `pnpm add` as it can cause issues in the monorepo. Add the dependency with the correct version range (typically `^x.y.z`).
+
+3. **Run `pnpm i` from the repo root** after editing to install and update the lockfile:
+   ```bash
+   pnpm i
+   ```
+   Always run from the repo root, not from the package directory.

CLAUDE.md

@@ -6,6 +6,8 @@ This file provides guidance to Claude Code when working with this repository. Su
 
 This is a pnpm 10.23.0 monorepo using Turborepo. Run commands from root with `pnpm run`.
 
+**Adding dependencies:** Edit `package.json` directly instead of using `pnpm add`, then run `pnpm i` from the repo root. See `.claude/rules/package-installation.md` for the full process.
+
 ```bash
 pnpm run docker              # Start Docker services (PostgreSQL, Redis, Electric)
 pnpm run db:migrate           # Run database migrations

packages/core/src/v3/index.ts

@@ -80,6 +80,7 @@ export {
   getSchemaParseFn,
   type AnySchemaParseFn,
   type SchemaParseFn,
+  type inferSchemaOut,
   isSchemaZodEsque,
   isSchemaValibotEsque,
   isSchemaArkTypeEsque,

packages/core/src/v3/inputStreams/index.ts

@@ -51,6 +51,18 @@ export class InputStreamsAPI implements InputStreamManager {
     return this.#getManager().lastSeqNum(streamId);
   }
 
+  public setLastSeqNum(streamId: string, seqNum: number): void {
+    this.#getManager().setLastSeqNum(streamId, seqNum);
+  }
+
+  public shiftBuffer(streamId: string): boolean {
+    return this.#getManager().shiftBuffer(streamId);
+  }
+
+  public disconnectStream(streamId: string): void {
+    this.#getManager().disconnectStream(streamId);
+  }
+
   public clearHandlers(): void {
     this.#getManager().clearHandlers();
   }

packages/core/src/v3/inputStreams/manager.ts

@@ -40,6 +40,26 @@ export class StandardInputStreamManager implements InputStreamManager {
     return this.seqNums.get(streamId);
   }
 
+  setLastSeqNum(streamId: string, seqNum: number): void {
+    const current = this.seqNums.get(streamId);
+    // Only advance forward, never backward
+    if (current === undefined || seqNum > current) {
+      this.seqNums.set(streamId, seqNum);
+    }
+  }
+
+  shiftBuffer(streamId: string): boolean {
+    const buffered = this.buffer.get(streamId);
+    if (buffered && buffered.length > 0) {
+      buffered.shift();
+      if (buffered.length === 0) {
+        this.buffer.delete(streamId);
+      }
+      return true;
+    }
+    return false;
+  }
+
   setRunId(runId: string, streamsVersion?: string): void {
     this.currentRunId = runId;
     this.streamsVersion = streamsVersion;
@@ -158,6 +178,15 @@ export class StandardInputStreamManager implements InputStreamManager {
     }
   }
 
+  disconnectStream(streamId: string): void {
+    const tail = this.tails.get(streamId);
+    if (tail) {
+      tail.abortController.abort();
+      this.tails.delete(streamId);
+    }
+    this.buffer.delete(streamId);
+  }
+
   connectTail(runId: string, _fromSeq?: number): void {
     // No-op: tails are now created per-stream lazily
   }

packages/core/src/v3/inputStreams/noopManager.ts

@@ -22,6 +22,12 @@ export class NoopInputStreamManager implements InputStreamManager {
     return undefined;
   }
 
+  setLastSeqNum(_streamId: string, _seqNum: number): void {}
+
+  shiftBuffer(_streamId: string): boolean { return false; }
+
+  disconnectStream(_streamId: string): void {}
+
   clearHandlers(): void {}
   reset(): void {}
   disconnect(): void {}

packages/core/src/v3/inputStreams/types.ts

@@ -70,6 +70,28 @@ export interface InputStreamManager {
    */
   lastSeqNum(streamId: string): number | undefined;
 
+  /**
+   * Advance the last-seen S2 sequence number for the given input stream.
+   * Used after `.wait()` resumes to prevent the SSE tail from replaying
+   * the record that was consumed via the waitpoint path.
+   */
+  setLastSeqNum(streamId: string, seqNum: number): void;
+
+  /**
+   * Remove and discard the first buffered item for the given input stream.
+   * Used after `.wait()` resumes to remove the duplicate that the SSE tail
+   * buffered while the waitpoint was being completed via a separate path.
+   * Returns true if an item was removed, false if the buffer was empty.
+   */
+  shiftBuffer(streamId: string): boolean;
+
+  /**
+   * Disconnect the SSE tail and clear the buffer for a specific input stream.
+   * Used before suspending via `.wait()` so the tail doesn't buffer duplicates
+   * of data that will be delivered through the waitpoint path.
+   */
+  disconnectStream(streamId: string): void;
+
   /**
    * Clear all persistent `.on()` handlers and abort tails that have no remaining once waiters.
    * Called automatically when a task run completes.

packages/core/src/v3/realtimeStreams/manager.ts

@@ -6,6 +6,7 @@ import {
   RealtimeStreamInstance,
   RealtimeStreamOperationOptions,
   RealtimeStreamsManager,
+  StreamWriteResult,
 } from "./types.js";
 
 export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
@@ -16,7 +17,7 @@ export class StandardRealtimeStreamsManager implements RealtimeStreamsManager {
   ) {}
   // Track active streams - using a Set allows multiple streams for the same key to coexist
   private activeStreams = new Set<{
-    wait: () => Promise<void>;
+    wait: () => Promise<StreamWriteResult>;
     abortController: AbortController;
   }>();
 

packages/core/src/v3/realtimeStreams/noopManager.ts

@@ -15,7 +15,7 @@ export class NoopRealtimeStreamsManager implements RealtimeStreamsManager {
     options?: RealtimeStreamOperationOptions
   ): RealtimeStreamInstance<T> {
     return {
-      wait: () => Promise.resolve(),
+      wait: () => Promise.resolve({}),
       get stream(): AsyncIterableStream<T> {
         return createAsyncIterableStreamFromAsyncIterable(source);
       },

packages/core/src/v3/realtimeStreams/streamInstance.ts

@@ -3,7 +3,7 @@ import { AsyncIterableStream } from "../streams/asyncIterableStream.js";
 import { AnyZodFetchOptions } from "../zodfetch.js";
 import { StreamsWriterV1 } from "./streamsWriterV1.js";
 import { StreamsWriterV2 } from "./streamsWriterV2.js";
-import { StreamsWriter } from "./types.js";
+import { StreamsWriter, StreamWriteResult } from "./types.js";
 
 export type StreamInstanceOptions<T> = {
   apiClient: ApiClient;
@@ -63,8 +63,9 @@ export class StreamInstance<T> implements StreamsWriter {
     return streamWriter;
   }
 
-  public async wait(): Promise<void> {
-    return this.streamPromise.then((writer) => writer.wait());
+  public async wait(): Promise<StreamWriteResult> {
+    const writer = await this.streamPromise;
+    return writer.wait();
   }
 
   public get stream(): AsyncIterableStream<T> {

packages/core/src/v3/realtimeStreams/streamsWriterV1.ts

@@ -2,7 +2,7 @@ import { request as httpsRequest } from "node:https";
 import { request as httpRequest } from "node:http";
 import { URL } from "node:url";
 import { randomBytes } from "node:crypto";
-import { StreamsWriter } from "./types.js";
+import { StreamsWriter, StreamWriteResult } from "./types.js";
 
 export type StreamsWriterV1Options<T> = {
   baseUrl: string;
@@ -258,8 +258,9 @@ export class StreamsWriterV1<T> implements StreamsWriter {
     await this.makeRequest(0);
   }
 
-  public async wait(): Promise<void> {
-    return this.streamPromise;
+  public async wait(): Promise<StreamWriteResult> {
+    await this.streamPromise;
+    return {};
   }
 
   public [Symbol.asyncIterator]() {

packages/core/src/v3/realtimeStreams/streamsWriterV2.ts

@@ -1,5 +1,5 @@
 import { S2, AppendRecord, BatchTransform } from "@s2-dev/streamstore";
-import { StreamsWriter } from "./types.js";
+import { StreamsWriter, StreamWriteResult } from "./types.js";
 import { nanoid } from "nanoid";
 
 export type StreamsWriterV2Options<T = any> = {
@@ -54,6 +54,7 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
   private readonly maxInflightBytes: number;
   private aborted = false;
   private sessionWritable: WritableStream<any> | null = null;
+  private lastSeqNum: number | undefined;
 
   constructor(private options: StreamsWriterV2Options<T>) {
     this.debug = options.debug ?? false;
@@ -169,9 +170,9 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
       const lastAcked = session.lastAckedPosition();
 
       if (lastAcked?.end) {
-        const recordsWritten = lastAcked.end.seqNum;
+        this.lastSeqNum = lastAcked.end.seqNum;
         this.log(
-          `[S2MetadataStream] Written ${recordsWritten} records, ending at seqNum=${lastAcked.end.seqNum}`
+          `[S2MetadataStream] Written ${this.lastSeqNum} records, ending at seqNum=${this.lastSeqNum}`
         );
       }
     } catch (error) {
@@ -184,8 +185,9 @@ export class StreamsWriterV2<T = any> implements StreamsWriter {
     }
   }
 
-  public async wait(): Promise<void> {
+  public async wait(): Promise<StreamWriteResult> {
     await this.streamPromise;
+    return { lastEventId: this.lastSeqNum?.toString() };
   }
 
   public [Symbol.asyncIterator]() {

packages/core/src/v3/realtimeStreams/types.ts

@@ -26,13 +26,17 @@ export interface RealtimeStreamsManager {
   ): Promise<void>;
 }
 
+export type StreamWriteResult = {
+  lastEventId?: string;
+};
+
 export interface RealtimeStreamInstance<T> {
-  wait(): Promise<void>;
+  wait(): Promise<StreamWriteResult>;
   get stream(): AsyncIterableStream<T>;
 }
 
 export interface StreamsWriter {
-  wait(): Promise<void>;
+  wait(): Promise<StreamWriteResult>;
 }
 
 export type RealtimeDefinedStream<TPart> = {
@@ -71,6 +75,10 @@ export type PipeStreamOptions = {
    * Additional request options for the API call.
    */
   requestOptions?: ApiRequestOptions;
+  /** Override the default span name for this operation. */
+  spanName?: string;
+  /** When true, the span will be collapsed in the dashboard. */
+  collapsed?: boolean;
 };
 
 /**
@@ -89,7 +97,7 @@ export type PipeStreamResult<T> = {
    * to the realtime stream. Use this to wait for the stream to complete before
    * finishing your task.
    */
-  waitUntilComplete: () => Promise<void>;
+  waitUntilComplete: () => Promise<StreamWriteResult>;
 };
 
 /**
@@ -185,6 +193,14 @@ export type RealtimeDefinedInputStream<TData> = {
    * Uses a waitpoint token internally. Can only be called inside a task.run().
    */
   wait: (options?: InputStreamWaitOptions) => ManualWaitpointPromise<TData>;
+  /**
+   * Wait for data with a warm phase before suspending.
+   *
+   * Keeps the task warm (active, using compute) for `warmTimeoutInSeconds`,
+   * then suspends via `.wait()` if no data arrives. If data arrives during
+   * the warm phase the task responds instantly without suspending.
+   */
+  waitWithWarmup: (options: InputStreamWaitWithWarmupOptions) => Promise<{ ok: true; output: TData } | { ok: false; error?: any }>;
   /**
    * Send data to this input stream on a specific run.
    * This is used from outside the task (e.g., from your backend or another task).
@@ -199,6 +215,8 @@ export type InputStreamSubscription = {
 export type InputStreamOnceOptions = {
   signal?: AbortSignal;
   timeoutMs?: number;
+  /** Override the default span name for this operation. */
+  spanName?: string;
 };
 
 export type SendInputStreamOptions = {
@@ -234,6 +252,18 @@ export type InputStreamWaitOptions = {
    * and filtering waitpoints via `wait.listTokens()`.
    */
   tags?: string[];
+
+  /** Override the default span name for this operation. */
+  spanName?: string;
+};
+
+export type InputStreamWaitWithWarmupOptions = {
+  /** Seconds to keep the task warm before suspending. */
+  warmTimeoutInSeconds: number;
+  /** Maximum time to wait after suspending (duration string, e.g. "1h"). */
+  timeout?: string;
+  /** Override the default span name for the outer operation. */
+  spanName?: string;
 };
 
 export type InferInputStreamType<T> = T extends RealtimeDefinedInputStream<infer TData>