github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Add run agent view

Browse Source
This commit is contained in:
Eric Allam
2026-04-10 12:16:07 +01:00
parent 77a755c753
commit e9586959b6
17 changed files with 1511 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.changeset/trigger-chat-transport-watch-mode.md
+23
View File
@@ -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.
.server-changes/run-agent-view.md
+6
View File
@@ -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.
apps/webapp/app/components/runs/v3/agent/AgentMessageView.tsx
+246
View File
@@ -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 (
<div className="mx-auto flex max-w-[800px] flex-col gap-2">
{messages.map((msg) => (
<MessageBubble key={msg.id} message={msg} />
))}
</div>
);
}
// 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 (
<div className="flex justify-end">
<div className="max-w-[80%] rounded-lg bg-indigo-600 px-4 py-2.5 text-sm text-white">
<div className="whitespace-pre-wrap">{text}</div>
</div>
</div>
);
}
if (message.role === "assistant") {
const hasContent = message.parts && message.parts.length > 0;
if (!hasContent) return null;
return (
<div className="space-y-2">
{message.parts?.map((part, i) => renderPart(part, i))}
</div>
);
}
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 ? <AssistantResponse key={i} text={p.text} headerLabel="" /> : null;
}
// Reasoning — amber-bordered italic block
if (type === "reasoning") {
return (
<div key={i} className="border-l-2 border-amber-500/40 pl-2">
<ChatBubble>
<div className="whitespace-pre-wrap text-xs italic text-amber-200/70">
{p.text ?? ""}
</div>
</ChatBubble>
</div>
);
}
// 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 (
<ToolUseRow
key={i}
tool={{
toolCallId: p.toolCallId ?? `tool-${i}`,
toolName,
inputJson: JSON.stringify(p.input ?? {}, null, 2),
resultOutput,
resultSummary:
p.state === "input-streaming" || p.state === "input-available"
? "calling..."
: p.state === "output-error"
? `error: ${p.errorText ?? "unknown"}`
: undefined,
subAgent: isSubAgent
? {
parts: p.output.parts,
isStreaming: p.state === "output-available" && p.preliminary === true,
}
: undefined,
}}
/>
);
}
// Source URL — clickable citation link
if (type === "source-url") {
return (
<div key={i} className="text-xs">
<a
href={p.url}
target="_blank"
rel="noopener noreferrer"
className="text-indigo-400 underline hover:text-indigo-300"
>
{p.title || p.url}
</a>
</div>
);
}
// Source document — citation label
if (type === "source-document") {
return (
<div key={i} className="text-xs text-text-dimmed">
{p.title}
{p.mediaType ? ` (${p.mediaType})` : ""}
</div>
);
}
// 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 (
<img
key={i}
src={p.url}
alt={p.filename ?? "file"}
className="max-h-64 rounded border border-charcoal-650"
/>
);
}
return (
<div key={i} className="text-xs">
<a
href={p.url}
target="_blank"
rel="noopener noreferrer"
className="text-indigo-400 underline hover:text-indigo-300"
>
{p.filename ?? "Download file"}
</a>
</div>
);
}
// Step start — subtle dashed separator with centered label
if (type === "step-start") {
return (
<div key={i} className="flex items-center gap-2 py-0.5">
<div className="flex-1 border-t border-dashed border-charcoal-650" />
<span className="text-[10px] text-charcoal-500">step</span>
<div className="flex-1 border-t border-dashed border-charcoal-650" />
</div>
);
}
// Data parts — type: "data-{name}", show as labeled JSON popover
if (type.startsWith("data-")) {
const dataName = type.slice(5);
return <DataPartPopover key={i} name={dataName} data={p.data} />;
}
return null;
}
function DataPartPopover({ name, data }: { name: string; data: unknown }) {
const formatted = JSON.stringify(data, null, 2);
return (
<Popover>
<PopoverTrigger asChild>
<button
type="button"
className="inline-flex items-center gap-1 rounded border border-charcoal-650 bg-charcoal-800 px-1.5 py-0.5 font-mono text-[10px] text-text-dimmed transition-colors hover:border-charcoal-500 hover:text-text-bright"
>
<span className="text-purple-400">{name}</span>
<span className="text-charcoal-500">{"{}"}</span>
</button>
</PopoverTrigger>
<PopoverContent className="w-auto max-w-md p-0" align="start" sideOffset={4}>
<div className="flex items-center justify-between border-b border-charcoal-650 px-2.5 py-1.5">
<span className="text-[10px] font-medium text-text-dimmed">data-{name}</span>
</div>
<div className="max-h-60 overflow-auto scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
<pre className="p-2.5 text-[11px] leading-relaxed text-text-bright">{formatted}</pre>
</div>
</PopoverContent>
</Popover>
);
}
apps/webapp/app/components/runs/v3/agent/AgentView.tsx
+701
View File
@@ -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 (
<div ref={rootRef} className="py-3">
{messages.length === 0 ? (
<div className="flex h-full min-h-[12rem] items-center justify-center">
<div className="flex flex-col items-center gap-3">
<Spinner className="size-5" color="muted" />
<Paragraph variant="small" className="text-text-dimmed">
Loading conversation…
</Paragraph>
</div>
</div>
) : (
<AgentMessageView messages={messages} />
)}
</div>
);
}
// ---------------------------------------------------------------------------
// 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<string, number>;
activeReasoningPartIndexes: Map<string, number>;
};
/**
* `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<string, unknown> | null {
if (raw == null) return null;
if (typeof raw === "string") {
try {
const parsed = JSON.parse(raw);
return parsed && typeof parsed === "object" ? (parsed as Record<string, unknown>) : null;
} catch {
return null;
}
}
if (typeof raw === "object") return raw as Record<string, unknown>;
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<Map<string, UIMessage>>(
new Map(seedMessages.map((m) => [m.id, m]))
);
const timestampsRef = useRef<Map<string, number>>(
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<Map<string, MessageOrchestrationState>>(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<Map<string, UIMessage>>(
() => 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<number>(0);
const pendingTimerRef = useRef<ReturnType<typeof setTimeout> | 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<string, unknown> }).metadata ?? {}),
...(chunk.messageMetadata as Record<string, unknown>),
},
} 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<string, unknown> }).metadata ?? {}),
...(chunk.messageMetadata as Record<string, unknown>),
},
} 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;
}
apps/webapp/app/hooks/useAutoScrollToBottom.ts
+104
View File
@@ -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 (
* <div className="overflow-y-auto h-full">
* <div ref={rootRef}>
* {messages.map((m) => <Message key={m.id} message={m} />)}
* </div>
* </div>
* );
* }
* ```
*/
export function useAutoScrollToBottom(deps: ReadonlyArray<unknown>) {
const rootRef = useRef<HTMLDivElement | null>(null);
const containerRef = useRef<HTMLElement | null>(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;
}
apps/webapp/app/presenters/v3/SpanPresenter.server.ts
+44
View File
@@ -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: {
apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.$agentParam/route.tsx
+13 -215
View File
@@ -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() {
</div>
</MainCenteredContainer>
) : (
<div className="mx-auto w-full max-w-4xl space-y-4">
<div ref={messagesRootRef} className="mx-auto w-full max-w-4xl space-y-4">
{messages.map((msg) => (
<MessageBubble key={msg.id} message={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 (
<div className="flex justify-end">
<div className="max-w-[80%] rounded-lg bg-indigo-600 px-4 py-2.5 text-sm text-white">
<div className="whitespace-pre-wrap">{text}</div>
</div>
</div>
);
}
if (message.role === "assistant") {
const hasContent = message.parts && message.parts.length > 0;
if (!hasContent) return null;
return (
<div className="flex justify-start">
<div className="max-w-[85%] space-y-2">
{message.parts?.map((part, i) => renderPart(part, i))}
</div>
</div>
);
}
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 ? <AssistantResponse key={i} text={p.text} headerLabel="" /> : null;
}
// Reasoning — amber-bordered italic block
if (type === "reasoning") {
return (
<div key={i} className="border-l-2 border-amber-500/40 pl-2">
<ChatBubble>
<div className="whitespace-pre-wrap text-xs italic text-amber-200/70">{p.text ?? ""}</div>
</ChatBubble>
</div>
);
}
// 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 (
<ToolUseRow
key={i}
tool={{
toolCallId: p.toolCallId ?? `tool-${i}`,
toolName,
inputJson: JSON.stringify(p.input ?? {}, null, 2),
resultOutput,
resultSummary:
p.state === "input-streaming" || p.state === "input-available"
? "calling..."
: p.state === "output-error"
? `error: ${p.errorText ?? "unknown"}`
: undefined,
subAgent: isSubAgent
? {
parts: p.output.parts,
isStreaming: p.state === "output-available" && p.preliminary === true,
}
: undefined,
}}
/>
);
}
// Source URL — clickable citation link
if (type === "source-url") {
return (
<div key={i} className="text-xs">
<a
href={p.url}
target="_blank"
rel="noopener noreferrer"
className="text-indigo-400 underline hover:text-indigo-300"
>
{p.title || p.url}
</a>
</div>
);
}
// Source document — citation label
if (type === "source-document") {
return (
<div key={i} className="text-xs text-text-dimmed">
📄 {p.title}
{p.mediaType ? ` (${p.mediaType})` : ""}
</div>
);
}
// 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 (
<img
key={i}
src={p.url}
alt={p.filename ?? "file"}
className="max-h-64 rounded border border-charcoal-650"
/>
);
}
return (
<div key={i} className="text-xs">
<a
href={p.url}
target="_blank"
rel="noopener noreferrer"
className="text-indigo-400 underline hover:text-indigo-300"
>
{p.filename ?? "Download file"}
</a>
</div>
);
}
// Step start — subtle dashed separator with centered label
if (type === "step-start") {
return (
<div key={i} className="flex items-center gap-2 py-0.5">
<div className="flex-1 border-t border-dashed border-charcoal-650" />
<span className="text-[10px] text-charcoal-500">step</span>
<div className="flex-1 border-t border-dashed border-charcoal-650" />
</div>
);
}
// Data parts — type: "data-{name}", show as labeled JSON popover
if (type.startsWith("data-")) {
const dataName = type.slice(5);
return <DataPartPopover key={i} name={dataName} data={p.data} />;
}
return null;
}
function DataPartPopover({ name, data }: { name: string; data: unknown }) {
const formatted = JSON.stringify(data, null, 2);
return (
<Popover>
<PopoverTrigger asChild>
<button
type="button"
className="inline-flex items-center gap-1 rounded border border-charcoal-650 bg-charcoal-800 px-1.5 py-0.5 font-mono text-[10px] text-text-dimmed transition-colors hover:border-charcoal-500 hover:text-text-bright"
>
<span className="text-purple-400">{name}</span>
<span className="text-charcoal-500">{"{}"}</span>
</button>
</PopoverTrigger>
<PopoverContent className="w-auto max-w-md p-0" align="start" sideOffset={4}>
<div className="flex items-center justify-between border-b border-charcoal-650 px-2.5 py-1.5">
<span className="text-[10px] font-medium text-text-dimmed">data-{name}</span>
</div>
<div className="max-h-60 overflow-auto scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
<pre className="p-2.5 text-[11px] leading-relaxed text-text-bright">{formatted}</pre>
</div>
</PopoverContent>
</Popover>
);
}
// 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
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.playground.action.tsx
+5 -24
View File
@@ -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<typeof extractJwtSigningSecretKey>[0],
runFriendlyId: string
): Promise<string> {
return internal_generateJWT({
secretKey: extractJwtSigningSecretKey(environment),
payload: {
sub: environment.id,
pub: true,
scopes: [
`read:runs:${runFriendlyId}`,
`write:inputStreams:${runFriendlyId}`,
],
},
expirationTime: "1h",
});
}
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts
+79
View File
@@ -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,
});
}
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts
+87
View File
@@ -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,
}
);
}
apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx
+67 -2
View File
@@ -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 (
<RunBody
run={fetcher.data.run}
agentView={fetcher.data.agentView}
runParam={runParam}
spanId={spanId}
closePanel={closePanel}
@@ -348,11 +394,13 @@ function applySpanOverrides(span: Span, spanOverrides?: SpanOverride): Span {
function RunBody({
run,
agentView,
runParam,
spanId,
closePanel,
}: {
run: SpanRun;
agentView: AgentViewAuth | null;
runParam: string;
spanId: string;
closePanel?: () => void;
@@ -405,6 +453,18 @@ function RunBody({
>
Overview
</TabButton>
{run.isAgentRun && (
<TabButton
isActive={tab === "agent"}
layoutId="span-run"
onClick={() => {
replace({ tab: "agent" });
}}
shortcut={{ key: "a" }}
>
Agent
</TabButton>
)}
<TabButton
isActive={tab === "detail"}
layoutId="span-run"
@@ -440,7 +500,12 @@ function RunBody({
</div>
<div className="overflow-y-auto px-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
<div>
{tab === "detail" ? (
{tab === "agent" && run.isAgentRun && agentView ? (
<AgentView
run={{ friendlyId: run.friendlyId, taskIdentifier: run.taskIdentifier }}
agentView={agentView}
/>
) : tab === "detail" ? (
<div className="flex flex-col gap-4 py-3">
<Property.Table>
<Property.Item>
apps/webapp/app/services/realtime/mintRunToken.server.ts
+41
View File
@@ -0,0 +1,41 @@
import { generateJWT as internal_generateJWT } from "@trigger.dev/core/v3";
import { extractJwtSigningSecretKey } from "./jwtAuth.server";
type Environment = Parameters<typeof extractJwtSigningSecretKey>[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<string> {
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",
});
}
packages/trigger-sdk/src/v3/ai.ts
+13 -4
View File
@@ -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<ChatTurn> {
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",
});
packages/trigger-sdk/src/v3/chat.ts
+32
View File
@@ -257,6 +257,27 @@ type TriggerChatTransportOptionsBase<TClientData = unknown> = {
renewRunAccessToken?: (
params: RenewRunAccessTokenParams
) => string | undefined | null | Promise<string | undefined | null>;
/**
* 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<UIMessage> {
private readonly streamTimeoutSeconds: number;
private readonly defaultMetadata: Record<string, unknown> | undefined;
private readonly triggerOptions: TriggerChatTransportOptions["triggerOptions"];
private readonly watchMode: boolean;
private _onSessionChange:
| ((
chatId: string,
@@ -396,6 +418,7 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
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<UIMessage> {
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();
references/ai-chat/src/lib/pr-review-helpers.ts
+10 -6
View File
@@ -52,14 +52,18 @@ export async function cloneRepo({
clonePath: string;
token?: string | null;
}): Promise<void> {
const cloneUrl = token
? `https://x-access-token:${token}@github.com/${owner}/${repo}.git`
: `https://github.com/${owner}/${repo}.git`;
async function runClone(): Promise<void> {
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
references/ai-chat/src/lib/pr-review-tools.ts
+37 -1
View File
@@ -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
references/ai-chat/src/trigger/pr-review.ts
+3 -2
View File
@@ -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 }) => {