From 21018294040fa5d9b47bfb7be2a6522acc815d3b Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sun, 8 Mar 2026 16:33:07 +0000 Subject: [PATCH] feat(chat): expose typed chat.stream, add deepResearch subtask example, per-chat model persistence, debug panel - Export chat.stream (typed RealtimeDefinedStream) 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 --- packages/trigger-sdk/src/v3/ai.ts | 27 ++++- references/ai-chat/src/components/chat.tsx | 55 +++++++++++ references/ai-chat/src/trigger/chat.ts | 110 ++++++++++++++++++++- 3 files changed, 188 insertions(+), 4 deletions(-) diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index 544bdd7a5..c2ffd55be 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/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({ 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, }; /** diff --git a/references/ai-chat/src/components/chat.tsx b/references/ai-chat/src/components/chat.tsx index f6e1916b5..e41b132db 100644 --- a/references/ai-chat/src/components/chat.tsx +++ b/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 ( +
+
+ {isDone ? ( + ✓ + ) : ( + + )} + + {isDone + ? `Research complete — ${data.total} sources fetched` + : `Researching "${data.query}" (${data.current}/${data.total})`} + +
+ {data.currentUrl && !isDone && ( +
Fetching {data.currentUrl}
+ )} + {data.completedUrls.length > 0 && ( +
+ {data.completedUrls.map((url, i) => ( +
✓ {url}
+ ))} +
+ )} +
+ ); +} + function DebugPanel({ chatId, model, @@ -321,10 +361,25 @@ export function Chat({ ); } + if (part.type === "data-research-progress") { + return ; + } + if (part.type.startsWith("tool-") || part.type === "dynamic-tool") { return ; } + if (part.type.startsWith("data-")) { + return ( +
+ {part.type} + 
+                          {JSON.stringify((part as any).data, null, 2)}
+
+
+ ); + } + return null; })} diff --git a/references/ai-chat/src/trigger/chat.ts b/references/ai-chat/src/trigger/chat.ts index 226e6bcad..0eed9cf08 100644 --- a/references/ai-chat/src/trigger/chat.ts +++ b/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(//gi, "") + .replace(//gi, "") + .replace(/<[^>]+>/g, " ") + .replace(/ /g, " ") + .replace(/&/g, "&") + .replace(/</g, "<") + .replace(/>/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: {