feat(chat): expose typed chat.stream, add deepResearch subtask example, per-chat model persistence, debug panel

- Export chat.stream (typed RealtimeDefinedStream<UIMessageChunk>) for writing custom data to the chat stream
- Add deepResearch subtask using data-* chunks to stream progress back to parent chat via target: root
- Use AI SDK data-research-progress chunk protocol with id-based updates for live progress
- Add ResearchProgress component and generic data-* fallback renderer in frontend
- Persist model per chat in DB (schema + onChatStart), model selector only on new chats
- Add collapsible debug panel showing run ID (with dashboard link), chat ID, model, status, session info
- Document chat.stream API, data-* chunks, and subtask streaming pattern in docs

Author: ericallam

docs/guides/ai-chat.mdx

@@ -750,6 +750,140 @@ This removes tool invocation parts stuck in `partial-call` state and marks any `
   Stop signal delivery is best-effort. There is a small race window where the model may finish before the stop signal arrives, in which case the turn completes normally with `stopped: false`. This is expected and does not require special handling.
 </Note>
 
+## Writing to the chat stream
+
+### Custom chunks with `chat.stream`
+
+`chat.stream` is a typed stream bound to the chat output. Use it to write custom `UIMessageChunk` data alongside the AI-generated response — for example, status updates or progress indicators.
+
+```ts
+import { chat } from "@trigger.dev/sdk/ai";
+
+export const myChat = chat.task({
+  id: "my-chat",
+  run: async ({ messages, signal }) => {
+    // Write a custom data part to the chat stream.
+    // The AI SDK's data-* chunk protocol adds this to message.parts
+    // on the frontend, where you can render it however you like.
+    const { waitUntilComplete } = chat.stream.writer({
+      execute: ({ write }) => {
+        write({
+          type: "data-status",
+          id: "search-progress",
+          data: { message: "Searching the web...", progress: 0.5 },
+        });
+      },
+    });
+    await waitUntilComplete();
+
+    // Then stream the AI response
+    return streamText({ model: openai("gpt-4o"), messages, abortSignal: signal });
+  },
+});
+```
+
+<Tip>
+  Use `data-*` chunk types (e.g. `data-status`, `data-progress`) for custom data. The AI SDK processes these into `DataUIPart` objects in `message.parts` on the frontend. Writing the same `type` + `id` again updates the existing part instead of creating a new one — useful for live progress.
+</Tip>
+
+`chat.stream` exposes the full stream API:
+
+| Method | Description |
+|--------|-------------|
+| `chat.stream.writer(options)` | Write individual chunks via a callback |
+| `chat.stream.pipe(stream, options?)` | Pipe a `ReadableStream` or `AsyncIterable` |
+| `chat.stream.append(value, options?)` | Append raw data |
+| `chat.stream.read(runId, options?)` | Read the stream by run ID |
+
+### Streaming from subtasks
+
+When a tool invokes a subtask via `triggerAndWait`, the subtask can stream directly to the parent chat using `target: "root"`:
+
+```ts
+import { chat, ai } from "@trigger.dev/sdk/ai";
+import { schemaTask } from "@trigger.dev/sdk";
+import { streamText, generateId } from "ai";
+import { z } from "zod";
+
+// A subtask that streams progress back to the parent chat
+export const researchTask = schemaTask({
+  id: "research",
+  schema: z.object({ query: z.string() }),
+  run: async ({ query }) => {
+    const partId = generateId();
+
+    // Write a data-* chunk to the root run's chat stream.
+    // The frontend receives this as a DataUIPart in message.parts.
+    const { waitUntilComplete } = chat.stream.writer({
+      target: "root",
+      execute: ({ write }) => {
+        write({
+          type: "data-research-status",
+          id: partId,
+          data: { query, status: "in-progress" },
+        });
+      },
+    });
+    await waitUntilComplete();
+
+    // Do the work...
+    const result = await doResearch(query);
+
+    // Update the same part with the final status
+    const { waitUntilComplete: waitDone } = chat.stream.writer({
+      target: "root",
+      execute: ({ write }) => {
+        write({
+          type: "data-research-status",
+          id: partId,
+          data: { query, status: "done", resultCount: result.length },
+        });
+      },
+    });
+    await waitDone();
+
+    return result;
+  },
+});
+
+// The chat task uses it as a tool via ai.tool()
+export const myChat = chat.task({
+  id: "my-chat",
+  run: async ({ messages, signal }) => {
+    return streamText({
+      model: openai("gpt-4o"),
+      messages,
+      abortSignal: signal,
+      tools: {
+        research: ai.tool(researchTask),
+      },
+    });
+  },
+});
+```
+
+On the frontend, render the custom data part:
+
+```tsx
+{message.parts.map((part, i) => {
+  if (part.type === "data-research-status") {
+    const { query, status, resultCount } = part.data;
+    return (
+      <div key={i}>
+        {status === "done" ? `Found ${resultCount} results` : `Researching "${query}"...`}
+      </div>
+    );
+  }
+  // ...other part types
+})}
+```
+
+The `target` option accepts:
+- `"self"` — current run (default)
+- `"parent"` — parent task's run
+- `"root"` — root task's run (the chat task)
+- A specific run ID string
+
 ## Client data and metadata
 
 ### Transport-level client data
@@ -1075,6 +1209,7 @@ See [onTurnComplete](#onturncomplete) for the full field reference.
 | `chat.setWarmTimeoutInSeconds(seconds)` | Override warm timeout at runtime |
 | `chat.isStopped()` | Check if the current turn was stopped by the user (works anywhere during a turn) |
 | `chat.cleanupAbortedParts(message)` | Remove incomplete parts from a stopped response message |
+| `chat.stream` | Typed chat output stream — use `.writer()`, `.pipe()`, `.append()`, `.read()` |
 
 ## Self-hosting
 

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

@@ -14,7 +14,7 @@ import {
   type TaskSchema,
   type TaskWithSchema,
 } from "@trigger.dev/core/v3";
-import type { ModelMessage, UIMessage } from "ai";
+import type { ModelMessage, UIMessage, UIMessageChunk } from "ai";
 import type { StreamWriteResult } from "@trigger.dev/core/v3";
 import { convertToModelMessages, dynamicTool, generateId as generateMessageId, jsonSchema, JSONSchema7, Schema, Tool, ToolCallOptions, zodSchema } from "ai";
 import { type Attributes, trace } from "@opentelemetry/api";
@@ -175,6 +175,29 @@ export const CHAT_STREAM_KEY = _CHAT_STREAM_KEY;
 // Re-export input stream IDs for advanced usage
 export { CHAT_MESSAGES_STREAM_ID, CHAT_STOP_STREAM_ID };
 
+/**
+ * Typed chat output stream. Provides `.writer()`, `.pipe()`, `.append()`,
+ * and `.read()` methods pre-bound to the chat stream key and typed to `UIMessageChunk`.
+ *
+ * Use from within a `chat.task` run to write custom chunks:
+ * ```ts
+ * const { waitUntilComplete } = chat.stream.writer({
+ *   execute: ({ write }) => {
+ *     write({ type: "text-start", id: "status-1" });
+ *     write({ type: "text-delta", id: "status-1", delta: "Processing..." });
+ *     write({ type: "text-end", id: "status-1" });
+ *   },
+ * });
+ * await waitUntilComplete();
+ * ```
+ *
+ * Use from a subtask to stream back to the parent chat:
+ * ```ts
+ * chat.stream.pipe(myStream, { target: "root" });
+ * ```
+ */
+const chatStream = streams.define<UIMessageChunk>({ id: _CHAT_STREAM_KEY });
+
 /**
  * The wire payload shape sent by `TriggerChatTransport`.
  * Uses `metadata` to match the AI SDK's `ChatRequestOptions` field name.
@@ -1452,6 +1475,8 @@ export const chat = {
   isStopped,
   /** Clean up aborted parts from a UIMessage. See {@link cleanupAbortedParts}. */
   cleanupAbortedParts,
+  /** Typed chat output stream for writing custom chunks or piping from subtasks. */
+  stream: chatStream,
 };
 
 /**

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

@@ -70,6 +70,46 @@ function ToolInvocation({ part }: { part: any }) {
   );
 }
 
+function ResearchProgress({ part }: { part: any }) {
+  const data = part.data as {
+    status: "fetching" | "done";
+    query: string;
+    current: number;
+    total: number;
+    currentUrl?: string;
+    completedUrls: string[];
+  };
+
+  const isDone = data.status === "done";
+
+  return (
+    <div className="my-2 rounded border border-blue-200 bg-blue-50 px-3 py-2 text-xs">
+      <div className="flex items-center gap-2 font-medium text-blue-700">
+        {isDone ? (
+          <span className="text-green-600">&#10003;</span>
+        ) : (
+          <span className="inline-block h-3 w-3 animate-spin rounded-full border-2 border-blue-300 border-t-blue-600" />
+        )}
+        <span>
+          {isDone
+            ? `Research complete — ${data.total} sources fetched`
+            : `Researching "${data.query}" (${data.current}/${data.total})`}
+        </span>
+      </div>
+      {data.currentUrl && !isDone && (
+        <div className="mt-1 truncate text-blue-500">Fetching {data.currentUrl}</div>
+      )}
+      {data.completedUrls.length > 0 && (
+        <div className="mt-1 space-y-0.5 text-blue-400">
+          {data.completedUrls.map((url, i) => (
+            <div key={i} className="truncate">&#10003; {url}</div>
+          ))}
+        </div>
+      )}
+    </div>
+  );
+}
+
 function DebugPanel({
   chatId,
   model,
@@ -321,10 +361,25 @@ export function Chat({
                     );
                   }
 
+                  if (part.type === "data-research-progress") {
+                    return <ResearchProgress key={i} part={part} />;
+                  }
+
                   if (part.type.startsWith("tool-") || part.type === "dynamic-tool") {
                     return <ToolInvocation key={i} part={part} />;
                   }
 
+                  if (part.type.startsWith("data-")) {
+                    return (
+                      <div key={i} className="my-1 rounded border border-gray-200 bg-gray-50 p-2 text-xs text-gray-500">
+                        <span className="font-medium">{part.type}</span>
+                        <pre className="mt-1 overflow-x-auto whitespace-pre-wrap">
+                          {JSON.stringify((part as any).data, null, 2)}
+                        </pre>
+                      </div>
+                    );
+                  }
+
                   return null;
                 })}
               </div>

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

@@ -1,5 +1,6 @@
-import { chat } from "@trigger.dev/sdk/ai";
-import { streamText, tool, stepCountIs } from "ai";
+import { chat, ai } from "@trigger.dev/sdk/ai";
+import { schemaTask } from "@trigger.dev/sdk";
+import { streamText, tool, stepCountIs, generateId } from "ai";
 import type { LanguageModel } from "ai";
 import { openai } from "@ai-sdk/openai";
 import { anthropic } from "@ai-sdk/anthropic";
@@ -135,6 +136,105 @@ const userContext = chat.local<{
   messageCount: number;
 }>();
 
+// --------------------------------------------------------------------------
+// Subtask: deep research — fetches multiple URLs and streams progress
+// back to the parent chat via chat.stream using data-* chunks
+// --------------------------------------------------------------------------
+export const deepResearch = schemaTask({
+  id: "deep-research",
+  description:
+    "Research a topic by fetching multiple URLs and synthesizing the results. " +
+    "Streams progress updates to the chat as it works.",
+  schema: z.object({
+    query: z.string().describe("The research query or topic"),
+    urls: z.array(z.string().url()).describe("URLs to fetch and analyze"),
+  }),
+  run: async ({ query, urls }) => {
+    const partId = generateId();
+    const results: { url: string; status: number; snippet: string }[] = [];
+
+    // Stream progress using data-research-progress chunks.
+    // Using the same id means each write updates the same part in the message.
+    function streamProgress(progress: {
+      status: "fetching" | "done";
+      query: string;
+      current: number;
+      total: number;
+      currentUrl?: string;
+      completedUrls: string[];
+    }) {
+      return chat.stream.writer({
+        target: "root",
+        execute: ({ write }) => {
+          write({
+            type: "data-research-progress" as any,
+            id: partId,
+            data: progress,
+          });
+        },
+      });
+    }
+
+    for (let i = 0; i < urls.length; i++) {
+      const url = urls[i]!;
+
+      // Update progress — fetching
+      const { waitUntilComplete } = streamProgress({
+        status: "fetching",
+        query,
+        current: i + 1,
+        total: urls.length,
+        currentUrl: url,
+        completedUrls: results.map((r) => r.url),
+      });
+      await waitUntilComplete();
+
+      try {
+        const response = await fetch(url);
+        let text = await response.text();
+        const contentType = response.headers.get("content-type") ?? "";
+
+        if (contentType.includes("html")) {
+          text = text
+            .replace(/<script[\s\S]*?<\/script>/gi, "")
+            .replace(/<style[\s\S]*?<\/style>/gi, "")
+            .replace(/<[^>]+>/g, " ")
+            .replace(/&nbsp;/g, " ")
+            .replace(/&amp;/g, "&")
+            .replace(/&lt;/g, "<")
+            .replace(/&gt;/g, ">")
+            .replace(/\s+/g, " ")
+            .trim();
+        }
+
+        results.push({
+          url,
+          status: response.status,
+          snippet: text.slice(0, 500),
+        });
+      } catch (err) {
+        results.push({
+          url,
+          status: 0,
+          snippet: `Error: ${err instanceof Error ? err.message : String(err)}`,
+        });
+      }
+    }
+
+    // Final progress update — done
+    const { waitUntilComplete: waitForDone } = streamProgress({
+      status: "done",
+      query,
+      current: urls.length,
+      total: urls.length,
+      completedUrls: results.map((r) => r.url),
+    });
+    await waitForDone();
+
+    return { query, results };
+  },
+});
+
 export const aiChat = chat.task({
   id: "ai-chat",
   clientDataSchema: z.object({ model: z.string().optional(), userId: z.string() }),
@@ -228,7 +328,11 @@ export const aiChat = chat.task({
       model: getModel(modelId),
       system: `You are a helpful assistant for ${userContext.name} (${userContext.plan} plan). Be concise and friendly.`,
       messages,
-      tools: { inspectEnvironment, webFetch },
+      tools: {
+        inspectEnvironment,
+        webFetch,
+        deepResearch: ai.tool(deepResearch),
+      },
       stopWhen: stepCountIs(10),
       abortSignal: stopSignal,
       providerOptions: {