feat(webapp): agent-view dashboard for chat.agent runs (3/4) (#3545)

## Summary

A chat-aware run inspector and a `/playground` UI for testing
`chat.agent` tasks interactively. Builds on #3543's runtime.

## Design

The run inspector grows a new tab that renders the conversation chain
for any `chat.agent`-kind run. It subscribes to the run's session
streams, threads chat parts through a per-message renderer, and uses a
shared markdown + Shiki component for code highlighting (also used by
the test-payload panel).

The playground is a standalone `/playground` route that lets you drive a
deployed chat agent from the dashboard — pick a task, send messages,
watch tool calls render, and see span detail on every turn. The matching
`/agents` list view shows all deployed agents in the project.
This commit is contained in:
Eric Allam
2026-05-14 16:58:15 +01:00
committed by GitHub
31 changed files with 5345 additions and 26 deletions
+12
View File
@@ -0,0 +1,12 @@
---
area: webapp
type: improvement
---
Migrate the dashboard Agent tab (span inspector) to subscribe to the backing Session's `.out` and `.in` channels instead of the run-scoped chat output + chat-messages input streams. Pairs with the SDK + MCP migrations on the ai-chat branch.
- `SpanPresenter.server.ts` extracts `agentSession` from the run payload (prefers `sessionId`, falls back to `chatId` for pre-Sessions agent runs — matches `resolveSessionByIdOrExternalId`).
- Span route threads `agentSession` through `AgentViewAuth` and gates `agentView` creation on having one.
- New dashboard resource route `resources.orgs.../runs.$runParam/realtime/v1/sessions/$sessionId/$io` proxies `S2RealtimeStreams.streamResponseFromSessionStream` under dashboard session auth. The run param binds resource hierarchy; the session identity is verified against the environment.
- `AgentView.tsx` subscribes to `/out` and `/in` URLs, drops local `CHAT_STREAM_KEY`/`CHAT_MESSAGES_STREAM_ID` constants, and parses the `.in` stream as `ChatInputChunk` (`{kind: "message", payload}` for user turns; `{kind: "stop"}` ignored). Output-stream parsing is unchanged — session v2 SSE already delivers UIMessageChunk objects from `record.body.data`.
- Smoke: opened a prior `test-agent` run in the dashboard, Agent tab rendered user + assistant messages end-to-end with zero console errors. Both SSE endpoints (`/out`, `/in`) returned 200.
@@ -0,0 +1,8 @@
---
area: webapp
type: fix
---
Playground action now forwards `maxDuration`, `version` (as `lockToVersion`), and `region` from the sidebar form into the Session's `triggerConfig`. Previously the form fields rendered as working controls but were silently dropped (`void`-suppressed) because `SessionTriggerConfig` didn't accept them — runs ignored the user's max duration, version pin, and region selection. With the schema extended in core, the playground now plumbs them through to `ensureRunForSession`.
Also fixes stale `clientData` in the playground transport: the JSON editor's value was captured at construction and never updated, so per-turn `metadata` merges used the original value across the whole conversation. Added a `useEffect` that calls `transport.setClientData(...)` whenever `clientDataJson` changes.
+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 backing Session's `.out` and `.in` channels — 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.
+6
View File
@@ -0,0 +1,6 @@
---
area: webapp
type: improvement
---
Upgrade streamdown from v1.4.0 to v2.5.0. Custom Shiki syntax highlighting theme matching our CodeMirror dark theme colors. Consolidate duplicated lazy StreamdownRenderer into a shared component.
@@ -2,6 +2,7 @@ import {
AdjustmentsHorizontalIcon,
ArrowPathRoundedSquareIcon,
ArrowRightOnRectangleIcon,
ArrowsRightLeftIcon,
ArrowTopRightOnSquareIcon,
BeakerIcon,
BellAlertIcon,
@@ -10,6 +11,7 @@ import {
ClockIcon,
Cog8ToothIcon,
CogIcon,
CpuChipIcon,
CubeIcon,
ExclamationTriangleIcon,
FolderIcon,
@@ -69,7 +71,9 @@ import {
organizationTeamPath,
queryPath,
regionsPath,
v3AgentsPath,
v3ApiKeysPath,
v3PlaygroundPath,
v3BatchesPath,
v3BillingPath,
v3BuiltInDashboardPath,
@@ -88,6 +92,7 @@ import {
v3QueuesPath,
v3RunsPath,
v3SchedulesPath,
v3SessionsPath,
v3TestPath,
v3UsagePath,
v3WaitpointTokensPath,
@@ -467,6 +472,31 @@ export function SideMenu({
initialCollapsed={getSectionCollapsed(user.dashboardPreferences.sideMenu, "ai")}
onCollapseToggle={handleSectionToggle("ai")}
>
<SideMenuItem
name="Agents"
icon={CpuChipIcon}
activeIconColor="text-indigo-500"
inactiveIconColor="text-indigo-500"
to={v3AgentsPath(organization, project, environment)}
isCollapsed={isCollapsed}
/>
<SideMenuItem
name="Sessions"
icon={ArrowsRightLeftIcon}
activeIconColor="text-teal-500"
inactiveIconColor="text-teal-500"
to={v3SessionsPath(organization, project, environment)}
data-action="sessions"
isCollapsed={isCollapsed}
/>
<SideMenuItem
name="Playground"
icon={BeakerIcon}
activeIconColor="text-indigo-400"
inactiveIconColor="text-indigo-400"
to={v3PlaygroundPath(organization, project, environment)}
isCollapsed={isCollapsed}
/>
<SideMenuItem
name="Prompts"
icon={AIPromptsIcon}
@@ -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 w-full min-w-0 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 min-w-0 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 [overflow-wrap:anywhere]">{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>
);
}
@@ -0,0 +1,717 @@
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;
/**
* Session identifier the AgentView uses to address the backing
* {@link Session} when subscribing to `.in` / `.out`. Accepts either
* a `session_*` friendlyId or the transport-supplied externalId
* (typically the browser's `chatId`) — the dashboard resource route
* resolves either form via `resolveSessionByIdOrExternalId`.
*/
sessionId: 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 arrives over the session's `.in`
* channel and is merged in by the AgentView subscription.
*/
initialMessages: UIMessage[];
};
/**
* 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 a Session's chat conversation as it unfolds.
*
* Subscribes to both channels of the {@link Session}:
* - **`.out`** delivers assistant `UIMessageChunk`s (text deltas, tool
* calls, reasoning, etc.) produced by the agent's
* `chatStream.writer(...)` calls — objects, already parsed by the S2
* SSE reader.
* - **`.in`** delivers {@link ChatInputChunk}s sent by
* {@link TriggerChatTransport} (or any other session writer). Each
* chunk is a tagged union (`{kind: "message", payload}` for user
* turns, `{kind: "stop"}` for stop signals) — the AgentView only
* cares about `kind: "message"` and pulls `.payload.messages`.
*
* 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
* (S2 sequence numbers) from both streams, which we use to produce a
* chronologically correct merged message list that works for replays,
* multi-message turns, cross-run session resumes, and steering messages.
*
* Intended to be mounted inside a scrollable container — the component
* does not own its own scrollbar.
*/
export function AgentView({ agentView }: { agentView: AgentViewAuth }) {
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const messages = useAgentSessionMessages({
sessionId: agentView.sessionId,
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>
);
}
// ---------------------------------------------------------------------------
// useAgentSessionMessages — reads both realtime streams for a session and
// maintains a chronologically ordered, merged message list.
// ---------------------------------------------------------------------------
/**
* Shape of each chunk on the session's `.in` channel. Mirrors the
* `ChatInputChunk` tagged union produced by {@link TriggerChatTransport}:
* - `kind: "message"` carries a `ChatTaskWirePayload` in `.payload`
* (user-submitted messages or regenerate calls); we dedupe by id.
* - `kind: "stop"` is a stop signal — no messages, nothing to render
* here, so it's filtered.
*
* The server wraps records in `{data, id}` and writes `data` as a JSON
* string; SSE v2 delivers the parsed string back. {@link parseChunkPayload}
* re-parses to recover the object.
*/
type InputStreamChunk = {
kind?: "message" | "stop";
payload?: {
messages?: Array<{ id?: string; role?: string; parts?: unknown[] }>;
trigger?: string;
};
message?: 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 session channels diverge by direction:
*
* - `.in`: {@link TriggerChatTransport.serializeInputChunk} writes the
* `ChatInputChunk` as a JSON **string**, so `data` is a string that
* needs a second `JSON.parse` to recover the tagged union.
* - `.out`: the agent's `chatStream.writer(...)` writes
* {@link UIMessageChunk} **objects** directly; `data` arrives
* already-parsed.
*
* This helper accepts both shapes defensively: a string is parsed; an
* object is returned as-is. Returns `null` for unparseable 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 useAgentSessionMessages({
sessionId,
apiOrigin,
orgSlug,
projectSlug,
envSlug,
initialMessages,
}: {
sessionId: 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 encodedSession = encodeURIComponent(sessionId);
// Always use the page's own origin to avoid CORS preflight failures
// when the configured `apiOrigin` (e.g. `localhost`) differs from the
// origin the dashboard was loaded from (e.g. `127.0.0.1`). The dashboard
// resource route is same-origin by construction.
const origin = typeof window !== "undefined" ? window.location.origin : apiOrigin;
const sessionBase =
`${origin}/resources/orgs/${orgSlug}/projects/${projectSlug}/env/${envSlug}` +
`/sessions/${encodedSession}/realtime/v1`;
const outputUrl = `${sessionBase}/out`;
const inputUrl = `${sessionBase}/in`;
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 channel: user messages (`ChatInputChunk`) -------------------
//
// The transport appends a `{kind: "message", payload}` ChatInputChunk
// for every user turn (and `{kind: "stop"}` for stop signals). We pull
// user messages out of `payload.messages` for `kind: "message"` chunks
// and ignore the rest.
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 || chunk.kind !== "message") continue;
const payload = chunk.payload;
if (!payload || !Array.isArray(payload.messages)) continue;
const incomingUsers = payload.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;
}
};
}, [sessionId, 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;
}
@@ -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;
}
@@ -0,0 +1,288 @@
import {
type PrismaClientOrTransaction,
type RuntimeEnvironmentType,
type TaskTriggerSource,
} from "@trigger.dev/database";
import { ClickHouse } from "@internal/clickhouse";
import { z } from "zod";
import { $replica } from "~/db.server";
import { clickhouseClient } from "~/services/clickhouseInstance.server";
import { singleton } from "~/utils/singleton";
import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server";
export type AgentListItem = {
slug: string;
filePath: string;
createdAt: Date;
triggerSource: TaskTriggerSource;
config: unknown;
};
export type AgentActiveState = {
running: number;
suspended: number;
};
export class AgentListPresenter {
constructor(
private readonly clickhouse: ClickHouse,
private readonly _replica: PrismaClientOrTransaction
) {}
public async call({
organizationId,
projectId,
environmentId,
environmentType,
}: {
organizationId: string;
projectId: string;
environmentId: string;
environmentType: RuntimeEnvironmentType;
}) {
const currentWorker = await findCurrentWorkerFromEnvironment(
{
id: environmentId,
type: environmentType,
},
this._replica
);
if (!currentWorker) {
return {
agents: [],
activeStates: Promise.resolve({} as Record<string, AgentActiveState>),
conversationSparklines: Promise.resolve({} as Record<string, number[]>),
costSparklines: Promise.resolve({} as Record<string, number[]>),
tokenSparklines: Promise.resolve({} as Record<string, number[]>),
};
}
const agents = await this._replica.backgroundWorkerTask.findMany({
where: {
workerId: currentWorker.id,
triggerSource: "AGENT",
},
select: {
id: true,
slug: true,
filePath: true,
triggerSource: true,
config: true,
createdAt: true,
},
orderBy: {
slug: "asc",
},
});
const slugs = agents.map((a) => a.slug);
if (slugs.length === 0) {
return {
agents,
activeStates: Promise.resolve({} as Record<string, AgentActiveState>),
conversationSparklines: Promise.resolve({} as Record<string, number[]>),
costSparklines: Promise.resolve({} as Record<string, number[]>),
tokenSparklines: Promise.resolve({} as Record<string, number[]>),
};
}
// All queries are deferred for streaming
const activeStates = this.#getActiveStates(environmentId, slugs);
const conversationSparklines = this.#getConversationSparklines(environmentId, slugs);
const costSparklines = this.#getCostSparklines(environmentId, slugs);
const tokenSparklines = this.#getTokenSparklines(environmentId, slugs);
return { agents, activeStates, conversationSparklines, costSparklines, tokenSparklines };
}
/** Count runs currently executing vs suspended per agent */
async #getActiveStates(
environmentId: string,
slugs: string[]
): Promise<Record<string, AgentActiveState>> {
const queryFn = this.clickhouse.reader.query({
name: "agentActiveStates",
query: `SELECT
task_identifier,
countIf(status = 'EXECUTING') AS running,
countIf(status IN ('WAITING_TO_RESUME', 'QUEUED_EXECUTING')) AS suspended
FROM trigger_dev.task_runs_v2
WHERE environment_id = {environmentId: String}
AND task_identifier IN {slugs: Array(String)}
AND task_kind = 'AGENT'
AND status IN ('EXECUTING', 'WAITING_TO_RESUME', 'QUEUED_EXECUTING')
GROUP BY task_identifier`,
params: z.object({
environmentId: z.string(),
slugs: z.array(z.string()),
}),
schema: z.object({
task_identifier: z.string(),
running: z.coerce.number(),
suspended: z.coerce.number(),
}),
});
const [error, rows] = await queryFn({ environmentId, slugs });
if (error) {
console.error("Agent active states query failed:", error);
return {};
}
const result: Record<string, AgentActiveState> = {};
for (const row of rows) {
result[row.task_identifier] = { running: row.running, suspended: row.suspended };
}
return result;
}
/** 24h hourly sparkline of conversation (run) count per agent */
async #getConversationSparklines(
environmentId: string,
slugs: string[]
): Promise<Record<string, number[]>> {
const queryFn = this.clickhouse.reader.query({
name: "agentConversationSparklines",
query: `SELECT
task_identifier,
toStartOfHour(created_at) AS bucket,
count() AS val
FROM trigger_dev.task_runs_v2
WHERE environment_id = {environmentId: String}
AND task_identifier IN {slugs: Array(String)}
AND task_kind = 'AGENT'
AND created_at >= now() - INTERVAL 24 HOUR
GROUP BY task_identifier, bucket
ORDER BY task_identifier, bucket`,
params: z.object({
environmentId: z.string(),
slugs: z.array(z.string()),
}),
schema: z.object({
task_identifier: z.string(),
bucket: z.string(),
val: z.coerce.number(),
}),
});
return this.#buildSparklineMap(await queryFn({ environmentId, slugs }), slugs);
}
/** 24h hourly sparkline of LLM cost per agent */
async #getCostSparklines(
environmentId: string,
slugs: string[]
): Promise<Record<string, number[]>> {
const queryFn = this.clickhouse.reader.query({
name: "agentCostSparklines",
query: `SELECT
task_identifier,
toStartOfHour(start_time) AS bucket,
sum(total_cost) AS val
FROM trigger_dev.llm_metrics_v1
WHERE environment_id = {environmentId: String}
AND task_identifier IN {slugs: Array(String)}
AND start_time >= now() - INTERVAL 24 HOUR
GROUP BY task_identifier, bucket
ORDER BY task_identifier, bucket`,
params: z.object({
environmentId: z.string(),
slugs: z.array(z.string()),
}),
schema: z.object({
task_identifier: z.string(),
bucket: z.string(),
val: z.coerce.number(),
}),
});
return this.#buildSparklineMap(await queryFn({ environmentId, slugs }), slugs);
}
/** 24h hourly sparkline of total tokens per agent */
async #getTokenSparklines(
environmentId: string,
slugs: string[]
): Promise<Record<string, number[]>> {
const queryFn = this.clickhouse.reader.query({
name: "agentTokenSparklines",
query: `SELECT
task_identifier,
toStartOfHour(start_time) AS bucket,
sum(total_tokens) AS val
FROM trigger_dev.llm_metrics_v1
WHERE environment_id = {environmentId: String}
AND task_identifier IN {slugs: Array(String)}
AND start_time >= now() - INTERVAL 24 HOUR
GROUP BY task_identifier, bucket
ORDER BY task_identifier, bucket`,
params: z.object({
environmentId: z.string(),
slugs: z.array(z.string()),
}),
schema: z.object({
task_identifier: z.string(),
bucket: z.string(),
val: z.coerce.number(),
}),
});
return this.#buildSparklineMap(await queryFn({ environmentId, slugs }), slugs);
}
/** Convert ClickHouse query result to sparkline map with zero-filled 24 hourly buckets */
#buildSparklineMap(
queryResult: [Error, null] | [null, { task_identifier: string; bucket: string; val: number }[]],
slugs: string[]
): Record<string, number[]> {
const [error, rows] = queryResult;
if (error) {
console.error("Agent sparkline query failed:", error);
return {};
}
return this.#buildSparklineFromRows(rows, slugs);
}
#buildSparklineFromRows(
rows: { task_identifier: string; bucket: string; val: number }[],
slugs: string[]
): Record<string, number[]> {
const now = new Date();
const startHour = new Date(
Date.UTC(
now.getUTCFullYear(),
now.getUTCMonth(),
now.getUTCDate(),
now.getUTCHours() - 23,
0,
0,
0
)
);
const bucketKeys: string[] = [];
for (let i = 0; i < 24; i++) {
const h = new Date(startHour.getTime() + i * 3600_000);
bucketKeys.push(h.toISOString().slice(0, 13).replace("T", " ") + ":00:00");
}
const rowMap = new Map<string, number>();
for (const row of rows) {
rowMap.set(`${row.task_identifier}|${row.bucket}`, row.val);
}
const result: Record<string, number[]> = {};
for (const slug of slugs) {
result[slug] = bucketKeys.map((key) => rowMap.get(`${slug}|${key}`) ?? 0);
}
return result;
}
}
export const agentListPresenter = singleton("agentListPresenter", setupAgentListPresenter);
function setupAgentListPresenter() {
return new AgentListPresenter(clickhouseClient, $replica);
}
@@ -0,0 +1,147 @@
import type { RuntimeEnvironmentType, TaskRunStatus, TaskTriggerSource } from "@trigger.dev/database";
import { $replica } from "~/db.server";
import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server";
import { isFinalRunStatus } from "~/v3/taskStatus";
export type PlaygroundAgent = {
slug: string;
filePath: string;
triggerSource: TaskTriggerSource;
config: unknown;
payloadSchema: unknown;
};
export type PlaygroundConversation = {
id: string;
chatId: string;
title: string;
agentSlug: string;
runFriendlyId: string | null;
runStatus: TaskRunStatus | null;
clientData: unknown;
messages: unknown;
lastEventId: string | null;
isActive: boolean;
createdAt: Date;
updatedAt: Date;
};
export class PlaygroundPresenter {
async listAgents({
environmentId,
environmentType,
}: {
environmentId: string;
environmentType: RuntimeEnvironmentType;
}): Promise<PlaygroundAgent[]> {
const currentWorker = await findCurrentWorkerFromEnvironment(
{ id: environmentId, type: environmentType },
$replica
);
if (!currentWorker) return [];
return $replica.backgroundWorkerTask.findMany({
where: {
workerId: currentWorker.id,
triggerSource: "AGENT",
},
select: {
slug: true,
filePath: true,
triggerSource: true,
config: true,
payloadSchema: true,
},
orderBy: { slug: "asc" },
});
}
async getAgent({
environmentId,
environmentType,
agentSlug,
}: {
environmentId: string;
environmentType: RuntimeEnvironmentType;
agentSlug: string;
}): Promise<PlaygroundAgent | null> {
const currentWorker = await findCurrentWorkerFromEnvironment(
{ id: environmentId, type: environmentType },
$replica
);
if (!currentWorker) return null;
return $replica.backgroundWorkerTask.findFirst({
where: {
workerId: currentWorker.id,
triggerSource: "AGENT",
slug: agentSlug,
},
select: {
slug: true,
filePath: true,
triggerSource: true,
config: true,
payloadSchema: true,
},
});
}
async getRecentConversations({
environmentId,
agentSlug,
userId,
limit = 10,
}: {
environmentId: string;
agentSlug: string;
userId: string;
limit?: number;
}): Promise<PlaygroundConversation[]> {
const conversations = await $replica.playgroundConversation.findMany({
where: {
runtimeEnvironmentId: environmentId,
agentSlug,
userId,
},
select: {
id: true,
chatId: true,
title: true,
agentSlug: true,
clientData: true,
messages: true,
lastEventId: true,
createdAt: true,
updatedAt: true,
run: {
select: {
friendlyId: true,
status: true,
},
},
},
orderBy: { updatedAt: "desc" },
take: limit,
});
return conversations.map((c) => ({
id: c.id,
chatId: c.chatId,
title: c.title,
agentSlug: c.agentSlug,
runFriendlyId: c.run?.friendlyId ?? null,
runStatus: c.run?.status ?? null,
clientData: c.clientData,
messages: c.messages,
lastEventId: c.lastEventId,
isActive: c.run?.status ? !isFinalRunStatus(c.run.status) : false,
createdAt: c.createdAt,
updatedAt: c.updatedAt,
}));
}
}
export const playgroundPresenter = new PlaygroundPresenter();
@@ -0,0 +1,153 @@
import { type Span } from "@opentelemetry/api";
import { type PrismaClientOrTransaction } from "@trigger.dev/database";
import { env } from "~/env.server";
import { findDisplayableEnvironment } from "~/models/runtimeEnvironment.server";
import { resolveSessionByIdOrExternalId } from "~/services/realtime/sessions.server";
import { ServiceValidationError } from "~/v3/services/baseService.server";
import { startActiveSpan } from "~/v3/tracer.server";
export type SessionDetail = NonNullable<Awaited<ReturnType<SessionPresenter["call"]>>>;
export class SessionPresenter {
constructor(private readonly replica: PrismaClientOrTransaction) {}
public async call(args: {
userId: string;
environmentId: string;
sessionParam: string;
}) {
return startActiveSpan(
"SessionPresenter.call",
(span) => this.#call(args, span),
{
attributes: {
environmentId: args.environmentId,
sessionParam: args.sessionParam,
},
}
);
}
async #call(
{
userId,
environmentId,
sessionParam,
}: {
userId: string;
environmentId: string;
sessionParam: string;
},
rootSpan: Span
) {
const session = await startActiveSpan(
"SessionPresenter.resolveSession",
() => resolveSessionByIdOrExternalId(this.replica, environmentId, sessionParam)
);
if (!session) {
rootSpan.setAttribute("session.found", false);
return null;
}
rootSpan.setAttribute("session.found", true);
rootSpan.setAttribute("session.id", session.id);
const displayableEnvironment = await startActiveSpan(
"SessionPresenter.findDisplayableEnvironment",
() => findDisplayableEnvironment(environmentId, userId)
);
if (!displayableEnvironment) {
throw new ServiceValidationError("No environment found");
}
// Run history is append-only; latest first matches the runs list.
// 50 covers the vast majority of sessions; longer histories link out
// to the runs page via tag filter.
const sessionRuns = await startActiveSpan(
"SessionPresenter.findSessionRuns",
async (span) => {
const rows = await this.replica.sessionRun.findMany({
where: { sessionId: session.id },
orderBy: { triggeredAt: "desc" },
take: 50,
select: {
id: true,
runId: true,
reason: true,
triggeredAt: true,
},
});
span.setAttribute("sessionRuns.count", rows.length);
return rows;
}
);
const runIds = sessionRuns.map((r) => r.runId);
const runs = await startActiveSpan(
"SessionPresenter.findRuns",
async (span) => {
span.setAttribute("runIds.count", runIds.length);
return runIds.length > 0
? this.replica.taskRun.findMany({
where: { id: { in: runIds } },
select: { id: true, friendlyId: true, status: true },
})
: [];
}
);
const runsById = new Map(runs.map((r) => [r.id, r] as const));
const currentRun = session.currentRunId
? runsById.get(session.currentRunId) ??
(await startActiveSpan(
"SessionPresenter.findCurrentRunFallback",
() =>
this.replica.taskRun.findFirst({
where: { id: session.currentRunId! },
select: { id: true, friendlyId: true, status: true },
})
))
: null;
// The dashboard SSE route is cookie-authed, so `publicAccessToken` is
// unused — kept here to match the existing `AgentViewAuth` shape.
const addressingKey = session.externalId ?? session.friendlyId;
return {
id: session.id,
friendlyId: session.friendlyId,
externalId: session.externalId,
type: session.type,
taskIdentifier: session.taskIdentifier,
tags: session.tags ? [...session.tags].sort((a, b) => a.localeCompare(b)) : [],
metadata: session.metadata,
triggerConfig: session.triggerConfig,
streamBasinName: session.streamBasinName,
closedAt: session.closedAt ? session.closedAt.toISOString() : undefined,
closedReason: session.closedReason ?? undefined,
expiresAt: session.expiresAt ? session.expiresAt.toISOString() : undefined,
createdAt: session.createdAt.toISOString(),
updatedAt: session.updatedAt.toISOString(),
environment: displayableEnvironment,
currentRun: currentRun
? { friendlyId: currentRun.friendlyId, status: currentRun.status }
: null,
runs: sessionRuns.map((r) => {
const run = runsById.get(r.runId);
return {
id: r.id,
reason: r.reason,
triggeredAt: r.triggeredAt.toISOString(),
run: run
? { friendlyId: run.friendlyId, status: run.status }
: null,
};
}),
agentView: {
publicAccessToken: "",
apiOrigin: env.API_ORIGIN || env.LOGIN_ORIGIN,
sessionId: addressingKey,
initialMessages: [],
},
};
}
}
@@ -1,12 +1,14 @@
import {
type MachinePreset,
prettyPrintPacket,
RunAnnotations,
SemanticInternalAttributes,
type TaskRunContext,
TaskRunError,
TriggerTraceContext,
type V3TaskRunContext,
} from "@trigger.dev/core/v3";
import { AttemptId, getMaxDuration, parseTraceparent } from "@trigger.dev/core/v3/isomorphic";
import {
extractIdempotencyKeyScope,
@@ -240,6 +242,9 @@ export class SpanPresenter extends BasePresenter {
const externalTraceId = this.#getExternalTraceId(run.traceContext);
const taskKind = RunAnnotations.safeParse(run.annotations).data?.taskKind;
const isAgentRun = taskKind === "AGENT";
let region: { name: string; location: string | null } | null = null;
if (run.runtimeEnvironment.type !== "DEVELOPMENT" && run.engine !== "V1") {
@@ -256,6 +261,48 @@ export class SpanPresenter extends BasePresenter {
region = workerGroup ?? null;
}
// Only AGENT-tagged runs (chat.agent and friends) can be session-bound,
// so skip the SessionRun lookup for the much larger set of standard runs.
// Lookup is by the unique `runId` index, but the cheapest query is the
// one we don't run.
const sessionRun = isAgentRun
? await this._replica.sessionRun.findFirst({
where: { runId: run.id },
select: {
reason: true,
triggeredAt: true,
session: {
select: {
friendlyId: true,
externalId: true,
type: true,
taskIdentifier: true,
closedAt: true,
expiresAt: true,
},
},
},
})
: null;
const session = sessionRun
? {
friendlyId: sessionRun.session.friendlyId,
externalId: sessionRun.session.externalId,
type: sessionRun.session.type,
taskIdentifier: sessionRun.session.taskIdentifier,
status:
sessionRun.session.closedAt != null
? ("CLOSED" as const)
: sessionRun.session.expiresAt != null &&
sessionRun.session.expiresAt.getTime() < Date.now()
? ("EXPIRED" as const)
: ("ACTIVE" as const),
reason: sessionRun.reason,
triggeredAt: sessionRun.triggeredAt,
}
: undefined;
return {
id: run.id,
friendlyId: run.friendlyId,
@@ -297,6 +344,7 @@ export class SpanPresenter extends BasePresenter {
isFinished,
isRunning: RUNNING_STATUSES.includes(run.status),
isError: isFailedRunStatus(run.status),
isAgentRun,
payload,
payloadType: run.payloadType,
output,
@@ -315,6 +363,7 @@ export class SpanPresenter extends BasePresenter {
metadata,
maxDurationInSeconds: getMaxDuration(run.maxDurationInSeconds),
batch: run.batch ? { friendlyId: run.batch.friendlyId } : undefined,
session,
engine: run.engine,
region,
workerQueue: run.workerQueue,
@@ -455,6 +504,7 @@ export class SpanPresenter extends BasePresenter {
payloadType: true,
metadata: true,
metadataType: true,
annotations: true,
maxAttempts: true,
project: {
include: {
@@ -0,0 +1,360 @@
import { BeakerIcon, CpuChipIcon, MagnifyingGlassIcon } from "@heroicons/react/20/solid";
import { type MetaFunction } from "@remix-run/node";
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { Suspense } from "react";
import { TypedAwait, typeddefer, useTypedLoaderData } from "remix-typedjson";
import { RunsIcon } from "~/assets/icons/RunsIcon";
import { MainCenteredContainer, PageBody, PageContainer } from "~/components/layout/AppLayout";
import { Badge } from "~/components/primitives/Badge";
import { Header2 } from "~/components/primitives/Headers";
import { Input } from "~/components/primitives/Input";
import { LinkButton } from "~/components/primitives/Buttons";
import { NavBar, PageTitle } from "~/components/primitives/PageHeader";
import { Paragraph } from "~/components/primitives/Paragraph";
import { Spinner } from "~/components/primitives/Spinner";
import {
Table,
TableBlankRow,
TableBody,
TableCell,
TableCellMenu,
TableHeader,
TableHeaderCell,
TableRow,
} from "~/components/primitives/Table";
import { SimpleTooltip } from "~/components/primitives/Tooltip";
import { PopoverMenuItem } from "~/components/primitives/Popover";
import { TaskFileName } from "~/components/runs/v3/TaskPath";
import { useFuzzyFilter } from "~/hooks/useFuzzyFilter";
import { useEnvironment } from "~/hooks/useEnvironment";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import {
type AgentListItem,
type AgentActiveState,
agentListPresenter,
} from "~/presenters/v3/AgentListPresenter.server";
import { requireUserId } from "~/services/session.server";
import { EnvironmentParamSchema, v3RunsPath, v3PlaygroundAgentPath } from "~/utils/pathBuilder";
import { cn } from "~/utils/cn";
export const meta: MetaFunction = () => {
return [{ title: "Agents | Trigger.dev" }];
};
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
if (!project) {
throw new Response(undefined, { status: 404, statusText: "Project not found" });
}
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
if (!environment) {
throw new Response(undefined, { status: 404, statusText: "Environment not found" });
}
const result = await agentListPresenter.call({
organizationId: project.organizationId,
projectId: project.id,
environmentId: environment.id,
environmentType: environment.type,
});
return typeddefer(result);
};
export default function AgentsPage() {
const { agents, activeStates, conversationSparklines, costSparklines, tokenSparklines } =
useTypedLoaderData<typeof loader>();
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const { filterText, setFilterText, filteredItems } = useFuzzyFilter({
items: agents,
keys: ["slug", "filePath"],
});
if (agents.length === 0) {
return (
<PageContainer>
<NavBar>
<PageTitle title="Agents" />
</NavBar>
<PageBody>
<MainCenteredContainer>
<div className="flex flex-col items-center gap-4 py-20">
<CpuChipIcon className="size-12 text-indigo-500" />
<Header2>No agents deployed</Header2>
<Paragraph variant="small" className="max-w-md text-center">
Create a chat agent using <code>chat.agent()</code> from{" "}
<code>@trigger.dev/sdk/ai</code> and deploy it to see it here.
</Paragraph>
</div>
</MainCenteredContainer>
</PageBody>
</PageContainer>
);
}
return (
<PageContainer>
<NavBar>
<PageTitle title="Agents" />
</NavBar>
<PageBody scrollable={false}>
<div className="grid h-full grid-rows-1">
<div className="flex min-w-0 max-w-full flex-col">
<div className="max-h-full overflow-hidden">
<div className="flex items-center gap-1 p-2">
<Input
placeholder="Search agents"
variant="tertiary"
icon={MagnifyingGlassIcon}
fullWidth={true}
value={filterText}
onChange={(e) => setFilterText(e.target.value)}
autoFocus
/>
</div>
<Table containerClassName="max-h-full pb-[2.5rem]">
<TableHeader>
<TableRow>
<TableHeaderCell>ID</TableHeaderCell>
<TableHeaderCell>Type</TableHeaderCell>
<TableHeaderCell>File</TableHeaderCell>
<TableHeaderCell>Active</TableHeaderCell>
<TableHeaderCell>Conversations (24h)</TableHeaderCell>
<TableHeaderCell>Cost (24h)</TableHeaderCell>
<TableHeaderCell>Tokens (24h)</TableHeaderCell>
<TableHeaderCell hiddenLabel>Go to page</TableHeaderCell>
</TableRow>
</TableHeader>
<TableBody>
{filteredItems.length > 0 ? (
filteredItems.map((agent) => {
const path = v3RunsPath(organization, project, environment, {
tasks: [agent.slug],
});
const agentType =
(agent.config as { type?: string } | null)?.type ?? "unknown";
return (
<TableRow key={agent.slug} className="group">
<TableCell to={path} isTabbableCell>
<div className="flex items-center gap-2">
<SimpleTooltip
button={
<CpuChipIcon className="size-[1.125rem] min-w-[1.125rem] text-indigo-500" />
}
content="Agent"
/>
<span>{agent.slug}</span>
</div>
</TableCell>
<TableCell to={path}>
<Badge variant="extra-small">{formatAgentType(agentType)}</Badge>
</TableCell>
<TableCell to={path}>
<TaskFileName fileName={agent.filePath} variant="extra-extra-small" />
</TableCell>
<TableCell to={path}>
<Suspense fallback={<Spinner color="muted" />}>
<TypedAwait resolve={activeStates} errorElement={<></>}>
{(data) => {
const state = data[agent.slug];
if (!state || (state.running === 0 && state.suspended === 0)) {
return (
<span className="text-text-dimmed"></span>
);
}
return (
<span className="flex items-center gap-1.5 text-xs">
{state.running > 0 && (
<span className="flex items-center gap-0.5">
<span className="size-1.5 rounded-full bg-success" />
<span>{state.running}</span>
</span>
)}
{state.running > 0 && state.suspended > 0 && (
<span className="text-text-dimmed">·</span>
)}
{state.suspended > 0 && (
<span className="flex items-center gap-0.5">
<span className="size-1.5 rounded-full bg-blue-500" />
<span>{state.suspended}</span>
</span>
)}
</span>
);
}}
</TypedAwait>
</Suspense>
</TableCell>
<TableCell to={path} actionClassName="py-1.5">
<Suspense fallback={<SparklinePlaceholder />}>
<TypedAwait resolve={conversationSparklines} errorElement={<></>}>
{(data) => (
<SparklineWithTotal
data={data[agent.slug]}
formatTotal={formatCount}
/>
)}
</TypedAwait>
</Suspense>
</TableCell>
<TableCell to={path} actionClassName="py-1.5">
<Suspense fallback={<SparklinePlaceholder />}>
<TypedAwait resolve={costSparklines} errorElement={<></>}>
{(data) => (
<SparklineWithTotal
data={data[agent.slug]}
formatTotal={formatCost}
color="text-amber-400"
barColor="#F59E0B"
/>
)}
</TypedAwait>
</Suspense>
</TableCell>
<TableCell to={path} actionClassName="py-1.5">
<Suspense fallback={<SparklinePlaceholder />}>
<TypedAwait resolve={tokenSparklines} errorElement={<></>}>
{(data) => (
<SparklineWithTotal
data={data[agent.slug]}
formatTotal={formatTokens}
color="text-purple-400"
barColor="#A855F7"
/>
)}
</TypedAwait>
</Suspense>
</TableCell>
<TableCellMenu
isSticky
popoverContent={
<>
<PopoverMenuItem
icon={RunsIcon}
to={path}
title="View runs"
leadingIconClassName="text-runs"
/>
<PopoverMenuItem
icon={BeakerIcon}
to={v3PlaygroundAgentPath(organization, project, environment, agent.slug)}
title="Playground"
leadingIconClassName="text-indigo-400"
/>
</>
}
hiddenButtons={
<LinkButton
variant="minimal/small"
LeadingIcon={BeakerIcon}
leadingIconClassName="text-text-bright"
to={v3PlaygroundAgentPath(organization, project, environment, agent.slug)}
>
Playground
</LinkButton>
}
/>
</TableRow>
);
})
) : (
<TableBlankRow colSpan={8}>
<Paragraph variant="small" className="flex items-center justify-center">
No agents match your filters
</Paragraph>
</TableBlankRow>
)}
</TableBody>
</Table>
</div>
</div>
</div>
</PageBody>
</PageContainer>
);
}
function formatAgentType(type: string): string {
switch (type) {
case "ai-sdk-chat":
return "AI SDK Chat";
default:
return type;
}
}
function formatCount(total: number): string {
if (total === 0) return "0";
if (total >= 1000) return `${(total / 1000).toFixed(1)}k`;
return total.toString();
}
function formatCost(total: number): string {
if (total === 0) return "$0";
if (total < 0.01) return `$${total.toFixed(4)}`;
if (total < 1) return `$${total.toFixed(2)}`;
return `$${total.toFixed(2)}`;
}
function formatTokens(total: number): string {
if (total === 0) return "0";
if (total >= 1_000_000) return `${(total / 1_000_000).toFixed(1)}M`;
if (total >= 1000) return `${(total / 1000).toFixed(1)}k`;
return total.toString();
}
function SparklinePlaceholder() {
return <div className="h-6 w-24" />;
}
function SparklineWithTotal({
data,
formatTotal,
color = "text-text-bright",
barColor = "#3B82F6",
}: {
data?: number[];
formatTotal: (total: number) => string;
color?: string;
barColor?: string;
}) {
if (!data || data.every((v) => v === 0)) {
return <span className="text-text-dimmed"></span>;
}
const total = data.reduce((sum, v) => sum + v, 0);
const max = Math.max(...data);
return (
<div className="flex items-center gap-2">
<div className="flex h-5 items-end gap-px">
{data.map((value, i) => {
const height = max > 0 ? Math.max((value / max) * 100, value > 0 ? 8 : 0) : 0;
return (
<div
key={i}
className="w-[3px] rounded-t-[1px]"
style={{
height: `${height}%`,
backgroundColor: value > 0 ? barColor : "transparent",
opacity: value > 0 ? 0.8 : 0,
}}
/>
);
})}
</div>
<span className={cn("text-xs tabular-nums", color)}>{formatTotal(total)}</span>
</div>
);
}
@@ -0,0 +1,189 @@
import { BookOpenIcon, CpuChipIcon } from "@heroicons/react/20/solid";
import { json, type MetaFunction } from "@remix-run/node";
import { Outlet, useNavigate, useParams, useLoaderData } from "@remix-run/react";
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { CodeBlock } from "~/components/code/CodeBlock";
import { InlineCode } from "~/components/code/InlineCode";
import { MainCenteredContainer, PageBody, PageContainer } from "~/components/layout/AppLayout";
import { LinkButton } from "~/components/primitives/Buttons";
import { Header2 } from "~/components/primitives/Headers";
import { InfoPanel } from "~/components/primitives/InfoPanel";
import { NavBar, PageTitle } from "~/components/primitives/PageHeader";
import { Paragraph } from "~/components/primitives/Paragraph";
import {
Select,
SelectItem,
} from "~/components/primitives/Select";
import { $replica } from "~/db.server";
import { useEnvironment } from "~/hooks/useEnvironment";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { playgroundPresenter } from "~/presenters/v3/PlaygroundPresenter.server";
import { RegionsPresenter } from "~/presenters/v3/RegionsPresenter.server";
import { requireUser } from "~/services/session.server";
import { docsPath, EnvironmentParamSchema, v3PlaygroundAgentPath } from "~/utils/pathBuilder";
export const meta: MetaFunction = () => {
return [{ title: "Playground | Trigger.dev" }];
};
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const user = await requireUser(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const project = await findProjectBySlug(organizationSlug, projectParam, user.id);
if (!project) {
throw new Response(undefined, { status: 404, statusText: "Project not found" });
}
const environment = await findEnvironmentBySlug(project.id, envParam, user.id);
if (!environment) {
throw new Response(undefined, { status: 404, statusText: "Environment not found" });
}
const [agents, backgroundWorkers, regionsResult] = await Promise.all([
playgroundPresenter.listAgents({
environmentId: environment.id,
environmentType: environment.type,
}),
$replica.backgroundWorker.findMany({
where: { runtimeEnvironmentId: environment.id },
select: { version: true },
orderBy: { createdAt: "desc" },
take: 20,
}),
new RegionsPresenter().call({
userId: user.id,
projectSlug: projectParam,
isAdmin: user.admin || user.isImpersonating,
}),
]);
return json({
agents,
versions: backgroundWorkers.map((w) => w.version),
regions: regionsResult.regions,
isDev: environment.type === "DEVELOPMENT",
});
};
export default function PlaygroundPage() {
const { agents } = useLoaderData<typeof loader>();
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const navigate = useNavigate();
const params = useParams();
const selectedAgent = params.agentParam ?? "";
if (agents.length === 0) {
return (
<PageContainer>
<NavBar>
<PageTitle title="Playground" />
</NavBar>
<PageBody>
<MainCenteredContainer className="max-w-2xl">
<InfoPanel
title="Create your first agent"
icon={CpuChipIcon}
iconClassName="text-indigo-500"
panelClassName="max-w-2xl"
accessory={
<LinkButton
to={docsPath("ai-chat/overview")}
variant="docs/small"
LeadingIcon={BookOpenIcon}
>
Agent docs
</LinkButton>
}
>
<Paragraph spacing variant="small">
The Playground lets you test your AI agents with an interactive chat interface,
realtime streaming, and conversation history.
</Paragraph>
<Paragraph spacing variant="small">
Define a chat agent using{" "}
<InlineCode variant="small">chat.agent()</InlineCode>:
</Paragraph>
<CodeBlock
code={`import { chat } from "@trigger.dev/sdk/ai";
import { streamText } from "ai";
import { openai } from "@ai-sdk/openai";
export const myAgent = chat.agent({
id: "my-agent",
run: async ({ messages, signal }) => {
return streamText({
model: openai("gpt-4o"),
messages,
abortSignal: signal,
});
},
});`}
showLineNumbers={false}
showOpenInModal={false}
/>
<Paragraph variant="small" className="mt-2">
Deploy your project and your agents will appear here ready to test.
</Paragraph>
</InfoPanel>
</MainCenteredContainer>
</PageBody>
</PageContainer>
);
}
return (
<PageContainer>
<NavBar>
<PageTitle title="Playground" />
</NavBar>
<PageBody scrollable={false}>
{selectedAgent ? (
<Outlet />
) : (
<MainCenteredContainer>
<div className="flex flex-col items-center gap-4 py-20">
<CpuChipIcon className="size-10 text-indigo-500/50" />
<Header2 className="text-text-dimmed">Select an agent</Header2>
<Paragraph variant="small" className="mb-2 max-w-md text-center text-text-dimmed">
Choose an agent to start a conversation.
</Paragraph>
<Select
value={selectedAgent}
setValue={(slug) => {
if (slug && typeof slug === "string") {
navigate(v3PlaygroundAgentPath(organization, project, environment, slug));
}
}}
icon={<CpuChipIcon className="size-4 text-indigo-500" />}
text={(val) => val || undefined}
placeholder="Select an agent..."
variant="tertiary/small"
items={agents}
filter={(item, search) =>
item.slug.toLowerCase().includes(search.toLowerCase())
}
>
{(matches) =>
matches.map((agent) => (
<SelectItem key={agent.slug} value={agent.slug}>
<div className="flex items-center gap-2">
<CpuChipIcon className="size-3.5 text-indigo-500" />
<span>{agent.slug}</span>
</div>
</SelectItem>
))
}
</Select>
</div>
</MainCenteredContainer>
)}
</PageBody>
</PageContainer>
);
}
@@ -0,0 +1,539 @@
import { ArrowsRightLeftIcon, BookOpenIcon, XCircleIcon } from "@heroicons/react/24/solid";
import { type MetaFunction } from "@remix-run/react";
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { typedjson, useTypedLoaderData } from "remix-typedjson";
import { z } from "zod";
import { CodeBlock } from "~/components/code/CodeBlock";
import { PageBody } from "~/components/layout/AppLayout";
import { Button, LinkButton } from "~/components/primitives/Buttons";
import { Dialog, DialogTrigger } from "~/components/primitives/Dialog";
import { CopyableText } from "~/components/primitives/CopyableText";
import { DateTime } from "~/components/primitives/DateTime";
import { Header2 } from "~/components/primitives/Headers";
import { NavBar, PageAccessories, PageTitle } from "~/components/primitives/PageHeader";
import SegmentedControl from "~/components/primitives/SegmentedControl";
import { Paragraph } from "~/components/primitives/Paragraph";
import * as Property from "~/components/primitives/PropertyTable";
import {
ResizableHandle,
ResizablePanel,
ResizablePanelGroup,
} from "~/components/primitives/Resizable";
import { TabButton, TabContainer } from "~/components/primitives/Tabs";
import { TextLink } from "~/components/primitives/TextLink";
import { SimpleTooltip } from "~/components/primitives/Tooltip";
import { AgentView } from "~/components/runs/v3/agent/AgentView";
import { RealtimeStreamViewer } from "~/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route";
import { RunTag } from "~/components/runs/v3/RunTag";
import {
descriptionForTaskRunStatus,
TaskRunStatusCombo,
} from "~/components/runs/v3/TaskRunStatus";
import { CloseSessionDialog } from "~/components/sessions/v1/CloseSessionDialog";
import { SessionStatusCombo } from "~/components/sessions/v1/SessionStatus";
import { $replica } from "~/db.server";
import { useEnvironment } from "~/hooks/useEnvironment";
import { useOrganization } from "~/hooks/useOrganizations";
import { useProject } from "~/hooks/useProject";
import { useSearchParams } from "~/hooks/useSearchParam";
import { useHasAdminAccess } from "~/hooks/useUser";
import { redirectWithErrorMessage } from "~/models/message.server";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { SessionPresenter } from "~/presenters/v3/SessionPresenter.server";
import { type SessionStatus } from "~/services/sessionsRepository/sessionsRepository.server";
import { requireUserId } from "~/services/session.server";
import { cn } from "~/utils/cn";
import {
docsPath,
EnvironmentParamSchema,
v3RunPath,
v3RunsPath,
v3SessionsPath,
} from "~/utils/pathBuilder";
const ParamsSchema = EnvironmentParamSchema.extend({
sessionParam: z.string(),
});
export const meta: MetaFunction = () => {
return [{ title: `Session | Trigger.dev` }];
};
export const loader = async ({ request, params }: LoaderFunctionArgs) => {
const userId = await requireUserId(request);
const { projectParam, organizationSlug, envParam, sessionParam } = ParamsSchema.parse(params);
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
if (!project) {
return redirectWithErrorMessage("/", request, "Project not found");
}
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
if (!environment) {
throw new Error("Environment not found");
}
const presenter = new SessionPresenter($replica);
const session = await presenter.call({
userId,
environmentId: environment.id,
sessionParam,
});
if (!session) {
throw new Response("Session not found", { status: 404 });
}
return typedjson({ session });
};
export default function Page() {
const { session } = useTypedLoaderData<typeof loader>();
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const status: SessionStatus =
session.closedAt != null
? "CLOSED"
: session.expiresAt != null && new Date(session.expiresAt).getTime() < Date.now()
? "EXPIRED"
: "ACTIVE";
const displayId = session.externalId ?? session.friendlyId;
const sessionsPath = v3SessionsPath(organization, project, environment);
return (
<>
<NavBar>
<PageTitle
backButton={{ to: sessionsPath, text: "Sessions" }}
title={
<CopyableText
value={displayId}
variant="text-below"
className="-ml-[0.4375rem] h-6 px-1.5 font-mono text-xs hover:text-text-bright"
/>
}
/>
<PageAccessories>
<LinkButton
variant={"docs/small"}
LeadingIcon={BookOpenIcon}
to={docsPath("/ai-chat/overview")}
>
Sessions docs
</LinkButton>
{status === "ACTIVE" && (
<Dialog key={`close-${session.friendlyId}`}>
<DialogTrigger asChild>
<Button variant="danger/small" LeadingIcon={XCircleIcon}>
Close session
</Button>
</DialogTrigger>
<CloseSessionDialog
sessionParam={session.friendlyId}
environmentId={environment.id}
redirectPath={`${sessionsPath}/${session.friendlyId}`}
/>
</Dialog>
)}
</PageAccessories>
</NavBar>
<PageBody scrollable={false}>
<ResizablePanelGroup orientation="horizontal" className="max-h-full">
<ResizablePanel id="session-conversation" min={"300px"}>
<ConversationPane session={session} />
</ResizablePanel>
<ResizableHandle id="session-handle" />
<ResizablePanel
id="session-inspector"
min="380px"
default="420px"
className="overflow-hidden"
>
<InspectorPane session={session} status={status} />
</ResizablePanel>
</ResizablePanelGroup>
</PageBody>
</>
);
}
type LoadedSession = ReturnType<typeof useTypedLoaderData<typeof loader>>["session"];
function ConversationPane({ session }: { session: LoadedSession }) {
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const { value, replace } = useSearchParams();
const isRaw = value("raw") === "1";
const stream: "out" | "in" = value("stream") === "in" ? "in" : "out";
const sessionId = session.agentView.sessionId;
const encodedSession = encodeURIComponent(sessionId);
const sessionResourceBase = `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/sessions/${encodedSession}/realtime/v1`;
return (
<div className="grid h-full max-h-full grid-rows-[2.5rem_1fr] overflow-hidden bg-background-bright">
<div className="flex items-center justify-between gap-2 overflow-x-hidden border-b border-grid-bright px-3">
<div className="flex items-center gap-2 overflow-x-hidden">
<ArrowsRightLeftIcon className="size-4 text-teal-500" />
<Header2 className={cn("overflow-x-hidden text-text-bright")}>
<span className="truncate">Conversation</span>
</Header2>
</div>
<SegmentedControl
name="conversation-view"
value={isRaw ? "raw" : "rendered"}
variant="secondary/small"
options={[
{ label: "Rendered", value: "rendered" },
{ label: "Raw", value: "raw" },
]}
onChange={(v) => replace({ raw: v === "raw" ? "1" : undefined })}
/>
</div>
{isRaw ? (
<div className="overflow-hidden">
<RealtimeStreamViewer
key={stream}
resourcePath={`${sessionResourceBase}/${stream}`}
displayName={`.${stream}`}
headerLeft={
<TabContainer>
<TabButton
isActive={stream === "out"}
layoutId="conversation-stream"
onClick={() => replace({ stream: undefined })}
>
Output
</TabButton>
<TabButton
isActive={stream === "in"}
layoutId="conversation-stream"
onClick={() => replace({ stream: "in" })}
>
Input
</TabButton>
</TabContainer>
}
/>
</div>
) : (
<div className="min-w-0 overflow-x-hidden overflow-y-auto px-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
<AgentView agentView={session.agentView} />
</div>
)}
</div>
);
}
function InspectorPane({
session,
status,
}: {
session: LoadedSession;
status: SessionStatus;
}) {
const { value, replace } = useSearchParams();
const tab = value("tab") ?? "overview";
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const displayId = session.externalId ?? session.friendlyId;
const allRunsPath = v3RunsPath(organization, project, environment, {
tags: [`chat:${displayId}`],
});
return (
<div className="grid h-full max-h-full grid-rows-[2.5rem_2rem_1fr] overflow-hidden bg-background-bright">
<div className="flex items-center justify-between gap-2 overflow-x-hidden px-3">
<div className="flex items-center gap-2 overflow-x-hidden">
<SessionStatusCombo status={status} />
<span className="truncate font-mono text-xs text-text-dimmed">
{session.friendlyId}
</span>
</div>
</div>
<div className="h-fit overflow-x-auto px-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
<TabContainer>
<TabButton
isActive={tab === "overview"}
layoutId="session-inspector"
onClick={() => replace({ tab: "overview" })}
shortcut={{ key: "o" }}
>
Overview
</TabButton>
<TabButton
isActive={tab === "runs"}
layoutId="session-inspector"
onClick={() => replace({ tab: "runs" })}
shortcut={{ key: "r" }}
>
Runs
</TabButton>
<TabButton
isActive={tab === "metadata"}
layoutId="session-inspector"
onClick={() => replace({ tab: "metadata" })}
shortcut={{ key: "m" }}
>
Metadata
</TabButton>
</TabContainer>
</div>
<div className="overflow-y-auto px-3 py-3 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
{tab === "overview" ? (
<OverviewTab session={session} status={status} />
) : tab === "runs" ? (
<RunsTab session={session} allRunsPath={allRunsPath} />
) : (
<MetadataTab session={session} />
)}
</div>
</div>
);
}
function OverviewTab({
session,
status,
}: {
session: LoadedSession;
status: SessionStatus;
}) {
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const isAdmin = useHasAdminAccess();
return (
<div className="flex flex-col gap-4">
<Property.Table>
<Property.Item>
<Property.Label>Status</Property.Label>
<Property.Value>
<SessionStatusCombo status={status} />
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Friendly ID</Property.Label>
<Property.Value>
<CopyableText value={session.friendlyId} className="font-mono text-xs" />
</Property.Value>
</Property.Item>
{session.externalId ? (
<Property.Item>
<Property.Label>External ID</Property.Label>
<Property.Value>
<CopyableText value={session.externalId} className="font-mono text-xs" />
</Property.Value>
</Property.Item>
) : null}
<Property.Item>
<Property.Label>Type</Property.Label>
<Property.Value>
<span className="font-mono text-xs">{session.type}</span>
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Task</Property.Label>
<Property.Value>
<span className="font-mono text-xs">{session.taskIdentifier}</span>
</Property.Value>
</Property.Item>
{session.currentRun ? (
<Property.Item>
<Property.Label>Current run</Property.Label>
<Property.Value>
<TextLink
to={v3RunPath(organization, project, environment, {
friendlyId: session.currentRun.friendlyId,
})}
>
<span className="flex items-center gap-2">
<span className="font-mono text-xs">{session.currentRun.friendlyId}</span>
<SimpleTooltip
button={<TaskRunStatusCombo status={session.currentRun.status} />}
content={descriptionForTaskRunStatus(session.currentRun.status)}
disableHoverableContent
/>
</span>
</TextLink>
</Property.Value>
</Property.Item>
) : null}
<Property.Item>
<Property.Label>Tags</Property.Label>
<Property.Value>
{session.tags.length > 0 ? (
<div className="flex flex-wrap gap-1">
{session.tags.map((tag) => (
<RunTag key={tag} tag={tag} />
))}
</div>
) : (
<span className="text-text-dimmed"></span>
)}
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Created</Property.Label>
<Property.Value>
<DateTime date={session.createdAt} />
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Updated</Property.Label>
<Property.Value>
<DateTime date={session.updatedAt} />
</Property.Value>
</Property.Item>
{session.expiresAt ? (
<Property.Item>
<Property.Label>
{new Date(session.expiresAt).getTime() < Date.now() ? "Expired" : "Expires"}
</Property.Label>
<Property.Value>
<DateTime date={session.expiresAt} />
</Property.Value>
</Property.Item>
) : null}
{session.closedAt ? (
<Property.Item>
<Property.Label>Closed</Property.Label>
<Property.Value>
<DateTime date={session.closedAt} />
</Property.Value>
</Property.Item>
) : null}
{session.closedReason ? (
<Property.Item>
<Property.Label>Close reason</Property.Label>
<Property.Value>
<span className="text-xs">{session.closedReason}</span>
</Property.Value>
</Property.Item>
) : null}
</Property.Table>
<CodeBlock
code={JSON.stringify(session.triggerConfig, null, 2)}
language="json"
rowTitle="Trigger config"
maxLines={20}
showLineNumbers={false}
showTextWrapping
/>
{isAdmin && (
<div className="border-t border-yellow-500/50 pt-2">
<Paragraph spacing variant="small" className="text-yellow-500">
Admin only
</Paragraph>
<Property.Table>
<Property.Item>
<Property.Label>Session ID</Property.Label>
<Property.Value>
<span className="font-mono text-xs">{session.id}</span>
</Property.Value>
</Property.Item>
<Property.Item>
<Property.Label>Stream basin</Property.Label>
<Property.Value>
<span className="font-mono text-xs">
{session.streamBasinName ?? "(global)"}
</span>
</Property.Value>
</Property.Item>
</Property.Table>
</div>
)}
</div>
);
}
function MetadataTab({ session }: { session: LoadedSession }) {
if (session.metadata == null) {
return (
<Paragraph variant="small/dimmed">No metadata.</Paragraph>
);
}
const json = JSON.stringify(session.metadata, null, 2);
return (
<CodeBlock code={json} language="json" showLineNumbers={false} showTextWrapping />
);
}
function RunsTab({
session,
allRunsPath,
}: {
session: LoadedSession;
allRunsPath: string;
}) {
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
if (session.runs.length === 0) {
return <Paragraph variant="small/dimmed">No runs yet.</Paragraph>;
}
return (
<div className="flex flex-col gap-3">
<Property.Table>
{session.runs.map((entry) => {
const runPath = entry.run
? v3RunPath(organization, project, environment, {
friendlyId: entry.run.friendlyId,
})
: undefined;
return (
<Property.Item key={entry.id}>
<Property.Label>
<div className="flex flex-col gap-0.5">
<span className="capitalize">{entry.reason}</span>
<span className="text-xs text-text-dimmed">
<DateTime date={entry.triggeredAt} />
</span>
</div>
</Property.Label>
<Property.Value>
{entry.run && runPath ? (
<SimpleTooltip
button={
<TextLink
to={runPath}
className="group flex flex-wrap items-center gap-x-2 gap-y-0"
>
<CopyableText
value={entry.run.friendlyId}
copyValue={entry.run.friendlyId}
asChild
/>
<TaskRunStatusCombo status={entry.run.status} />
</TextLink>
}
content={`Jump to run`}
disableHoverableContent
/>
) : (
<span className="text-text-dimmed"></span>
)}
</Property.Value>
</Property.Item>
);
})}
</Property.Table>
<div className="flex justify-end">
<LinkButton variant="tertiary/small" to={allRunsPath}>
View all runs
</LinkButton>
</div>
</div>
);
}
@@ -9,18 +9,34 @@ import { docsPath } from "~/utils/pathBuilder";
export function SchemaTabContent({
schema,
inferredSchema,
title = "Payload schema",
description,
showDocsLink = true,
}: {
schema?: unknown;
inferredSchema?: unknown;
title?: string;
description?: string;
showDocsLink?: boolean;
}) {
if (schema) {
return (
<div className="space-y-2">
<Header3 className="text-text-bright">Payload schema</Header3>
<Paragraph variant="extra-small" className="text-text-dimmed">
JSON Schema defined by this task via{" "}
<TextLink to={docsPath("tasks/schemaTask")}>schemaTask</TextLink>.
</Paragraph>
<Header3 className="text-text-bright">{title}</Header3>
{showDocsLink ? (
<Paragraph variant="extra-small" className="text-text-dimmed">
{description ?? (
<>
JSON Schema defined by this task via{" "}
<TextLink to={docsPath("tasks/schemaTask")}>schemaTask</TextLink>.
</>
)}
</Paragraph>
) : description ? (
<Paragraph variant="extra-small" className="text-text-dimmed">
{description}
</Paragraph>
) : null}
<CodeBlock
code={JSON.stringify(schema, null, 2)}
language="json"
@@ -0,0 +1,307 @@
import { json } from "@remix-run/server-runtime";
import { type ActionFunctionArgs } from "@remix-run/server-runtime";
import { z } from "zod";
import type { Prisma } from "@trigger.dev/database";
import { SessionId } from "@trigger.dev/core/v3/isomorphic";
import { prisma } from "~/db.server";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { requireUserId } from "~/services/session.server";
import { EnvironmentParamSchema } from "~/utils/pathBuilder";
import { mintSessionToken } from "~/services/realtime/mintSessionToken.server";
import { ensureRunForSession } from "~/services/realtime/sessionRunManager.server";
const PlaygroundAction = z.object({
intent: z.enum(["create", "start", "save", "delete"]),
agentSlug: z.string(),
// For create
conversationId: z.string().optional(),
// For start (replaces "trigger" — atomically creates the Session and
// triggers its first run, returns a session-scoped PAT)
chatId: z.string().optional(),
payload: z.string().optional(),
clientData: z.string().optional(),
tags: z.string().optional(),
machine: z.string().optional(),
maxAttempts: z.string().optional(),
maxDuration: z.string().optional(),
version: z.string().optional(),
region: z.string().optional(),
// For save
messages: z.string().optional(),
lastEventId: z.string().optional(),
// For delete
deleteConversationId: z.string().optional(),
});
export const action = async ({ request, params }: ActionFunctionArgs) => {
const userId = await requireUserId(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
if (!project) {
return json({ error: "Project not found" }, { status: 404 });
}
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
if (!environment) {
return json({ error: "Environment not found" }, { status: 404 });
}
const formData = await request.formData();
const parsed = PlaygroundAction.safeParse(Object.fromEntries(formData));
if (!parsed.success) {
return json({ error: "Invalid request", details: parsed.error.issues }, { status: 400 });
}
const { intent } = parsed.data;
switch (intent) {
case "create": {
const { agentSlug } = parsed.data;
const chatId = crypto.randomUUID();
const conversation = await prisma.playgroundConversation.create({
data: {
chatId,
agentSlug,
projectId: project.id,
runtimeEnvironmentId: environment.id,
userId,
},
});
return json({
conversationId: conversation.id,
chatId,
});
}
case "start": {
const {
agentSlug,
chatId,
payload: payloadStr,
clientData,
tags: tagsStr,
machine,
maxAttempts,
maxDuration,
version,
region,
} = parsed.data;
if (!chatId) {
return json({ error: "chatId is required" }, { status: 400 });
}
// Parse the optional initial payload — used as the basePayload
// for the first run trigger. After session create, the agent
// reads subsequent messages from `.in/append` so the payload
// here is just the bootstrap.
let payload: Record<string, any> = {};
try {
payload = payloadStr ? (JSON.parse(payloadStr) as Record<string, any>) : {};
} catch {
return json({ error: "Invalid payload JSON" }, { status: 400 });
}
let parsedClientData: unknown;
try {
parsedClientData = clientData ? JSON.parse(clientData) : undefined;
} catch {
/* invalid JSON — fall through with undefined */
}
const tags = [
`chat:${chatId}`,
"playground:true",
...(tagsStr ? tagsStr.split(",").map((t) => t.trim()).filter(Boolean) : []),
].slice(0, 5);
const triggerConfig = {
basePayload: {
// The first run boots before the user's first message lands on
// `.in/append`, so it sees `messages: []` and `trigger: "preload"`.
// Mirrors the defaults in `chat.createStartSessionAction` —
// chat.agent's runtime reads `payload.messages.length` so the
// field must be an array, not undefined.
messages: [],
trigger: "preload",
...payload,
chatId,
...(parsedClientData ? { metadata: parsedClientData } : {}),
},
...(machine ? { machine } : {}),
tags,
...(maxAttempts ? { maxAttempts: parseInt(maxAttempts, 10) } : {}),
...(maxDuration ? { maxDuration: parseInt(maxDuration, 10) } : {}),
...(version ? { lockToVersion: version } : {}),
...(region ? { region } : {}),
};
// Atomic: upsert the Session, then trigger the first run via
// the optimistic-claim path. The transport's `accessToken`
// callback hits this endpoint on initial start AND on 401 — the
// upsert + ensureRunForSession combo is idempotent so repeat
// calls converge to the same session and (if alive) reuse the
// existing run.
const { id: sessionId, friendlyId } = SessionId.generate();
const session = await prisma.session.upsert({
where: {
runtimeEnvironmentId_externalId: {
runtimeEnvironmentId: environment.id,
externalId: chatId,
},
},
create: {
id: sessionId,
friendlyId,
externalId: chatId,
type: "chat.agent",
taskIdentifier: agentSlug,
triggerConfig: triggerConfig as unknown as Prisma.InputJsonValue,
tags: ["playground"],
projectId: project.id,
runtimeEnvironmentId: environment.id,
environmentType: environment.type,
organizationId: project.organizationId,
// Stamp the org's S2 basin so realtime reads on this
// session's `.in/.out` channels resolve without joining
// Organization. Null until per-org basins are provisioned.
streamBasinName: environment.organization.streamBasinName,
},
update: {
// Refresh trigger config in case agent version / params changed
triggerConfig: triggerConfig as unknown as Prisma.InputJsonValue,
},
});
const ensureResult = await ensureRunForSession({
session,
environment,
reason: "initial",
});
const run = await prisma.taskRun.findFirst({
where: { id: ensureResult.runId },
select: { friendlyId: true },
});
if (!run) {
return json({ error: "Triggered run not found" }, { status: 500 });
}
// Title: prefer the user message text on first start, else a
// generic placeholder. The conversation row is the playground's
// own surface — separate from the Session row that drives the
// trigger.
const firstMessage = payload?.messages?.[0];
const firstText =
firstMessage?.parts?.find((p: any) => p.type === "text")?.text ?? "New conversation";
const title = firstText.length > 60 ? firstText.slice(0, 60) + "..." : firstText;
const conversation = await prisma.playgroundConversation.upsert({
where: {
chatId_runtimeEnvironmentId: {
chatId,
runtimeEnvironmentId: environment.id,
},
},
create: {
chatId,
title,
agentSlug,
runId: ensureResult.runId,
clientData: parsedClientData as any,
projectId: project.id,
runtimeEnvironmentId: environment.id,
userId,
},
update: {
runId: ensureResult.runId,
clientData: parsedClientData as any,
title,
},
});
const publicAccessToken = await mintSessionToken(environment, chatId);
return json({
runId: run.friendlyId,
publicAccessToken,
conversationId: conversation.id,
});
}
case "save": {
const { chatId, messages: messagesStr, lastEventId } = parsed.data;
if (!chatId) {
return json({ error: "chatId is required" }, { status: 400 });
}
let messagesData: unknown;
try {
messagesData = messagesStr ? JSON.parse(messagesStr) : undefined;
} catch {
return json({ error: "Invalid messages JSON" }, { status: 400 });
}
// Extract title from the first user message if the conversation still has the default title.
// This handles the case where a preloaded conversation gets its first real message
// via the input stream (bypassing the trigger action that normally sets the title).
let titleUpdate: { title: string } | undefined;
if (messagesData && Array.isArray(messagesData)) {
const existing = await prisma.playgroundConversation.findFirst({
where: { chatId, runtimeEnvironmentId: environment.id, userId },
select: { title: true },
});
if (existing?.title === "New conversation") {
const firstUserMsg = messagesData.find(
(m: any) => m.role === "user"
) as Record<string, any> | undefined;
const firstText =
firstUserMsg?.parts?.find((p: any) => p.type === "text")?.text ??
firstUserMsg?.content;
if (firstText && typeof firstText === "string") {
titleUpdate = {
title: firstText.length > 60 ? firstText.slice(0, 60) + "..." : firstText,
};
}
}
}
await prisma.playgroundConversation.updateMany({
where: {
chatId,
runtimeEnvironmentId: environment.id,
userId,
},
data: {
...(messagesData ? { messages: messagesData as any } : {}),
...(lastEventId ? { lastEventId } : {}),
...titleUpdate,
},
});
return json({ ok: true });
}
case "delete": {
const { deleteConversationId } = parsed.data;
if (!deleteConversationId) {
return json({ error: "deleteConversationId is required" }, { status: 400 });
}
await prisma.playgroundConversation.deleteMany({
where: {
id: deleteConversationId,
runtimeEnvironmentId: environment.id,
userId,
},
});
return json({ ok: true });
}
}
};
@@ -0,0 +1,163 @@
import { json, type ActionFunctionArgs } from "@remix-run/server-runtime";
import { tryCatch } from "@trigger.dev/core/utils";
import { nanoid } from "nanoid";
import { z } from "zod";
import { $replica } from "~/db.server";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { logger } from "~/services/logger.server";
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
import { ensureRunForSession } from "~/services/realtime/sessionRunManager.server";
import {
canonicalSessionAddressingKey,
resolveSessionByIdOrExternalId,
} from "~/services/realtime/sessions.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import { drainSessionStreamWaitpoints } from "~/services/sessionStreamWaitpointCache.server";
import { requireUserId } from "~/services/session.server";
import { EnvironmentParamSchema } from "~/utils/pathBuilder";
import { engine } from "~/v3/runEngine.server";
import { ServiceValidationError } from "~/v3/services/common.server";
const ParamsSchema = z.object({
session: z.string(),
io: z.enum(["out", "in"]),
});
// S2 record body cap. Mirrors the public /realtime/v1/sessions/:s/:io/append
// route — keep it well under S2's 1 MiB per-record limit so JSON wrapping,
// string escaping, and any future per-record headers stay safe.
const MAX_APPEND_BODY_BYTES = 1024 * 512;
// POST: Append a single record to a Session channel from the dashboard
// playground. Mirrors the public `POST /realtime/v1/sessions/:session/:io/append`
// but authenticates via the dashboard session cookie instead of a
// session-scoped JWT.
export async function action({ request, params }: ActionFunctionArgs) {
const userId = await requireUserId(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const { session: sessionParam, io } = ParamsSchema.parse(params);
const project = await findProjectBySlug(organizationSlug, projectParam, userId);
if (!project) {
return json({ ok: false, error: "Project not found" }, { status: 404 });
}
const environment = await findEnvironmentBySlug(project.id, envParam, userId);
if (!environment) {
return json({ ok: false, error: "Environment not found" }, { status: 404 });
}
const contentLength = request.headers.get("content-length");
const contentLengthNum = contentLength ? parseInt(contentLength, 10) : NaN;
if (Number.isNaN(contentLengthNum) || contentLengthNum > MAX_APPEND_BODY_BYTES) {
return json({ ok: false, error: "Request body too large" }, { status: 413 });
}
const session = await resolveSessionByIdOrExternalId(
$replica,
environment.id,
sessionParam
);
if (!session) {
return json({ ok: false, error: "Session not found" }, { status: 404 });
}
if (session.closedAt) {
return json(
{ ok: false, error: "Cannot append to a closed session" },
{ status: 400 }
);
}
if (session.expiresAt && session.expiresAt.getTime() < Date.now()) {
return json(
{ ok: false, error: "Cannot append to an expired session" },
{ status: 400 }
);
}
const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session });
if (!(realtimeStream instanceof S2RealtimeStreams)) {
return json(
{ ok: false, error: "Session channels require the S2 realtime backend" },
{ status: 501 }
);
}
// Probe + ensure a live run before appending (mirrors public route).
// Best-effort: failure here doesn't block the append — the record is
// durable; the next append retries the ensure.
const [ensureError] = await tryCatch(
ensureRunForSession({
session,
environment,
reason: "continuation",
})
);
if (ensureError) {
logger.error("Failed to ensureRunForSession on playground .in/append", {
sessionId: session.id,
externalId: session.externalId,
error: ensureError,
});
}
const addressingKey = canonicalSessionAddressingKey(session, sessionParam);
const part = await request.text();
const partId = request.headers.get("X-Part-Id") ?? nanoid(7);
const [appendError] = await tryCatch(
realtimeStream.appendPartToSessionStream(part, partId, addressingKey, io)
);
if (appendError) {
if (appendError instanceof ServiceValidationError) {
return json(
{ ok: false, error: appendError.message },
{ status: appendError.status ?? 422 }
);
}
return json({ ok: false, error: appendError.message }, { status: 500 });
}
// Drain any waitpoints registered for this channel — same as the
// public append. Best-effort; failure doesn't fail the append.
const [drainError, waitpointIds] = await tryCatch(
drainSessionStreamWaitpoints(addressingKey, io)
);
if (drainError) {
logger.error("Failed to drain session stream waitpoints (playground)", {
addressingKey,
io,
error: drainError,
});
} else if (waitpointIds && waitpointIds.length > 0) {
await Promise.all(
waitpointIds.map(async (waitpointId) => {
const [completeError] = await tryCatch(
engine.completeWaitpoint({
id: waitpointId,
output: {
value: part,
type: "application/json",
isError: false,
},
})
);
if (completeError) {
logger.error("Failed to complete session stream waitpoint (playground)", {
addressingKey,
io,
waitpointId,
error: completeError,
});
}
})
);
}
return json({ ok: true }, { status: 200 });
}
@@ -0,0 +1,91 @@
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { z } from "zod";
import { $replica } from "~/db.server";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
import {
canonicalSessionAddressingKey,
resolveSessionByIdOrExternalId,
} from "~/services/realtime/sessions.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import { requireUserId } from "~/services/session.server";
import { EnvironmentParamSchema } from "~/utils/pathBuilder";
const ParamsSchema = z.object({
session: z.string(),
io: z.enum(["out", "in"]),
});
// HEAD/GET: SSE subscribe to a Session channel from the dashboard
// playground. Mirrors the public `GET /realtime/v1/sessions/:session/:io`
// route but authenticates via the dashboard session cookie instead of a
// session-scoped JWT — the playground transport never holds a PAT.
//
// `:session` accepts either the `session_*` friendlyId or the externalId
// the playground assigned (`chatId`). Resolution is environment-scoped
// so users can't subscribe to sessions from other envs.
export async function loader({ request, params }: LoaderFunctionArgs) {
const userId = await requireUserId(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const { session: sessionParam, io } = ParamsSchema.parse(params);
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 session = await resolveSessionByIdOrExternalId(
$replica,
environment.id,
sessionParam
);
if (!session) {
return new Response("Session not found", { status: 404 });
}
const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session });
if (!(realtimeStream instanceof S2RealtimeStreams)) {
return new Response("Session channels require the S2 realtime backend", {
status: 501,
});
}
if (request.method === "HEAD") {
// No last-chunk-index on the S2 backend (clients resume via
// Last-Event-ID on the SSE stream directly). Return 200 with a
// zero index for compatibility with the run-stream shape.
return new Response(null, {
status: 200,
headers: { "X-Last-Chunk-Index": "0" },
});
}
const lastEventId = request.headers.get("Last-Event-ID") || undefined;
const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds");
let timeoutInSeconds: number | undefined;
if (timeoutInSecondsRaw !== null) {
timeoutInSeconds = Number(timeoutInSecondsRaw);
if (!Number.isInteger(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) {
return new Response("Invalid timeout", { status: 400 });
}
}
const addressingKey = canonicalSessionAddressingKey(session, sessionParam);
return realtimeStream.streamResponseFromSessionStream(
request,
addressingKey,
io,
getRequestAbortSignal(),
{ lastEventId, timeoutInSeconds }
);
}
@@ -0,0 +1,118 @@
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { z } from "zod";
import { $replica } from "~/db.server";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
import {
canonicalSessionAddressingKey,
resolveSessionByIdOrExternalId,
} from "~/services/realtime/sessions.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import { requireUserId } from "~/services/session.server";
import { EnvironmentParamSchema } from "~/utils/pathBuilder";
const ParamsSchema = z.object({
runParam: z.string(),
sessionId: z.string(),
io: z.enum(["out", "in"]),
});
// GET: SSE stream subscription for a backing Session's `.out` / `.in`
// channel. Dashboard-auth counterpart to the public API's
// `/realtime/v1/sessions/:sessionId/:io` endpoint. Used by the Agent tab
// in the span inspector to observe assistant chunks (`.out`) and
// user-side ChatInputChunk payloads (`.in`) for a chat.agent run.
//
// The `:sessionId` segment accepts either the `session_*` friendlyId or
// the externalId the transport registered for the chat (typically the
// browser's `chatId`). Runs pre-dating the Sessions migration that have
// `chatId` but no `sessionId` in the payload take the externalId path.
//
// Authenticated by the dashboard session — the user must have access to
// the project, environment, and run. The run binds this resource
// hierarchy; the session identity is verified against the environment.
export async function loader({ request, params }: LoaderFunctionArgs) {
const userId = await requireUserId(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const { runParam, sessionId, io } = ParamsSchema.parse(params);
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 });
}
// Verify the run lives in this environment — keeps callers from
// subscribing to arbitrary sessions via `/runs/$runParam/...`.
const run = await $replica.taskRun.findFirst({
where: {
friendlyId: runParam,
runtimeEnvironmentId: environment.id,
},
select: { id: true, friendlyId: true },
});
if (!run) {
return new Response("Run not found", { status: 404 });
}
const session = await resolveSessionByIdOrExternalId(
$replica,
environment.id,
sessionId
);
if (!session) {
return new Response("Session not found", { status: 404 });
}
// Enforce run ↔ session linkage. Without this, knowledge of a runId in
// this environment is enough to subscribe to any session in the same
// environment — defeats the point of scoping subscriptions through the
// run route. SessionRun.runId is indexed (@unique), so this is cheap.
const linkedSessionRun = await $replica.sessionRun.findFirst({
where: { runId: run.id, sessionId: session.id },
select: { id: true },
});
if (!linkedSessionRun) {
return new Response("Session not found for run", { status: 404 });
}
const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session });
if (!(realtimeStream instanceof S2RealtimeStreams)) {
return new Response("Session channels require the S2 realtime backend", {
status: 501,
});
}
const lastEventId = request.headers.get("Last-Event-ID") || undefined;
const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds");
let timeoutInSeconds: number | undefined;
if (timeoutInSecondsRaw !== null) {
timeoutInSeconds = Number(timeoutInSecondsRaw);
if (!Number.isInteger(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) {
return new Response("Invalid timeout", { status: 400 });
}
}
// The agent writes via the canonical addressing key (externalId if
// set, else friendlyId). Subscribe with the same key so the read
// hits the same S2 stream the agent is writing into.
const addressingKey = canonicalSessionAddressingKey(session, sessionId);
return realtimeStream.streamResponseFromSessionStream(
request,
addressingKey,
io,
getRequestAbortSignal(),
{ lastEventId, timeoutInSeconds }
);
}
@@ -0,0 +1,91 @@
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { z } from "zod";
import { $replica } from "~/db.server";
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.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,
streamBasinName: 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");
let timeoutInSeconds: number | undefined;
if (timeoutInSecondsRaw !== null) {
timeoutInSeconds = Number(timeoutInSecondsRaw);
if (!Number.isInteger(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) {
return new Response("Invalid timeout", { status: 400 });
}
}
const realtimeStream = getRealtimeStreamInstance(environment, run.realtimeStreamsVersion, {
run,
});
// `request.signal` is severed by Remix's Request.clone() + Node undici GC bug
// (see apps/webapp/CLAUDE.md). Use the Express res.on('close')-backed signal so
// the upstream stream fetch actually aborts when the user closes the tab.
return realtimeStream.streamResponse(
request,
run.friendlyId,
streamId,
getRequestAbortSignal(),
{
lastEventId,
timeoutInSeconds,
}
);
}
@@ -0,0 +1,92 @@
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { z } from "zod";
import { $replica } from "~/db.server";
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.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,
streamBasinName: 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");
let timeoutInSeconds: number | undefined;
if (timeoutInSecondsRaw !== null) {
timeoutInSeconds = Number(timeoutInSecondsRaw);
if (!Number.isInteger(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) {
return new Response("Invalid timeout", { status: 400 });
}
}
const realtimeStream = getRealtimeStreamInstance(environment, run.realtimeStreamsVersion, {
run,
});
// `request.signal` is severed by Remix's Request.clone() + Node undici GC bug
// (see apps/webapp/CLAUDE.md). Use the Express res.on('close')-backed signal.
return realtimeStream.streamResponse(
request,
run.friendlyId,
`$trigger.input:${streamId}`,
getRequestAbortSignal(),
{
lastEventId,
timeoutInSeconds,
}
);
}
@@ -53,6 +53,7 @@ import {
TableRow,
} from "~/components/primitives/Table";
import { TabButton, TabContainer } from "~/components/primitives/Tabs";
import { SessionStatusCombo } from "~/components/sessions/v1/SessionStatus";
import { TextLink } from "~/components/primitives/TextLink";
import { InfoIconTooltip, SimpleTooltip } from "~/components/primitives/Tooltip";
import { RunTimeline, RunTimelineEvent, SpanTimeline } from "~/components/run/RunTimeline";
@@ -88,6 +89,7 @@ import { formatCurrencyAccurate } from "~/utils/numberFormatter";
import {
docsPath,
v3BatchPath,
v3SessionPath,
v3DeploymentVersionPath,
v3LogsPath,
v3RunDownloadLogsPath,
@@ -124,7 +126,26 @@ 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.`
);
}
// Reconstruct the discriminated union explicitly. Spreading
// `{ ...result }` 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 });
}
return typedjson({ type: "span" as const, span: result.span });
} catch (error) {
logger.error("Error loading span", {
projectParam,
@@ -618,6 +639,32 @@ function RunBody({
</Property.Value>
</Property.Item>
)}
{run.session && (
<Property.Item>
<Property.Label>Session</Property.Label>
<Property.Value>
<SimpleTooltip
button={
<TextLink
to={v3SessionPath(organization, project, environment, {
friendlyId: run.session.friendlyId,
})}
className="group flex flex-wrap items-center gap-x-2 gap-y-0"
>
<CopyableText
value={run.session.externalId ?? run.session.friendlyId}
copyValue={run.session.externalId ?? run.session.friendlyId}
asChild
/>
<SessionStatusCombo status={run.session.status} />
</TextLink>
}
content={`Jump to session (${run.session.reason})`}
disableHoverableContent
/>
</Property.Value>
</Property.Item>
)}
<Property.Item>
<Property.Label>
<div className="flex items-center justify-between">
@@ -101,17 +101,32 @@ export function RealtimeStreamViewer({
streamKey,
metadata,
displayName,
resourcePath: resourcePathOverride,
headerLabel,
headerLeft,
}: {
runId: string;
streamKey: string;
metadata: Record<string, unknown> | undefined;
runId?: string;
streamKey?: string;
metadata?: Record<string, unknown> | undefined;
displayName?: string;
/** Pre-built resource path. When provided, `runId`/`streamKey` are unused. */
resourcePath?: string;
/** Override the "Stream:" / "Input stream:" prefix in the header. */
headerLabel?: string;
/**
* Replaces the default "Stream: <name>" content next to the connection
* icon. Use to inline tabs or other navigation in place of a static
* label.
*/
headerLeft?: React.ReactNode;
}) {
const organization = useOrganization();
const project = useProject();
const environment = useEnvironment();
const resourcePath = `/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/runs/${runId}/streams/${streamKey}`;
const resourcePath =
resourcePathOverride ??
`/resources/orgs/${organization.slug}/projects/${project.slug}/env/${environment.slug}/runs/${runId}/streams/${streamKey}`;
const startIndex = typeof metadata?.startIndex === "number" ? metadata.startIndex : undefined;
const { chunks, error, isConnected } = useRealtimeStream(resourcePath, startIndex);
@@ -229,7 +244,7 @@ export function RealtimeStreamViewer({
{/* Header */}
<div className="border-b border-grid-bright bg-background-bright @container">
<div className="flex flex-wrap items-center justify-between gap-2 px-3 py-2.5 @[300px]:flex-nowrap">
<div className="flex min-w-0 items-center gap-1.5">
<div className="flex min-w-0 items-center gap-3">
<TooltipProvider>
<Tooltip>
<TooltipTrigger>
@@ -244,13 +259,17 @@ export function RealtimeStreamViewer({
</TooltipContent>
</Tooltip>
</TooltipProvider>
<Paragraph
variant="small/bright"
className="mb-0 flex min-w-0 items-center gap-1 truncate whitespace-nowrap"
>
<span>{displayName ? "Input stream:" : "Stream:"}</span>
<span className="truncate font-mono text-text-dimmed">{displayName ?? streamKey}</span>
</Paragraph>
{headerLeft ?? (
<Paragraph
variant="small/bright"
className="mb-0 flex min-w-0 items-center gap-1 truncate whitespace-nowrap"
>
<span>{headerLabel ?? (displayName ? "Input stream:" : "Stream:")}</span>
<span className="truncate font-mono text-text-dimmed">
{displayName ?? streamKey ?? ""}
</span>
</Paragraph>
)}
</div>
<div className="flex w-full flex-wrap items-center justify-between gap-3 @[300px]:w-auto @[300px]:flex-nowrap">
<Paragraph variant="small" className="mb-0 whitespace-nowrap">
@@ -0,0 +1,83 @@
import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
import { z } from "zod";
import { $replica } from "~/db.server";
import { findProjectBySlug } from "~/models/project.server";
import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
import { S2RealtimeStreams } from "~/services/realtime/s2realtimeStreams.server";
import {
canonicalSessionAddressingKey,
resolveSessionByIdOrExternalId,
} from "~/services/realtime/sessions.server";
import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
import { requireUserId } from "~/services/session.server";
import { EnvironmentParamSchema } from "~/utils/pathBuilder";
const ParamsSchema = z.object({
sessionParam: z.string(),
io: z.enum(["out", "in"]),
});
// GET: SSE stream subscription for a Session's `.out` / `.in` channel.
// Dashboard-auth counterpart to the public API's
// `/realtime/v1/sessions/:sessionId/:io`. Used by the Sessions detail
// view (and the run page's Agent tab) to observe assistant chunks
// (`.out`) and user-side ChatInputChunk payloads (`.in`).
//
// The `:sessionParam` segment accepts either the `session_*` friendlyId
// or the externalId the transport registered for the chat (typically the
// browser's `chatId`).
//
// Authenticated by the dashboard session — the user must have access to
// the project and environment. The session must live in that environment.
export async function loader({ request, params }: LoaderFunctionArgs) {
const userId = await requireUserId(request);
const { organizationSlug, projectParam, envParam } = EnvironmentParamSchema.parse(params);
const { sessionParam, io } = ParamsSchema.parse(params);
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 session = await resolveSessionByIdOrExternalId($replica, environment.id, sessionParam);
if (!session) {
return new Response("Session not found", { status: 404 });
}
const realtimeStream = getRealtimeStreamInstance(environment, "v2", { session });
if (!(realtimeStream instanceof S2RealtimeStreams)) {
return new Response("Session channels require the S2 realtime backend", {
status: 501,
});
}
const lastEventId = request.headers.get("Last-Event-ID") || undefined;
const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds");
let timeoutInSeconds: number | undefined;
if (timeoutInSecondsRaw !== null) {
timeoutInSeconds = Number(timeoutInSecondsRaw);
if (!Number.isInteger(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600) {
return new Response("Invalid timeout", { status: 400 });
}
}
// The agent writes via the canonical addressing key (externalId if
// set, else friendlyId). Subscribe with the same key so the read
// hits the same S2 stream the agent is writing into.
const addressingKey = canonicalSessionAddressingKey(session, sessionParam);
return realtimeStream.streamResponseFromSessionStream(
request,
addressingKey,
io,
getRequestAbortSignal(),
{ lastEventId, timeoutInSeconds }
);
}
@@ -21,6 +21,7 @@ const RequestSchema = z.object({
taskIdentifier: z.string().max(256),
payloadSchema: z.string().max(50_000).optional(),
currentPayload: z.string().max(50_000).optional(),
isAgent: z.enum(["true", "false"]).optional(),
});
export async function action({ request, params }: ActionFunctionArgs) {
@@ -64,16 +65,20 @@ export async function action({ request, params }: ActionFunctionArgs) {
);
}
const { prompt, taskIdentifier, payloadSchema, currentPayload } = submission.data;
const { prompt, taskIdentifier, payloadSchema, currentPayload, isAgent } = submission.data;
const agentMode = isAgent === "true";
logger.info("[AI payload] Generating payload", {
taskIdentifier,
hasPayloadSchema: !!payloadSchema,
hasCurrentPayload: !!currentPayload,
promptLength: prompt.length,
agentMode,
});
const systemPrompt = buildSystemPrompt(taskIdentifier, payloadSchema, currentPayload);
const systemPrompt = agentMode
? buildAgentClientDataPrompt(taskIdentifier, payloadSchema, currentPayload)
: buildSystemPrompt(taskIdentifier, payloadSchema, currentPayload);
const stream = new ReadableStream({
async start(controller) {
@@ -234,6 +239,60 @@ async function getTaskFromDeployment(environmentId: string, taskIdentifier: stri
return { fileId: task.fileId };
}
function buildAgentClientDataPrompt(
taskIdentifier: string,
payloadSchema?: string,
currentPayload?: string
): string {
let prompt = `You are a JSON generator for client data (metadata) of a Trigger.dev chat agent with id "${taskIdentifier}".
IMPORTANT: You are generating ONLY the client data object this is the metadata sent alongside each chat message. It is NOT the full task payload. Do NOT generate fields like "chatId", "messages", "trigger", or "idleTimeoutInSeconds" those are internal transport fields managed by the framework.
The client data typically contains user context like user IDs, preferences, configuration, or session info. Return ONLY valid JSON wrapped in a \`\`\`json code block.
Requirements:
- Generate realistic, meaningful example data
- All string values should be plausible (real-looking IDs, names, etc.)
- The JSON must be valid and parseable
- Keep it simple client data is usually a flat or shallow object`;
if (payloadSchema) {
prompt += `
The agent has the following JSON Schema for its client data:
\`\`\`json
${payloadSchema}
\`\`\`
Generate client data that strictly conforms to this schema.`;
} else {
prompt += `
No JSON Schema is available for this agent's client data. Use the getTaskSourceCode tool to look up the agent's source code file.
IMPORTANT instructions for reading the source code:
- The file may contain multiple task/agent definitions. Find the one with id "${taskIdentifier}".
- Look for \`withClientData({ schema: ... })\` or \`clientDataSchema\` to find the expected client data shape.
- If using \`chat.agent()\` or \`chat.customAgent()\`, the client data is accessed via \`clientData\` in hooks and \`payload.metadata\` in raw tasks.
- Look for how \`clientData\` or \`payload.metadata\` is accessed/destructured to infer the shape.
- Do NOT generate the full ChatTaskWirePayload (messages, chatId, trigger, etc.) ONLY the metadata/clientData portion.
- If no client data schema or usage is found, generate a simple \`{ "userId": "user_..." }\` object.`;
}
if (currentPayload) {
prompt += `
The current client data in the editor is:
\`\`\`json
${currentPayload}
\`\`\`
Use this as context but generate new client data based on the user's prompt.`;
}
return prompt;
}
function buildSystemPrompt(
taskIdentifier: string,
payloadSchema?: string,
@@ -0,0 +1,117 @@
import { Suspense } from "react";
import { StreamdownRenderer } from "~/components/code/StreamdownRenderer";
import { Header2 } from "~/components/primitives/Headers";
const sampleMarkdown = `# Streamdown Rendering
This is a paragraph with **bold**, *italic*, and \`inline code\` formatting.
## Code Block (TypeScript)
\`\`\`typescript
import { task } from "@trigger.dev/sdk";
export const myTask = task({
id: "my-task",
run: async (payload: { message: string }) => {
const result = await processMessage(payload.message);
this.logger.info("Task completed", { result });
return { success: true, count: 42 };
},
});
\`\`\`
## Code Block (JSON)
\`\`\`json
{
"id": "run_1234",
"status": "completed",
"output": {
"success": true,
"count": 42
}
}
\`\`\`
## Lists
- First item
- Second item with \`code\`
- Third item
1. Ordered first
2. Ordered second
3. Ordered third
## Table
| Feature | Status | Notes |
|---------|--------|-------|
| Syntax highlighting | Done | Custom Shiki theme |
| Markdown rendering | Done | Streamdown v2 |
| Lazy loading | Done | SSR safe |
## Blockquote
> This is a blockquote with some **bold** text and a [link](https://trigger.dev).
---
That's all the elements.
`;
const codeOnlyMarkdown = `Here's a function that demonstrates the color palette:
\`\`\`typescript
const API_URL = "https://api.trigger.dev";
const MAX_RETRIES = 3;
interface TaskConfig {
id: string;
retry: { maxAttempts: number };
}
export async function executeTask(config: TaskConfig): Promise<boolean> {
// Validate the configuration
if (!config.id || config.retry.maxAttempts < 1) {
throw new Error("Invalid task config");
}
for (let i = 0; i < MAX_RETRIES; i++) {
const response = await fetch(\`\${API_URL}/tasks/\${config.id}\`);
const data = response.json();
if (response.ok) {
return true;
}
}
return false;
}
\`\`\`
`;
export default function Story() {
return (
<div className="flex flex-col items-start gap-y-8 p-8">
<div className="max-w-3xl">
<Header2 className="mb-4">Full Markdown</Header2>
<div className="streamdown-container rounded-lg border border-charcoal-700 bg-charcoal-900 p-6 text-sm text-text-bright/90">
<Suspense fallback={<p className="text-text-dimmed">Loading streamdown...</p>}>
<StreamdownRenderer>{sampleMarkdown}</StreamdownRenderer>
</Suspense>
</div>
</div>
<div className="max-w-3xl">
<Header2 className="mb-4">Code Highlighting Theme</Header2>
<div className="streamdown-container rounded-lg border border-charcoal-700 bg-charcoal-900 p-6 text-sm text-text-bright/90">
<Suspense fallback={<p className="text-text-dimmed">Loading streamdown...</p>}>
<StreamdownRenderer>{codeOnlyMarkdown}</StreamdownRenderer>
</Suspense>
</div>
</div>
</div>
);
}
@@ -104,6 +104,10 @@ const stories: Story[] = [
name: "Spinners",
slug: "spinner",
},
{
name: "Streamdown",
slug: "streamdown",
},
{
name: "Switch",
slug: "switch",
+12 -5
View File
@@ -151,11 +151,18 @@
/* Streamdown markdown styling */
.streamdown-container {
/* Streamdown uses shadcn/ui CSS variables - define them for our theme */
--muted: 220 13% 20%;
--muted-foreground: 215 14% 60%;
--foreground: 210 20% 90%;
--border: 217 19% 27%;
/* Streamdown uses shadcn/ui CSS variables - define them for our theme.
These map Tailwind utility classes like bg-background, bg-primary, etc.
that streamdown uses internally for its link safety modal, code blocks,
and other interactive elements. */
--background: 230 16% 9%; /* charcoal-900 #121317 */
--foreground: 215 19% 87%; /* charcoal-200 #D7D9DD */
--muted: 220 8% 17%; /* charcoal-775 #1C1E21 */
--muted-foreground: 220 8% 57%; /* charcoal-400 #878C99 */
--border: 216 7% 27%; /* charcoal-650 #2C3034 */
--primary: 95 100% 66%; /* apple-500 #A8FF53 */
--primary-foreground: 230 16% 9%; /* charcoal-900 */
--sidebar: 228 10% 11%; /* charcoal-850 #15171A */;
/* Code block styling */
& [data-code-block-container] {
+13 -1
View File
@@ -184,7 +184,7 @@ const radius = "0.5rem";
/** @type {import('tailwindcss').Config} */
module.exports = {
content: ["./app/**/*.{ts,jsx,tsx}"],
content: ["./app/**/*.{ts,jsx,tsx}", "./node_modules/streamdown/dist/**/*.js"],
theme: {
container: {
center: true,
@@ -264,6 +264,18 @@ module.exports = {
aiPrompts,
aiMetrics,
errors,
// shadcn/ui color tokens used by streamdown's internal components
// (link safety modal, code block actions, etc.)
// Values are defined via CSS variables in .streamdown-container
background: "hsl(var(--background, 230 16% 9%) / <alpha-value>)",
foreground: "hsl(var(--foreground, 215 19% 87%) / <alpha-value>)",
muted: {
DEFAULT: "hsl(var(--muted, 220 8% 17%) / <alpha-value>)",
foreground: "hsl(var(--muted-foreground, 220 8% 57%) / <alpha-value>)",
},
border: "hsl(var(--border, 216 7% 27%) / <alpha-value>)",
sidebar: "hsl(var(--sidebar, 228 10% 11%) / <alpha-value>)",
"primary-foreground": "hsl(var(--primary-foreground, 230 16% 9%) / <alpha-value>)",
},
focusStyles: {
outline: "1px solid",