feat(webapp): agent-view dashboard for chat.agent runs

Dashboard surfaces for inspecting and debugging chat.agent runs.
Depends on the Sessions primitive (L1) and chat.agent runtime (L2+L3).

Run inspector — chat-aware:
- AgentView + AgentMessageView (run inspector tab for chat.agent runs)
- AIChatMessages + AISpanDetails + types.ts (per-span chat message
  rendering, tool-call/tool-output handling)
- PromptSpanDetails (gen_ai.* span detail panel)
- StreamdownRenderer + shikiTheme (markdown renderer with shiki
  highlighting and v2 patch)
- useAutoScrollToBottom hook

Playground UI (interactive chat.agent debugger):
- /playground index + /playground/$agentParam routes
- /agents route + AgentListPresenter
- PlaygroundPresenter (per-org basin variants, clientData wiring)
- realtime session routes for playground + run inspector chat
- AI-generate-payload + AIPayloadTabContent for the test panel

Navigation + theming:
- SideMenu links for Agents and Playground
- BlankStatePanels copy updates
- tailwind config + tailwind.css storybook hooks
- streamdown@2 dep in apps/webapp/package.json

Includes agent-view-sessions, playground-trigger-config-fields,
run-agent-view, and streamdown-v2-upgrade .server-changes.
This commit is contained in:
Eric Allam
2026-05-10 22:28:40 +01:00
parent a97365d1d5
commit 23ff763359
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",