diff --git a/.changeset/trigger-chat-transport-watch-mode.md b/.changeset/trigger-chat-transport-watch-mode.md new file mode 100644 index 000000000..1b2346d85 --- /dev/null +++ b/.changeset/trigger-chat-transport-watch-mode.md @@ -0,0 +1,23 @@ +--- +"@trigger.dev/sdk": patch +--- + +Add `watch` option to `TriggerChatTransport` for read-only observation of an existing chat run. + +When set to `true`, the transport keeps its internal `ReadableStream` open across `trigger:turn-complete` control chunks instead of closing it after each turn. This lets a single `useChat` / `resumeStream` subscription observe every turn of a long-lived agent run — useful for dashboard viewers or debug UIs that only want to watch an existing conversation as it unfolds, rather than drive it. + +```tsx +const transport = new TriggerChatTransport({ + task: "my-chat-task", + accessToken: runScopedPat, + watch: true, + sessions: { + [chatId]: { runId, publicAccessToken: runScopedPat }, + }, +}); + +const { messages, resumeStream } = useChat({ id: chatId, transport }); +useEffect(() => { resumeStream(); }, [resumeStream]); +``` + +Non-watch transports are unaffected — the default remains `false` and existing behavior (close on turn-complete so `useChat` can flip to `"ready"` between turns) is preserved for interactive playground-style flows. diff --git a/.server-changes/run-agent-view.md b/.server-changes/run-agent-view.md new file mode 100644 index 000000000..76580157d --- /dev/null +++ b/.server-changes/run-agent-view.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: feature +--- + +Add an Agent view to the run details page for runs whose `taskKind` annotation is `AGENT`. The view renders the agent's `UIMessage` conversation by subscribing to the run's `chat` realtime stream — the same data source as the Agent Playground content view. Switching is via a `Trace view` / `Agent view` segmented control above the run body, and the selected view is reflected in the URL via `?view=agent` so it's shareable. diff --git a/apps/webapp/app/components/runs/v3/agent/AgentMessageView.tsx b/apps/webapp/app/components/runs/v3/agent/AgentMessageView.tsx new file mode 100644 index 000000000..72f019bb5 --- /dev/null +++ b/apps/webapp/app/components/runs/v3/agent/AgentMessageView.tsx @@ -0,0 +1,246 @@ +import type { UIMessage } from "@ai-sdk/react"; +import { memo } from "react"; +import { + AssistantResponse, + ChatBubble, + ToolUseRow, +} from "~/components/runs/v3/ai/AIChatMessages"; +import { Popover, PopoverContent, PopoverTrigger } from "~/components/primitives/Popover"; + +// --------------------------------------------------------------------------- +// AgentMessageView — renders an AI SDK UIMessage[] conversation. +// +// Extracted from the playground route so it can be reused on the run details +// page when the user picks the Agent view. +// +// UIMessage part types (AI SDK): +// text — markdown text content +// reasoning — model reasoning/thinking +// tool-{name} — tool call with input/output/state +// source-url — citation link +// source-document — citation document reference +// file — file attachment (image, etc.) +// step-start — visual separator between steps +// data-{name} — custom data parts (rendered as a small popover) +// --------------------------------------------------------------------------- + +export function AgentMessageView({ messages }: { messages: UIMessage[] }) { + return ( +
+ {messages.map((msg) => ( + + ))} +
+ ); +} + +// Memoized so stable messages (anything older than the one currently +// streaming) don't re-render on every chunk. This matters a lot during +// `resumeStream()` history replay, where each re-render would otherwise +// re-run Prism highlighting on every tool-call CodeBlock in the list. +// +// Default shallow prop comparison is fine: AI SDK's useChat keeps stable +// references for messages that haven't changed, so only the last message +// (the one receiving new chunks) re-renders. +export const MessageBubble = memo(function MessageBubble({ + message, +}: { + message: UIMessage; +}) { + if (message.role === "user") { + const text = + message.parts + ?.filter((p) => p.type === "text") + .map((p) => (p as { type: "text"; text: string }).text) + .join("") ?? ""; + + return ( +
+
+
{text}
+
+
+ ); + } + + if (message.role === "assistant") { + const hasContent = message.parts && message.parts.length > 0; + if (!hasContent) return null; + + return ( +
+ {message.parts?.map((part, i) => renderPart(part, i))} +
+ ); + } + + return null; +}); + +export function renderPart(part: UIMessage["parts"][number], i: number) { + const p = part as any; + const type = part.type as string; + + // Text — markdown rendered via AssistantResponse + if (type === "text") { + return p.text ? : null; + } + + // Reasoning — amber-bordered italic block + if (type === "reasoning") { + return ( +
+ +
+ {p.text ?? ""} +
+ +
+ ); + } + + // Tool call — type: "tool-{name}" with toolCallId, input, output, state + if (type.startsWith("tool-")) { + const toolName = type.slice(5); + + // Sub-agent tool: output is a UIMessage with parts + const isSubAgent = + p.output != null && typeof p.output === "object" && Array.isArray(p.output.parts); + + // For sub-agent tools, show the last text part as the "output" tab + // (mirrors what toModelOutput typically sends to the parent LLM) + // instead of dumping the full UIMessage JSON. + let resultOutput: string | undefined; + if (isSubAgent) { + const lastText = (p.output.parts as any[]) + .filter((part: any) => part.type === "text" && part.text) + .pop(); + resultOutput = lastText?.text ?? undefined; + } else if (p.output != null) { + resultOutput = + typeof p.output === "string" ? p.output : JSON.stringify(p.output, null, 2); + } + + return ( + + ); + } + + // Source URL — clickable citation link + if (type === "source-url") { + return ( +
+ + {p.title || p.url} + +
+ ); + } + + // Source document — citation label + if (type === "source-document") { + return ( +
+ {p.title} + {p.mediaType ? ` (${p.mediaType})` : ""} +
+ ); + } + + // File — render as image if image type, otherwise as download link + if (type === "file") { + const isImage = typeof p.mediaType === "string" && p.mediaType.startsWith("image/"); + if (isImage) { + return ( + {p.filename + ); + } + return ( +
+ + {p.filename ?? "Download file"} + +
+ ); + } + + // Step start — subtle dashed separator with centered label + if (type === "step-start") { + return ( +
+
+ step +
+
+ ); + } + + // Data parts — type: "data-{name}", show as labeled JSON popover + if (type.startsWith("data-")) { + const dataName = type.slice(5); + return ; + } + + return null; +} + +function DataPartPopover({ name, data }: { name: string; data: unknown }) { + const formatted = JSON.stringify(data, null, 2); + + return ( + + + + + +
+ data-{name} +
+
+ 
{formatted}
+
+ + + ); +} diff --git a/apps/webapp/app/components/runs/v3/agent/AgentView.tsx b/apps/webapp/app/components/runs/v3/agent/AgentView.tsx new file mode 100644 index 000000000..fd3dfb5cb --- /dev/null +++ b/apps/webapp/app/components/runs/v3/agent/AgentView.tsx @@ -0,0 +1,701 @@ +import type { UIMessage } from "@ai-sdk/react"; +import { SSEStreamSubscription } from "@trigger.dev/core/v3"; +import { useEffect, useMemo, useRef, useState } from "react"; +import { Paragraph } from "~/components/primitives/Paragraph"; +import { Spinner } from "~/components/primitives/Spinner"; +import { AgentMessageView } from "~/components/runs/v3/agent/AgentMessageView"; +import { useAutoScrollToBottom } from "~/hooks/useAutoScrollToBottom"; +import { useEnvironment } from "~/hooks/useEnvironment"; +import { useOrganization } from "~/hooks/useOrganizations"; +import { useProject } from "~/hooks/useProject"; + +export type AgentViewAuth = { + publicAccessToken: string; + apiOrigin: string; + /** + * User messages extracted from the run's task payload at load time. + * Empty array for runs started with `trigger: "preload"` — in that case + * the first user message will arrive over the chat-messages input stream + * and get merged in by the AgentView subscription. + */ + initialMessages: UIMessage[]; +}; + +type AgentViewRun = { + friendlyId: string; + taskIdentifier: string; +}; + +// Default stream IDs for Trigger.dev chat tasks — kept as literals so we +// don't pull server-only constants from `@trigger.dev/core/v3/chat-client` +// into a browser bundle. +const CHAT_STREAM_KEY = "chat"; +const CHAT_MESSAGES_STREAM_ID = "chat-messages"; + +/** + * Max state-update interval while assistant chunks are streaming. Matches + * the `experimental_throttle: 100` we previously passed to `useChat`. + * Chunks mutate a staging ref synchronously; a throttled flush copies the + * ref into React state at most ~10x/sec so tool-call Prism highlighting + * etc. doesn't re-run on every single text-delta. + */ +const STATE_FLUSH_THROTTLE_MS = 100; + +/** + * Sentinel timestamp for messages that came from the run's initial task + * payload — they predate any stream activity, so 0 guarantees they sort + * first regardless of stream race order. + */ +const INITIAL_PAYLOAD_TIMESTAMP = 0; + +/** + * Renders an agent run's chat conversation as it unfolds. + * + * Subscribes to two separate realtime streams for the run: + * - The **chat output stream** delivers assistant `UIMessageChunk`s (text + * deltas, tool calls, reasoning, etc.) produced by `pipeChat` on the + * task side. + * - The **chat-messages input stream** delivers user messages sent to the + * task via `sendInputStream` — each chunk carries a `ChatTaskWirePayload` + * with the most recent `messages` array. + * + * Both streams are read directly via `SSEStreamSubscription` through the + * dashboard's session-authed resource routes — not through `useChat` or + * `TriggerChatTransport`. This gives us per-chunk server-side timestamps + * (Redis stream IDs) from both streams, which we use to produce a + * chronologically correct merged message list that works for replays, + * multi-message turns, and steering messages. + * + * Intended to be mounted inside a scrollable container — the component + * does not own its own scrollbar. + */ +export function AgentView({ + run, + agentView, +}: { + run: AgentViewRun; + agentView: AgentViewAuth; +}) { + const organization = useOrganization(); + const project = useProject(); + const environment = useEnvironment(); + + const messages = useAgentRunMessages({ + runFriendlyId: run.friendlyId, + apiOrigin: agentView.apiOrigin, + orgSlug: organization.slug, + projectSlug: project.slug, + envSlug: environment.slug, + initialMessages: agentView.initialMessages, + }); + + // Sticky-bottom auto-scroll: walks up to find the inspector's scroll + // container, then scrolls to bottom whenever `messages` changes — but + // only if the user was at (or near) the bottom at the time. Scrolling + // away pauses auto-scroll; scrolling back resumes it. + const rootRef = useAutoScrollToBottom([messages]); + + return ( +
+ {messages.length === 0 ? ( +
+
+ + + Loading conversation… + +
+
+ ) : ( + + )} +
+ ); +} + +// --------------------------------------------------------------------------- +// useAgentRunMessages — reads both realtime streams for a run and maintains +// a chronologically ordered, merged message list. +// --------------------------------------------------------------------------- + +/** + * Shape of each chunk on the chat-messages input stream. Each chunk is a + * `ChatTaskWirePayload` whose `messages` field holds either the latest user + * message (for `submit-message`) or the full history (for + * `regenerate-message`). We dedupe by ID either way. + */ +type InputStreamChunk = { + messages?: Array<{ id?: string; role?: string; parts?: unknown[] }>; + trigger?: string; +}; + +/** + * Minimal typing for the chunks we care about on the chat output stream. + * Covers the AI SDK `UIMessageChunk` variants that `renderPart` actually + * knows how to display, plus the Trigger.dev control chunks that we filter. + */ +type OutputChunk = { type: string; [key: string]: unknown }; + +/** + * Per-message orchestration state for the output stream accumulator. Mirrors + * the active-part tracking that AI SDK's `processUIMessageStream` keeps in + * its `state` object: a registry of streaming text/reasoning parts so deltas + * can be matched to the right part by id, plus a way to clear them at step + * boundaries (`finish-step`) so the next step's `text-start`/`reasoning-start` + * with the same id starts a fresh part instead of appending to the previous + * step's part. + */ +/** + * Per-message orchestration state — index-based active-part tracking. + * + * Each map points from a part id (text or reasoning) to **the index of the + * currently-streaming part with that id in `message.parts`**. We need + * indexes (not just a `Set` of "active ids") because part ids are *only + * unique within a step*: the SDK happily reuses `text-start id="0"` after + * a `finish-step` boundary. Without index tracking, a `text-delta` for the + * reused id would have to find the right part by id alone — and a search + * would match BOTH the previous step's frozen part and the current step's + * fresh one, which produces a duplication where the previous text gets + * the new content appended to it AND a fresh part with the same content + * also appears. + * + * Mirrors AI SDK's `processUIMessageStream`'s `state.activeTextParts` / + * `state.activeReasoningParts` (which hold direct references in the + * mutating canonical impl). We use indexes here because we do immutable + * updates and need indices that survive `parts.map()` rewrites — adding + * new parts and updating existing ones never reorders, so an index is + * stable for the lifetime of the part. + */ +type MessageOrchestrationState = { + activeTextPartIndexes: Map; + activeReasoningPartIndexes: Map; +}; + +/** + * `SSEStreamSubscription`'s v2 batch path delivers `parsedBody.data` as-is + * — but for streams written via `sendInputStream` (which stores the user + * payload as a JSON string in the record body), `data` is itself a string + * that needs a second `JSON.parse` to recover the actual object. This + * happens for the chat-messages input stream because the action handler + * does `JSON.stringify(body.data.data)` before storing. + * + * Output streams from `pipeChat` write objects directly, so the v2 path + * delivers them already-parsed. Either way this helper accepts both shapes + * defensively: a string is parsed; an object is returned as-is. + * + * Returns `null` for unparseable / unexpected payloads. + */ +function parseChunkPayload(raw: unknown): Record | null { + if (raw == null) return null; + if (typeof raw === "string") { + try { + const parsed = JSON.parse(raw); + return parsed && typeof parsed === "object" ? (parsed as Record) : null; + } catch { + return null; + } + } + if (typeof raw === "object") return raw as Record; + return null; +} + +function createOrchestrationState(): MessageOrchestrationState { + return { + activeTextPartIndexes: new Map(), + activeReasoningPartIndexes: new Map(), + }; +} + +function useAgentRunMessages({ + runFriendlyId, + apiOrigin, + orgSlug, + projectSlug, + envSlug, + initialMessages, +}: { + runFriendlyId: string; + apiOrigin: string; + orgSlug: string; + projectSlug: string; + envSlug: string; + initialMessages: UIMessage[]; +}): UIMessage[] { + // Seed with the user messages from the run's task payload. + const seedMessages = useMemo( + () => initialMessages.filter((m) => m.role === "user"), + [initialMessages] + ); + + // `pendingRef` is the authoritative, eagerly-updated message state: + // chunks mutate this synchronously as they arrive. A throttled flush + // copies it into React state so UI updates are capped at ~10x/sec. + const pendingRef = useRef>( + new Map(seedMessages.map((m) => [m.id, m])) + ); + const timestampsRef = useRef>( + new Map(seedMessages.map((m) => [m.id, INITIAL_PAYLOAD_TIMESTAMP])) + ); + // Side-table of orchestration state, keyed by assistant message id. Lives + // outside the UIMessage so React doesn't see it as a renderable prop. + const orchestrationRef = useRef>(new Map()); + + // React state snapshot of pendingRef. Only updated via the throttled + // `scheduleFlush`. The Map *reference* changes on every flush so React + // detects the state update and the downstream `useMemo` recomputes. + const [messagesById, setMessagesById] = useState>( + () => new Map(pendingRef.current) + ); + + // Throttled flush scheduler — leading edge within a single throttle + // window: the first chunk after a quiet period flushes immediately, then + // subsequent chunks coalesce until the next window opens. + const lastFlushAtRef = useRef(0); + const pendingTimerRef = useRef | null>(null); + const scheduleFlush = useRef<() => void>(() => {}); + scheduleFlush.current = () => { + if (pendingTimerRef.current !== null) return; // already scheduled + const now = Date.now(); + const sinceLast = now - lastFlushAtRef.current; + const delay = Math.max(0, STATE_FLUSH_THROTTLE_MS - sinceLast); + pendingTimerRef.current = setTimeout(() => { + pendingTimerRef.current = null; + lastFlushAtRef.current = Date.now(); + setMessagesById(new Map(pendingRef.current)); + }, delay); + }; + + useEffect(() => { + const abort = new AbortController(); + + const outputUrl = + `${apiOrigin}/resources/orgs/${orgSlug}/projects/${projectSlug}/env/${envSlug}` + + `/runs/${runFriendlyId}/realtime/v1/streams/${runFriendlyId}/${CHAT_STREAM_KEY}`; + + const inputUrl = + `${apiOrigin}/resources/orgs/${orgSlug}/projects/${projectSlug}/env/${envSlug}` + + `/runs/${runFriendlyId}/realtime/v1/streams/${runFriendlyId}/input/${CHAT_MESSAGES_STREAM_ID}`; + + const commonSubOptions = { + signal: abort.signal, + timeoutInSeconds: 120, + } as const; + + // ---- Output stream: assistant messages --------------------------------- + // + // The output stream delivers UIMessageChunks interleaved with + // Trigger-specific control chunks (`trigger:turn-complete`, etc.). We + // filter the control chunks and fold everything else into an assistant + // `UIMessage` via our own `applyOutputChunk` accumulator — the AI SDK's + // `readUIMessageStream` helper is only available in `ai@6`, and the + // webapp is pinned to `ai@4`, so we re-implement just the chunk types + // that `renderPart` actually displays. + // + // We capture the **server timestamp of each assistant message's first + // `start` chunk** so later sort-by-timestamp merges with the input + // stream correctly. + const runOutput = async () => { + try { + const sub = new SSEStreamSubscription(outputUrl, commonSubOptions); + const raw = await sub.subscribe(); + const reader = raw.getReader(); + + let currentMessageId: string | null = null; + + try { + while (!abort.signal.aborted) { + const { done, value } = await reader.read(); + if (done) return; + + const chunk = parseChunkPayload(value.chunk) as OutputChunk | null; + if (!chunk || typeof chunk.type !== "string") continue; + if (chunk.type.startsWith("trigger:")) continue; + + if (chunk.type === "start") { + const messageId = + typeof chunk.messageId === "string" && chunk.messageId.length > 0 + ? chunk.messageId + : `asst-${crypto.randomUUID()}`; + currentMessageId = messageId; + + if (!timestampsRef.current.has(messageId)) { + timestampsRef.current.set(messageId, value.timestamp); + } + + const existing = pendingRef.current.get(messageId); + if (existing) { + // Same message id seen again — merge metadata only, keep + // existing parts (canonical `processUIMessageStream` does + // the same on a repeated `start`). + if (chunk.messageMetadata != null) { + pendingRef.current.set(messageId, { + ...existing, + metadata: { + ...((existing as { metadata?: Record }).metadata ?? {}), + ...(chunk.messageMetadata as Record), + }, + } as UIMessage); + scheduleFlush.current(); + } + } else { + const message: UIMessage = { + id: messageId, + role: "assistant", + parts: [], + ...(chunk.messageMetadata != null + ? { metadata: chunk.messageMetadata as UIMessage["metadata"] } + : {}), + } as UIMessage; + pendingRef.current.set(messageId, message); + orchestrationRef.current.set(messageId, createOrchestrationState()); + scheduleFlush.current(); + } + continue; + } + + if (currentMessageId === null) continue; + const existing = pendingRef.current.get(currentMessageId); + if (!existing) continue; + let orchestration = orchestrationRef.current.get(currentMessageId); + if (!orchestration) { + // Defensive: a chunk arrived for a message we never saw a + // `start` for. Lazily create orchestration state so we can + // still display the parts. + orchestration = createOrchestrationState(); + orchestrationRef.current.set(currentMessageId, orchestration); + } + + const updated = applyOutputChunk(existing, chunk, orchestration); + if (updated !== existing) { + pendingRef.current.set(currentMessageId, updated); + scheduleFlush.current(); + } + } + } finally { + try { + reader.releaseLock(); + } catch { + // Lock may already be released. + } + } + } catch (err) { + if (abort.signal.aborted) return; + // eslint-disable-next-line no-console + console.debug("[AgentView] output stream subscription failed", err); + } + }; + + // ---- Input stream: user messages --------------------------------------- + const runInput = async () => { + try { + const sub = new SSEStreamSubscription(inputUrl, commonSubOptions); + const raw = await sub.subscribe(); + const reader = raw.getReader(); + try { + while (!abort.signal.aborted) { + const { done, value } = await reader.read(); + if (done) return; + + const chunk = parseChunkPayload(value.chunk) as InputStreamChunk | null; + if (!chunk || !Array.isArray(chunk.messages)) continue; + + const incomingUsers = chunk.messages.filter( + (m): m is UIMessage => + m != null && (m as { role?: string }).role === "user" && typeof m.id === "string" + ); + if (incomingUsers.length === 0) continue; + + let changed = false; + for (const msg of incomingUsers) { + if (pendingRef.current.has(msg.id)) continue; + pendingRef.current.set(msg.id, msg); + timestampsRef.current.set(msg.id, value.timestamp); + changed = true; + } + if (changed) scheduleFlush.current(); + } + } finally { + try { + reader.releaseLock(); + } catch { + // Lock may already be released. + } + } + } catch (err) { + if (abort.signal.aborted) return; + // eslint-disable-next-line no-console + console.debug("[AgentView] input stream subscription failed", err); + } + }; + + void runOutput(); + void runInput(); + + return () => { + abort.abort(); + if (pendingTimerRef.current !== null) { + clearTimeout(pendingTimerRef.current); + pendingTimerRef.current = null; + } + }; + }, [runFriendlyId, apiOrigin, orgSlug, projectSlug, envSlug]); + + return useMemo(() => { + const timestamps = timestampsRef.current; + const arr = Array.from(messagesById.values()); + arr.sort((a, b) => { + const ta = timestamps.get(a.id) ?? 0; + const tb = timestamps.get(b.id) ?? 0; + if (ta !== tb) return ta - tb; + // Tie-breaker for messages sharing a stream ID bucket (rare): fall + // back to message id string order so the output is deterministic. + return a.id < b.id ? -1 : a.id > b.id ? 1 : 0; + }); + return arr; + }, [messagesById]); +} + +// --------------------------------------------------------------------------- +// applyOutputChunk — minimal UIMessageChunk → UIMessage accumulator. +// --------------------------------------------------------------------------- +// +// A pared-down re-implementation of AI SDK's `processUIMessageStream` (in +// `ai@6`'s `index.mjs`). The webapp is pinned to `ai@4`, which doesn't ship +// the v5+ chunk-stream helpers, so we vendor the bits we actually use. +// +// Scope vs. canonical: +// - We render only the chunk shapes that `AgentMessageView`/`renderPart` +// actually display: text, reasoning, tool-* (input-{start,delta,available} +// + output-{available,error}), source-url, source-document, file, +// step-start/finish-step, data-*, plus metadata/finish lifecycle. +// - Unknown chunk types fall through as no-ops — defensive on purpose for a +// read-only viewer. +// - We **do not parse partial JSON for streaming tool inputs.** Canonical +// uses `parsePartialJson` (which depends on a 300-line `fixJson` state +// machine to repair incomplete JSON) so users see the input growing +// character-by-character. We skip it: tool inputs stay `undefined` +// throughout streaming and snap to the final value when +// `tool-input-available` lands. Acceptable for a viewer; can be added +// later by vendoring `fixJson` if the UX warrants it. +// +// `orchestration` carries per-message active-part trackers that mirror +// canonical's `state.activeTextParts` / `state.activeReasoningParts`. They +// let `text-delta` find the right text part by id and let `finish-step` +// clear them so a new step can re-use the same id without colliding. +// +// Returns the same object reference when nothing changes so the caller can +// skip unnecessary state flushes + React re-renders. + +type AnyPart = { [key: string]: unknown; type: string }; + +function applyOutputChunk( + msg: UIMessage, + chunk: OutputChunk, + orchestration: MessageOrchestrationState +): UIMessage { + const type = chunk.type; + + // Text parts --------------------------------------------------------------- + // + // Track each streaming text part by its index in `msg.parts`. Part ids + // are only unique *within a step* — the SDK happily reuses `text-start + // id="0"` after a `finish-step` boundary — so a delta arriving for a + // reused id needs to land on the *current* part, not every prior part + // that ever shared that id. The index map gives us O(1) "which slot is + // currently streaming this id" without any id-based search. + if (type === "text-start") { + const id = chunk.id as string; + const newIndex = (msg.parts ?? []).length; // index AFTER push + orchestration.activeTextPartIndexes.set(id, newIndex); + return withNewPart(msg, { + type: "text", + id, + text: "", + state: "streaming", + }); + } + if (type === "text-delta") { + const id = chunk.id as string; + const index = orchestration.activeTextPartIndexes.get(id); + if (index === undefined) return msg; // delta with no start — drop. + return updatePartAt(msg, index, (p) => ({ + ...p, + text: ((p as { text?: string }).text ?? "") + String(chunk.delta ?? ""), + })); + } + if (type === "text-end") { + const id = chunk.id as string; + const index = orchestration.activeTextPartIndexes.get(id); + if (index === undefined) return msg; + orchestration.activeTextPartIndexes.delete(id); + return updatePartAt(msg, index, (p) => ({ ...p, state: "done" })); + } + + // Reasoning parts ---------------------------------------------------------- + if (type === "reasoning-start") { + const id = chunk.id as string; + const newIndex = (msg.parts ?? []).length; + orchestration.activeReasoningPartIndexes.set(id, newIndex); + return withNewPart(msg, { + type: "reasoning", + id, + text: "", + state: "streaming", + }); + } + if (type === "reasoning-delta") { + const id = chunk.id as string; + const index = orchestration.activeReasoningPartIndexes.get(id); + if (index === undefined) return msg; + return updatePartAt(msg, index, (p) => ({ + ...p, + text: ((p as { text?: string }).text ?? "") + String(chunk.delta ?? ""), + })); + } + if (type === "reasoning-end") { + const id = chunk.id as string; + const index = orchestration.activeReasoningPartIndexes.get(id); + if (index === undefined) return msg; + orchestration.activeReasoningPartIndexes.delete(id); + return updatePartAt(msg, index, (p) => ({ ...p, state: "done" })); + } + + // Tool call parts ---------------------------------------------------------- + if (type === "tool-input-start") { + const toolName = String(chunk.toolName ?? ""); + return withNewPart(msg, { + type: `tool-${toolName}`, + toolCallId: chunk.toolCallId, + toolName, + state: "input-streaming", + input: undefined, + }); + } + if (type === "tool-input-delta") { + // We don't parse partial JSON, so streaming tool input deltas are a + // no-op. The full input snaps in when `tool-input-available` arrives. + return msg; + } + if (type === "tool-input-available") { + const toolName = String(chunk.toolName ?? ""); + const existingIdx = indexOfPart( + msg, + (p) => (p as { toolCallId?: string }).toolCallId === chunk.toolCallId + ); + if (existingIdx >= 0) { + return updatePartAt(msg, existingIdx, (p) => ({ + ...p, + state: "input-available", + input: chunk.input, + })); + } + // Tool input arrived without a preceding tool-input-start (some + // providers do this for fast tools) — synthesize a new part. + return withNewPart(msg, { + type: `tool-${toolName}`, + toolCallId: chunk.toolCallId, + toolName, + state: "input-available", + input: chunk.input, + }); + } + if (type === "tool-output-available") { + return updatePart(msg, (p) => + (p as { toolCallId?: string }).toolCallId === chunk.toolCallId + ? { + ...p, + state: "output-available", + output: chunk.output, + ...(chunk.preliminary === true ? { preliminary: true } : {}), + } + : null + ); + } + if (type === "tool-output-error") { + return updatePart(msg, (p) => + (p as { toolCallId?: string }).toolCallId === chunk.toolCallId + ? { ...p, state: "output-error", errorText: chunk.errorText } + : null + ); + } + + // Source / file / step / data parts — pass through as a whole ------------- + if (type === "source-url" || type === "source-document" || type === "file") { + return withNewPart(msg, chunk as unknown as AnyPart); + } + if (type === "start-step") { + return withNewPart(msg, { type: "step-start" }); + } + if (type === "finish-step") { + // Step boundary — canonical clears the active part trackers so a new + // step can re-use the same text/reasoning part IDs cleanly. The + // message itself doesn't structurally change; the previous step's + // parts stay frozen at their indexes in `msg.parts`. + orchestration.activeTextPartIndexes.clear(); + orchestration.activeReasoningPartIndexes.clear(); + return msg; + } + if (type.startsWith("data-")) { + return withNewPart(msg, chunk as unknown as AnyPart); + } + + // Metadata / lifecycle ----------------------------------------------------- + if (type === "finish" || type === "message-metadata") { + if (chunk.messageMetadata == null) return msg; + return { + ...msg, + metadata: { + ...((msg as { metadata?: Record }).metadata ?? {}), + ...(chunk.messageMetadata as Record), + }, + } as UIMessage; + } + + // Abort / error / unknown — no structural change. (`start` is handled at + // the orchestration level in the output reader, not here.) + return msg; +} + +// --- Small immutable helpers for UIMessage.parts mutation ------------------- + +function withNewPart(msg: UIMessage, part: AnyPart): UIMessage { + return { + ...msg, + parts: [...((msg.parts ?? []) as AnyPart[]), part], + } as UIMessage; +} + +function updatePart( + msg: UIMessage, + updater: (part: AnyPart) => AnyPart | null +): UIMessage { + const parts = (msg.parts ?? []) as AnyPart[]; + let changed = false; + const next = parts.map((p) => { + const updated = updater(p); + if (updated === null) return p; + changed = true; + return updated; + }); + return changed ? ({ ...msg, parts: next } as UIMessage) : msg; +} + +function indexOfPart(msg: UIMessage, predicate: (part: AnyPart) => boolean): number { + const parts = (msg.parts ?? []) as AnyPart[]; + for (let i = 0; i < parts.length; i++) { + if (predicate(parts[i]!)) return i; + } + return -1; +} + +function updatePartAt( + msg: UIMessage, + index: number, + updater: (part: AnyPart) => AnyPart +): UIMessage { + const parts = (msg.parts ?? []) as AnyPart[]; + if (index < 0 || index >= parts.length) return msg; + const next = parts.slice(); + next[index] = updater(parts[index]!); + return { ...msg, parts: next } as UIMessage; +} diff --git a/apps/webapp/app/hooks/useAutoScrollToBottom.ts b/apps/webapp/app/hooks/useAutoScrollToBottom.ts new file mode 100644 index 000000000..b8e59687e --- /dev/null +++ b/apps/webapp/app/hooks/useAutoScrollToBottom.ts @@ -0,0 +1,104 @@ +import { useEffect, useLayoutEffect, useRef } from "react"; + +const AT_BOTTOM_TOLERANCE_PX = 16; + +/** + * Chat-style sticky-bottom auto-scroll behavior. + * + * Behavior: + * - On mount, finds the closest scrollable ancestor of the returned ref + * (the inspector content panel, the playground messages panel, etc.). + * - Tracks whether the user is currently "at the bottom" of that scroll + * container via a passive scroll listener. Default is `true` so the very + * first render of an existing conversation lands at the bottom, and the + * "content fits without scrolling" case stays in auto-scroll mode. + * - Whenever the dependency array changes (typically the messages array), + * if the user was at the bottom, programmatically scrolls to the new + * bottom. Uses `useLayoutEffect` so the scroll happens before paint and + * there's no one-frame flicker showing new content above the viewport. + * - Scrolling away from the bottom flips the ref to `false` → auto-scroll + * pauses. Scrolling back into the bottom band (within + * `AT_BOTTOM_TOLERANCE_PX`) flips it back to `true` → auto-scroll + * resumes. + * + * The programmatic scroll fires its own scroll event, which immediately + * re-runs the stickiness check and confirms we're still at the bottom + * (distance ≈ 0 ≤ tolerance), so the ref stays `true`. No special + * "ignore programmatic scroll" flag needed. + * + * @param deps Pass the rendered list (or any dependency that should + * trigger a re-scroll). Typically `[messages]`. + * @returns A ref to attach to the component's root element. The hook + * walks up from this element's parent to locate the scroll + * container, so the root must be mounted *inside* the + * scrollable region. + * + * @example + * ```tsx + * function ChatPanel({ messages }) { + * const rootRef = useAutoScrollToBottom([messages]); + * return ( + *
+ *
+ * {messages.map((m) => )} + *
+ *
+ * ); + * } + * ``` + */ +export function useAutoScrollToBottom(deps: ReadonlyArray) { + const rootRef = useRef(null); + const containerRef = useRef(null); + // Default true so initial mount + replay land at the bottom, and the + // no-overflow case stays sticky once content starts to grow. + const stickToBottomRef = useRef(true); + + // Locate the scroll container on mount and attach a passive scroll + // listener that updates `stickToBottomRef`. + useEffect(() => { + const findScrollContainer = (start: HTMLElement | null): HTMLElement | null => { + let current: HTMLElement | null = start; + while (current) { + const style = getComputedStyle(current); + const overflowY = style.overflowY; + if (overflowY === "auto" || overflowY === "scroll") return current; + current = current.parentElement; + } + return null; + }; + + const container = findScrollContainer(rootRef.current?.parentElement ?? null); + if (!container) return; + containerRef.current = container; + + const updateStickiness = () => { + const distanceFromBottom = + container.scrollHeight - container.scrollTop - container.clientHeight; + stickToBottomRef.current = distanceFromBottom <= AT_BOTTOM_TOLERANCE_PX; + }; + + // Seed from current position so the first messages-effect uses an + // accurate value rather than the default `true` if the user happened + // to mount the view already scrolled. + updateStickiness(); + + container.addEventListener("scroll", updateStickiness, { passive: true }); + return () => { + container.removeEventListener("scroll", updateStickiness); + containerRef.current = null; + }; + }, []); + + // After each commit that changes the deps (typically the messages + // array), if we were at the bottom, scroll to the new bottom. + useLayoutEffect(() => { + if (!stickToBottomRef.current) return; + const container = containerRef.current; + if (!container) return; + container.scrollTop = container.scrollHeight; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, deps); + + return rootRef; +} diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts index de41aee44..6ba1ead0e 100644 --- a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts @@ -1,12 +1,29 @@ import { type MachinePreset, + parsePacket, prettyPrintPacket, + RunAnnotations, SemanticInternalAttributes, type TaskRunContext, TaskRunError, TriggerTraceContext, type V3TaskRunContext, } from "@trigger.dev/core/v3"; + +/** + * Minimal structural type for the user messages we extract from an agent + * run's task payload. We deliberately avoid importing AI SDK's `UIMessage` + * here because the webapp's pinned `ai@4` declares a wider role union + * (`'data' | ...`) than `@ai-sdk/react@3`'s `UIMessage` accepts. The data + * crosses a JSON boundary anyway (typedjson) — keeping this loose lets the + * client-side type be the source of truth. + */ +type AgentInitialMessage = { + id: string; + role: "user" | "assistant" | "system"; + parts?: unknown[]; + [key: string]: unknown; +}; import { AttemptId, getMaxDuration, parseTraceparent } from "@trigger.dev/core/v3/isomorphic"; import { extractIdempotencyKeyScope, @@ -240,6 +257,30 @@ export class SpanPresenter extends BasePresenter { const externalTraceId = this.#getExternalTraceId(run.traceContext); + const taskKind = RunAnnotations.safeParse(run.annotations).data?.taskKind; + const isAgentRun = taskKind === "AGENT"; + + // For agent runs, extract the initial user messages that were supplied + // via the task payload (from the original `triggerTask({ payload: { messages: [...] } })` + // call). When the run was started with `trigger: "preload"`, this array + // will be empty — in that case the first user message arrives later via + // the chat-messages input stream and is picked up by the AgentView. + let agentInitialMessages: AgentInitialMessage[] = []; + if (isAgentRun && run.payload && run.payloadType !== "application/store") { + try { + const parsed = await parsePacket({ + data: typeof run.payload === "string" ? run.payload : JSON.stringify(run.payload), + dataType: run.payloadType ?? "application/json", + }); + if (parsed && typeof parsed === "object" && Array.isArray((parsed as any).messages)) { + agentInitialMessages = (parsed as any).messages as AgentInitialMessage[]; + } + } catch { + // Fall back to an empty initial message list — the AgentView will + // render whatever arrives over the input/output streams. + } + } + let region: { name: string; location: string | null } | null = null; if (run.runtimeEnvironment.type !== "DEVELOPMENT" && run.engine !== "V1") { @@ -297,6 +338,8 @@ export class SpanPresenter extends BasePresenter { isFinished, isRunning: RUNNING_STATUSES.includes(run.status), isError: isFailedRunStatus(run.status), + isAgentRun, + agentInitialMessages, payload, payloadType: run.payloadType, output, @@ -455,6 +498,7 @@ export class SpanPresenter extends BasePresenter { payloadType: true, metadata: true, metadataType: true, + annotations: true, maxAttempts: true, project: { include: { diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.$agentParam/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.$agentParam/route.tsx index 19583c4a3..aa6e30be5 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.$agentParam/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.$agentParam/route.tsx @@ -33,6 +33,8 @@ import { DateTime } from "~/components/primitives/DateTime"; import { cn } from "~/utils/cn"; import { JSONEditor } from "~/components/code/JSONEditor"; import { ToolUseRow, AssistantResponse, ChatBubble } from "~/components/runs/v3/ai/AIChatMessages"; +import { MessageBubble } from "~/components/runs/v3/agent/AgentMessageView"; +import { useAutoScrollToBottom } from "~/hooks/useAutoScrollToBottom"; import { ResizableHandle, ResizablePanel, @@ -353,6 +355,13 @@ function PlaygroundChat() { const isStreaming = status === "streaming"; const isSubmitted = status === "submitted"; + // Sticky-bottom auto-scroll for the messages list. The hook walks up to + // the surrounding `overflow-y-auto` panel and follows the conversation + // as new chunks stream in — pauses if you scroll up to read history, + // resumes when you scroll back into the bottom band. Same behavior as + // the run-inspector Agent tab. + const messagesRootRef = useAutoScrollToBottom([messages, isSubmitted]); + // Pending messages — steering during streaming const pending = usePlaygroundPendingMessages({ transport, @@ -530,7 +539,7 @@ function PlaygroundChat() {
) : ( -
+
{messages.map((msg) => ( ))} @@ -655,220 +664,9 @@ function formatAgentType(type: string): string { } } -// --------------------------------------------------------------------------- -// Message rendering -// --------------------------------------------------------------------------- - -// UIMessage part types (AI SDK): -// text — markdown text content -// reasoning — model reasoning/thinking -// tool-{name} — tool call with input/output/state -// source-url — citation link -// source-document — citation document reference -// file — file attachment (image, etc.) -// step-start — visual separator between steps (skip) -// data-{name} — custom data parts (skip) - -function MessageBubble({ message }: { message: UIMessage }) { - if (message.role === "user") { - const text = - message.parts - ?.filter((p) => p.type === "text") - .map((p) => (p as { type: "text"; text: string }).text) - .join("") ?? ""; - - return ( -
-
-
{text}
-
-
- ); - } - - if (message.role === "assistant") { - const hasContent = message.parts && message.parts.length > 0; - if (!hasContent) return null; - - return ( -
-
- {message.parts?.map((part, i) => renderPart(part, i))} -
-
- ); - } - - return null; -} - -function renderPart(part: UIMessage["parts"][number], i: number) { - const p = part as any; - const type = part.type as string; - - // Text — markdown rendered via AssistantResponse - if (type === "text") { - return p.text ? : null; - } - - // Reasoning — amber-bordered italic block - if (type === "reasoning") { - return ( -
- -
{p.text ?? ""}
- -
- ); - } - - // Tool call — type: "tool-{name}" with toolCallId, input, output, state - if (type.startsWith("tool-")) { - const toolName = type.slice(5); - - // Sub-agent tool: output is a UIMessage with parts - const isSubAgent = - p.output != null && - typeof p.output === "object" && - Array.isArray(p.output.parts); - - // For sub-agent tools, show the last text part as the "output" tab - // (mirrors what toModelOutput typically sends to the parent LLM) - // instead of dumping the full UIMessage JSON. - let resultOutput: string | undefined; - if (isSubAgent) { - const lastText = (p.output.parts as any[]) - .filter((part: any) => part.type === "text" && part.text) - .pop(); - resultOutput = lastText?.text ?? undefined; - } else if (p.output != null) { - resultOutput = - typeof p.output === "string" ? p.output : JSON.stringify(p.output, null, 2); - } - - return ( - - ); - } - - // Source URL — clickable citation link - if (type === "source-url") { - return ( - - ); - } - - // Source document — citation label - if (type === "source-document") { - return ( -
- 📄 {p.title} - {p.mediaType ? ` (${p.mediaType})` : ""} -
- ); - } - - // File — render as image if image type, otherwise as download link - if (type === "file") { - const isImage = typeof p.mediaType === "string" && p.mediaType.startsWith("image/"); - if (isImage) { - return ( - {p.filename - ); - } - return ( - - ); - } - - // Step start — subtle dashed separator with centered label - if (type === "step-start") { - return ( -
-
- step -
-
- ); - } - - // Data parts — type: "data-{name}", show as labeled JSON popover - if (type.startsWith("data-")) { - const dataName = type.slice(5); - return ; - } - - return null; -} - -function DataPartPopover({ name, data }: { name: string; data: unknown }) { - const formatted = JSON.stringify(data, null, 2); - - return ( - - - - - -
- data-{name} -
-
- 
{formatted}
-
- - - ); -} +// Message rendering — `MessageBubble` is imported from +// `~/components/runs/v3/agent/AgentMessageView`. The same module is used by +// the run details Agent view so both surfaces stay in sync. // --------------------------------------------------------------------------- // Sidebar diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.action.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.action.tsx index 6fb42dcb6..b72423d3e 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.action.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.action.tsx @@ -1,8 +1,5 @@ import { json } from "@remix-run/server-runtime"; import { type ActionFunctionArgs } from "@remix-run/server-runtime"; -import { - generateJWT as internal_generateJWT, -} from "@trigger.dev/core/v3"; import { z } from "zod"; import { prisma } from "~/db.server"; import { findProjectBySlug } from "~/models/project.server"; @@ -10,7 +7,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; import { requireUserId } from "~/services/session.server"; import { EnvironmentParamSchema } from "~/utils/pathBuilder"; import { TriggerTaskService } from "~/v3/services/triggerTask.server"; -import { extractJwtSigningSecretKey } from "~/services/realtime/jwtAuth.server"; +import { mintRunToken } from "~/services/realtime/mintRunToken.server"; const PlaygroundAction = z.object({ intent: z.enum(["create", "trigger", "renew", "save", "delete"]), @@ -165,7 +162,9 @@ export const action = async ({ request, params }: ActionFunctionArgs) => { }, }); - const jwt = await mintRunToken(environment, result.run.friendlyId); + const jwt = await mintRunToken(environment, result.run.friendlyId, { + includeInputStreamWrite: true, + }); return json({ runId: result.run.friendlyId, @@ -180,7 +179,7 @@ export const action = async ({ request, params }: ActionFunctionArgs) => { return json({ error: "runId is required" }, { status: 400 }); } - const jwt = await mintRunToken(environment, runId); + const jwt = await mintRunToken(environment, runId, { includeInputStreamWrite: true }); return json({ publicAccessToken: jwt }); } @@ -250,21 +249,3 @@ export const action = async ({ request, params }: ActionFunctionArgs) => { } } }; - -async function mintRunToken( - environment: Parameters[0], - runFriendlyId: string -): Promise { - return internal_generateJWT({ - secretKey: extractJwtSigningSecretKey(environment), - payload: { - sub: environment.id, - pub: true, - scopes: [ - `read:runs:${runFriendlyId}`, - `write:inputStreams:${runFriendlyId}`, - ], - }, - expirationTime: "1h", - }); -} diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts new file mode 100644 index 000000000..77cd00d94 --- /dev/null +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts @@ -0,0 +1,79 @@ +import { type LoaderFunctionArgs } from "@remix-run/server-runtime"; +import { z } from "zod"; +import { $replica } from "~/db.server"; +import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server"; +import { findProjectBySlug } from "~/models/project.server"; +import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; +import { requireUserId } from "~/services/session.server"; +import { EnvironmentParamSchema } from "~/utils/pathBuilder"; + +const ParamsSchema = z.object({ + runParam: z.string(), + runId: z.string(), + streamId: z.string(), +}); + +// GET: SSE stream subscription for a run's realtime output stream. +// +// The run-scoped equivalent of the playground stream route. Used by the +// Agent tab in the span inspector to subscribe to the run's chat output +// stream (streamed via `pipeChat` on the task side) through the dashboard +// instead of hitting the public API directly. +// +// Authenticated by the dashboard session — the user must have access to +// the project and environment. +export async function loader({ request, params }: LoaderFunctionArgs) { + const userId = await requireUserId(request); + const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params); + const { runParam, runId, streamId } = ParamsSchema.parse(params); + + // Defensive: callers should pass the same friendly ID for both the route + // `:runParam` segment and the stream `:runId` segment. + if (runParam !== runId) { + return new Response("Run ID mismatch", { status: 400 }); + } + + const project = await findProjectBySlug(organizationSlug, projectParam, userId); + if (!project) { + return new Response("Project not found", { status: 404 }); + } + + const environment = await findEnvironmentBySlug(project.id, envParam, userId); + if (!environment) { + return new Response("Environment not found", { status: 404 }); + } + + const run = await $replica.taskRun.findFirst({ + where: { + friendlyId: runId, + runtimeEnvironmentId: environment.id, + }, + select: { + id: true, + friendlyId: true, + realtimeStreamsVersion: true, + }, + }); + + if (!run) { + return new Response("Run not found", { status: 404 }); + } + + const lastEventId = request.headers.get("Last-Event-ID") || undefined; + const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined; + const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined; + + if ( + timeoutInSeconds && + (isNaN(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) + ) { + return new Response("Invalid timeout", { status: 400 }); + } + + const realtimeStream = getRealtimeStreamInstance(environment, run.realtimeStreamsVersion); + + return realtimeStream.streamResponse(request, run.friendlyId, streamId, request.signal, { + lastEventId, + timeoutInSeconds, + }); +} diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts new file mode 100644 index 000000000..65e634df7 --- /dev/null +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts @@ -0,0 +1,87 @@ +import { type LoaderFunctionArgs } from "@remix-run/server-runtime"; +import { z } from "zod"; +import { $replica } from "~/db.server"; +import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server"; +import { findProjectBySlug } from "~/models/project.server"; +import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; +import { requireUserId } from "~/services/session.server"; +import { EnvironmentParamSchema } from "~/utils/pathBuilder"; + +const ParamsSchema = z.object({ + runParam: z.string(), + runId: z.string(), + streamId: z.string(), +}); + +// GET: SSE stream subscription for a run's realtime INPUT stream. +// +// Dashboard-auth counterpart to the public API's +// `/realtime/v1/streams/:runId/input/:streamId` endpoint. Used by the Agent +// tab in the span inspector to observe user messages sent to an agent run +// over the `chat-messages` input stream. +// +// The underlying S2 stream name is `$trigger.input:${streamId}` (mirrors the +// naming used on the write side in `sendInputStream`). The realtime stream +// instance handles the actual SSE proxy; this route just enforces session +// auth and resolves the run. +export async function loader({ request, params }: LoaderFunctionArgs) { + const userId = await requireUserId(request); + const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params); + const { runParam, runId, streamId } = ParamsSchema.parse(params); + + // Defensive: callers should pass the same friendly ID for both the route + // `:runParam` segment and the stream `:runId` segment. + if (runParam !== runId) { + return new Response("Run ID mismatch", { status: 400 }); + } + + const project = await findProjectBySlug(organizationSlug, projectParam, userId); + if (!project) { + return new Response("Project not found", { status: 404 }); + } + + const environment = await findEnvironmentBySlug(project.id, envParam, userId); + if (!environment) { + return new Response("Environment not found", { status: 404 }); + } + + const run = await $replica.taskRun.findFirst({ + where: { + friendlyId: runId, + runtimeEnvironmentId: environment.id, + }, + select: { + id: true, + friendlyId: true, + realtimeStreamsVersion: true, + }, + }); + + if (!run) { + return new Response("Run not found", { status: 404 }); + } + + const lastEventId = request.headers.get("Last-Event-ID") || undefined; + const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined; + const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined; + + if ( + timeoutInSeconds && + (isNaN(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) + ) { + return new Response("Invalid timeout", { status: 400 }); + } + + const realtimeStream = getRealtimeStreamInstance(environment, run.realtimeStreamsVersion); + + return realtimeStream.streamResponse( + request, + run.friendlyId, + `$trigger.input:${streamId}`, + request.signal, + { + lastEventId, + timeoutInSeconds, + } + ); +} diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx index e0bec66ff..e495ac8c3 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx @@ -79,9 +79,14 @@ import { useProject } from "~/hooks/useProject"; import { useSearchParams } from "~/hooks/useSearchParam"; import { useHasAdminAccess } from "~/hooks/useUser"; import { useCanViewLogsPage } from "~/hooks/useCanViewLogsPage"; +import { findProjectBySlug } from "~/models/project.server"; import { redirectWithErrorMessage } from "~/models/message.server"; +import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; import { type Span, SpanPresenter, type SpanRun } from "~/presenters/v3/SpanPresenter.server"; +import { AgentView, type AgentViewAuth } from "~/components/runs/v3/agent/AgentView"; +import { env } from "~/env.server"; import { logger } from "~/services/logger.server"; +import { mintRunToken } from "~/services/realtime/mintRunToken.server"; import { requireUserId } from "~/services/session.server"; import { cn } from "~/utils/cn"; import { formatCurrencyAccurate } from "~/utils/numberFormatter"; @@ -124,7 +129,47 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { linkedRunId, }); - return typedjson(result); + if (!result) { + return redirectWithErrorMessage( + v3RunPath( + { slug: organizationSlug }, + { slug: projectParam }, + { slug: envParam }, + { friendlyId: runParam } + ), + request, + `Event not found.` + ); + } + + // For agent runs, mint a read-only run-scoped token so the Agent tab + // can subscribe to the run's chat output stream from the browser. We + // also forward the initial user messages extracted from the task + // payload so the AgentView can seed its merged message list (empty for + // runs started via `trigger: "preload"`). + let agentView: AgentViewAuth | null = null; + if (result.type === "run" && result.run.isAgentRun) { + const project = await findProjectBySlug(organizationSlug, projectParam, userId); + const environment = project + ? await findEnvironmentBySlug(project.id, envParam, userId) + : null; + if (environment) { + const publicAccessToken = await mintRunToken(environment, result.run.friendlyId); + agentView = { + publicAccessToken, + apiOrigin: env.API_ORIGIN || env.LOGIN_ORIGIN, + initialMessages: (result.run.agentInitialMessages ?? []) as AgentViewAuth["initialMessages"], + }; + } + } + + // Reconstruct the discriminated union explicitly. Spreading + // `{ ...result, agentView }` collapses the union and loses the + // `type === "run" | "span"` discriminant downstream in `SpanView`. + if (result.type === "run") { + return typedjson({ type: "run" as const, run: result.run, agentView }); + } + return typedjson({ type: "span" as const, span: result.span, agentView }); } catch (error) { logger.error("Error loading span", { projectParam, @@ -214,6 +259,7 @@ export function SpanView({ return ( void; @@ -405,6 +453,18 @@ function RunBody({ > Overview + {run.isAgentRun && ( + { + replace({ tab: "agent" }); + }} + shortcut={{ key: "a" }} + > + Agent + + )}
- {tab === "detail" ? ( + {tab === "agent" && run.isAgentRun && agentView ? ( + + ) : tab === "detail" ? (
diff --git a/apps/webapp/app/services/realtime/mintRunToken.server.ts b/apps/webapp/app/services/realtime/mintRunToken.server.ts new file mode 100644 index 000000000..2cdc4316e --- /dev/null +++ b/apps/webapp/app/services/realtime/mintRunToken.server.ts @@ -0,0 +1,41 @@ +import { generateJWT as internal_generateJWT } from "@trigger.dev/core/v3"; +import { extractJwtSigningSecretKey } from "./jwtAuth.server"; + +type Environment = Parameters[0]; + +export type MintRunTokenOptions = { + /** Include the input-stream write scope (needed for steering messages from the playground). */ + includeInputStreamWrite?: boolean; + /** Token expiration. Defaults to "1h". */ + expirationTime?: string; +}; + +/** + * Mint a run-scoped public access token (JWT) for browser subscription to a + * run's realtime streams. + * + * Used by: + * - The playground action to give a freshly triggered chat session a token. + * - The run details page to let the agent view subscribe to the chat stream + * of an existing run (read-only). + */ +export async function mintRunToken( + environment: Environment, + runFriendlyId: string, + options: MintRunTokenOptions = {} +): Promise { + const scopes = [`read:runs:${runFriendlyId}`]; + if (options.includeInputStreamWrite) { + scopes.push(`write:inputStreams:${runFriendlyId}`); + } + + return internal_generateJWT({ + secretKey: extractJwtSigningSecretKey(environment), + payload: { + sub: environment.id, + pub: true, + scopes, + }, + expirationTime: options.expirationTime ?? "1h", + }); +} diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index f224b156d..9753adc75 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -2375,6 +2375,8 @@ export type ChatAgentOptions< * waiting for the first message before suspending. * * Only applies to preloaded runs (triggered via `transport.preload()`). + * Takes precedence over `transport.preload(..., { idleTimeoutInSeconds })` + * and over {@link ChatAgentOptions.idleTimeoutInSeconds}. * * @default Same as `idleTimeoutInSeconds` */ @@ -2747,9 +2749,13 @@ function chatAgent< ); } - // Wait for the first real message — use preload-specific timeouts if configured + // Wait for the first real message — task-level idle settings win over + // `transport.preload(..., { idleTimeoutInSeconds })` / wire payload so + // `chat.agent({ idleTimeoutInSeconds, preloadIdleTimeoutInSeconds })` is authoritative. const effectivePreloadIdleTimeout = - payload.idleTimeoutInSeconds ?? preloadIdleTimeoutInSeconds ?? idleTimeoutInSeconds; + preloadIdleTimeoutInSeconds ?? + idleTimeoutInSeconds ?? + payload.idleTimeoutInSeconds; const effectivePreloadTimeout = (metadata.get(TURN_TIMEOUT_METADATA_KEY) as string | undefined) ?? @@ -4864,13 +4870,15 @@ function createChatSession( ): AsyncIterable { const { signal: runSignal, - idleTimeoutInSeconds = 30, + idleTimeoutInSeconds: sessionIdleTimeoutOpt, timeout = "1h", maxTurns = 100, compaction: sessionCompaction, pendingMessages: sessionPendingMessages, } = options; + const idleTimeoutInSeconds = sessionIdleTimeoutOpt ?? 30; + return { [Symbol.asyncIterator]() { let currentPayload = payload; @@ -4887,7 +4895,8 @@ function createChatSession( // First turn: handle preload — wait for the first real message if (turn === 0 && currentPayload.trigger === "preload") { const result = await messagesInput.waitWithIdleTimeout({ - idleTimeoutInSeconds: currentPayload.idleTimeoutInSeconds ?? idleTimeoutInSeconds, + idleTimeoutInSeconds: + sessionIdleTimeoutOpt ?? currentPayload.idleTimeoutInSeconds ?? 30, timeout, spanName: "waiting for first message", }); diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts index 00b50eb35..2e4d8955c 100644 --- a/packages/trigger-sdk/src/v3/chat.ts +++ b/packages/trigger-sdk/src/v3/chat.ts @@ -257,6 +257,27 @@ type TriggerChatTransportOptionsBase = { renewRunAccessToken?: ( params: RenewRunAccessTokenParams ) => string | undefined | null | Promise; + + /** + * Read-only "watch" mode for observing an existing chat run from the + * outside (e.g. a dashboard viewer that wants to show an agent run's + * conversation as it unfolds). + * + * When `true`, the transport no longer terminates its internal + * `ReadableStream` on the `trigger:turn-complete` control chunk. Instead, + * it forwards the session update, filters the control chunk, and keeps + * reading — so `useChat` receives chunks from turn 2, 3, etc. through a + * single long-lived subscription instead of needing a new `sendMessages` + * call to open the next turn's stream. + * + * You should also seed an existing `sessions` entry for the chat and drive + * the stream via `reconnectToStream` (or `useChat`'s `resumeStream`/`resume` + * option), and provide a placeholder `task` — a watch-mode transport never + * triggers new runs. + * + * @default false + */ + watch?: boolean; }; /** Access token used for frontend-triggered runs. */ @@ -360,6 +381,7 @@ export class TriggerChatTransport implements ChatTransport { private readonly streamTimeoutSeconds: number; private readonly defaultMetadata: Record | undefined; private readonly triggerOptions: TriggerChatTransportOptions["triggerOptions"]; + private readonly watchMode: boolean; private _onSessionChange: | (( chatId: string, @@ -396,6 +418,7 @@ export class TriggerChatTransport implements ChatTransport { this.triggerOptions = options.triggerOptions; this._onSessionChange = options.onSessionChange; this.renewRunAccessToken = options.renewRunAccessToken; + this.watchMode = options.watch ?? false; // Restore sessions from external storage if (options.sessions) { @@ -1040,6 +1063,15 @@ export class TriggerChatTransport implements ChatTransport { if (session) { this.notifySessionChange(chatId, session); } + + // Watch mode: keep the subscription open across turn + // boundaries so the consumer sees turn 2, 3, etc. through + // a single long-lived ReadableStream. Filter the control + // chunk and continue the read loop instead of closing. + if (this.watchMode) { + continue; + } + internalAbort.abort(); try { controller.close(); diff --git a/references/ai-chat/src/lib/pr-review-helpers.ts b/references/ai-chat/src/lib/pr-review-helpers.ts index d402a9942..1bfb082f1 100644 --- a/references/ai-chat/src/lib/pr-review-helpers.ts +++ b/references/ai-chat/src/lib/pr-review-helpers.ts @@ -52,14 +52,18 @@ export async function cloneRepo({ clonePath: string; token?: string | null; }): Promise { - const cloneUrl = token - ? `https://x-access-token:${token}@github.com/${owner}/${repo}.git` - : `https://github.com/${owner}/${repo}.git`; + async function runClone(): Promise { + const cloneUrl = token + ? `https://x-access-token:${token}@github.com/${owner}/${repo}.git` + : `https://github.com/${owner}/${repo}.git`; - logger.info("Cloning repo", { owner, repo, clonePath }); + await execFileAsync("git", ["clone", "--depth=1", cloneUrl, clonePath], { + timeout: 60_000, + }); + } - await execFileAsync("git", ["clone", "--depth=1", cloneUrl, clonePath], { - timeout: 60_000, + await logger.trace("cloneRepo", runClone, { + icon: "tabler-brand-github", }); } // #endregion diff --git a/references/ai-chat/src/lib/pr-review-tools.ts b/references/ai-chat/src/lib/pr-review-tools.ts index bd6498da8..de153dc1b 100644 --- a/references/ai-chat/src/lib/pr-review-tools.ts +++ b/references/ai-chat/src/lib/pr-review-tools.ts @@ -218,7 +218,15 @@ export const executeCode = tool({ return { description, success: true as const, - result: execResult.exports, + // Sanitize the sandbox's `module.exports` so the value matches the + // strict JSON shape that AI SDK's `jsonValueSchema` accepts. Raw JS + // can produce `Infinity`, `NaN`, `undefined`, `BigInt`, etc., none + // of which survive Zod v4's `z.number()` (which rejects non-finite + // numbers). The full message history is re-validated at the start + // of every subsequent `streamText` call, so an unsanitized value + // here would crash the agent on the *next* turn even though the + // current turn appears to succeed. + result: toJsonValue(execResult.exports), }; }); @@ -230,6 +238,34 @@ export const executeCode = tool({ return result; }, }); + +/** + * Coerce arbitrary JS to a value compatible with AI SDK's `jsonValueSchema` + * (`null | string | number | boolean | object | array`, where `number` must + * be finite). + * + * Uses `JSON.parse(JSON.stringify(...))` with a replacer so non-finite + * numbers become `null` (matching `JSON.stringify`'s default loss for + * `NaN`/`Infinity` when encountered as object values), `BigInt` is + * stringified, and `undefined` / functions are dropped — same coercions + * `JSON.stringify` already applies, but called explicitly so the result + * is a plain JSON value tree the SDK can re-validate on later turns. + */ +function toJsonValue(value: unknown): unknown { + try { + return JSON.parse( + JSON.stringify(value, (_key, v) => { + if (typeof v === "number" && !Number.isFinite(v)) return null; + if (typeof v === "bigint") return v.toString(); + return v; + }) + ); + } catch { + // Circular references or other JSON.stringify failures — fall back to a + // descriptive placeholder so the tool result is still valid JSON. + return { error: "Result was not JSON-serializable" }; + } +} // #endregion // #region Exports diff --git a/references/ai-chat/src/trigger/pr-review.ts b/references/ai-chat/src/trigger/pr-review.ts index f58e7754b..ae69107f5 100644 --- a/references/ai-chat/src/trigger/pr-review.ts +++ b/references/ai-chat/src/trigger/pr-review.ts @@ -157,8 +157,9 @@ export const prReviewChat = chat }) .agent({ id: "pr-review", - idleTimeoutInSeconds: 120, - chatAccessTokenTTL: "1m", + idleTimeoutInSeconds: 10, + preloadIdleTimeoutInSeconds: 10, + chatAccessTokenTTL: "60m", // #region onPreload — clone repo + fetch PRs before first message onPreload: async ({ chatId, clientData }) => {