)}
-
-
- {
- replace({ tab: "overview" });
- }}
- shortcut={{ key: "o" }}
- >
- Overview
-
-
-
@@ -307,7 +296,7 @@ function RunBody({
return (
-
+
)}
@@ -1075,6 +1066,9 @@ function SpanEntity({ span }: { span: Span }) {
code={span.properties}
maxLines={20}
showLineNumbers={false}
+ showCopyButton
+ showTextWrapping
+ showOpenInModal
/>
) : null}
@@ -1120,6 +1114,9 @@ function SpanEntity({ span }: { span: Span }) {
code={span.properties}
maxLines={20}
showLineNumbers={false}
+ showCopyButton
+ showTextWrapping
+ showOpenInModal
/>
) : null}
@@ -1146,6 +1143,15 @@ function SpanEntity({ span }: { span: Span }) {
);
}
+ case "realtime-stream": {
+ return (
+
+ );
+ }
default: {
assertNever(span.entity);
}
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx
new file mode 100644
index 0000000000..f35922a8dc
--- /dev/null
+++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.streams.$streamKey/route.tsx
@@ -0,0 +1,502 @@
+import { BoltIcon, BoltSlashIcon } from "@heroicons/react/20/solid";
+import { type LoaderFunctionArgs } from "@remix-run/server-runtime";
+import { type SSEStreamPart, SSEStreamSubscription } from "@trigger.dev/core/v3";
+import { Clipboard, ClipboardCheck } from "lucide-react";
+import { useCallback, useEffect, useRef, useState } from "react";
+import simplur from "simplur";
+import { ListBulletIcon } from "~/assets/icons/ListBulletIcon";
+import { MoveToBottomIcon } from "~/assets/icons/MoveToBottomIcon";
+import { MoveToTopIcon } from "~/assets/icons/MoveToTopIcon";
+import { SnakedArrowIcon } from "~/assets/icons/SnakedArrowIcon";
+import { Paragraph } from "~/components/primitives/Paragraph";
+import { Spinner } from "~/components/primitives/Spinner";
+import {
+ Tooltip,
+ TooltipContent,
+ TooltipProvider,
+ TooltipTrigger,
+} from "~/components/primitives/Tooltip";
+import { $replica } from "~/db.server";
+import { useEnvironment } from "~/hooks/useEnvironment";
+import { useOrganization } from "~/hooks/useOrganizations";
+import { useProject } from "~/hooks/useProject";
+import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
+import { requireUserId } from "~/services/session.server";
+import { cn } from "~/utils/cn";
+import { v3RunStreamParamsSchema } from "~/utils/pathBuilder";
+
+type ViewMode = "list" | "compact";
+
+type StreamChunk = {
+ id: string;
+ data: unknown;
+ timestamp: number;
+};
+
+export const loader = async ({ request, params }: LoaderFunctionArgs) => {
+ const userId = await requireUserId(request);
+ const { projectParam, organizationSlug, envParam, runParam, streamKey } =
+ v3RunStreamParamsSchema.parse(params);
+
+ const project = await $replica.project.findFirst({
+ where: {
+ slug: projectParam,
+ organization: {
+ slug: organizationSlug,
+ members: {
+ some: {
+ userId,
+ },
+ },
+ },
+ },
+ });
+
+ if (!project) {
+ throw new Response("Not Found", { status: 404 });
+ }
+
+ const run = await $replica.taskRun.findFirst({
+ where: {
+ friendlyId: runParam,
+ projectId: project.id,
+ },
+ include: {
+ runtimeEnvironment: {
+ include: {
+ project: true,
+ organization: true,
+ orgMember: true,
+ },
+ },
+ },
+ });
+
+ if (!run) {
+ throw new Response("Not Found", { status: 404 });
+ }
+
+ if (run.runtimeEnvironment.slug !== envParam) {
+ throw new Response("Not Found", { status: 404 });
+ }
+
+ // Get Last-Event-ID header for resuming from a specific position
+ const lastEventId = request.headers.get("Last-Event-ID") || undefined;
+
+ const realtimeStream = getRealtimeStreamInstance(
+ run.runtimeEnvironment,
+ run.realtimeStreamsVersion
+ );
+
+ return realtimeStream.streamResponse(request, run.friendlyId, streamKey, request.signal, {
+ lastEventId,
+ });
+};
+
+export function RealtimeStreamViewer({
+ runId,
+ streamKey,
+ metadata,
+}: {
+ runId: string;
+ streamKey: string;
+ metadata: Record
| undefined;
+}) {
+ 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 startIndex = typeof metadata?.startIndex === "number" ? metadata.startIndex : undefined;
+ const { chunks, error, isConnected } = useRealtimeStream(resourcePath, startIndex);
+ const scrollRef = useRef(null);
+ const bottomRef = useRef(null);
+ const [isAtBottom, setIsAtBottom] = useState(true);
+ const [viewMode, setViewMode] = useState("list");
+ const [mouseOver, setMouseOver] = useState(false);
+ const [copied, setCopied] = useState(false);
+
+ const getCompactText = useCallback(() => {
+ return chunks
+ .map((chunk) => {
+ if (typeof chunk.data === "string") {
+ return chunk.data;
+ }
+ return JSON.stringify(chunk.data);
+ })
+ .join("");
+ }, [chunks]);
+
+ const onCopied = useCallback(
+ (event: React.MouseEvent) => {
+ event.preventDefault();
+ event.stopPropagation();
+ navigator.clipboard.writeText(getCompactText());
+ setCopied(true);
+ setTimeout(() => {
+ setCopied(false);
+ }, 1500);
+ },
+ [getCompactText]
+ );
+
+ // Use IntersectionObserver to detect when the bottom element is visible
+ useEffect(() => {
+ const bottomElement = bottomRef.current;
+ const scrollElement = scrollRef.current;
+ if (!bottomElement || !scrollElement) return;
+
+ const observer = new IntersectionObserver(
+ (entries) => {
+ const entry = entries[0];
+ if (entry) {
+ setIsAtBottom(entry.isIntersecting);
+ }
+ },
+ {
+ root: scrollElement,
+ threshold: 0.1,
+ rootMargin: "0px",
+ }
+ );
+
+ observer.observe(bottomElement);
+
+ // Also add a scroll listener as a backup to ensure state updates
+ let scrollTimeout: ReturnType | null = null;
+ const handleScroll = () => {
+ if (!scrollElement || !bottomElement) return;
+
+ // Clear any existing timeout
+ if (scrollTimeout) {
+ clearTimeout(scrollTimeout);
+ }
+
+ // Debounce the state update to avoid interrupting smooth scroll
+ scrollTimeout = setTimeout(() => {
+ const scrollBottom = scrollElement.scrollTop + scrollElement.clientHeight;
+ const isNearBottom = scrollElement.scrollHeight - scrollBottom < 50;
+ setIsAtBottom(isNearBottom);
+ }, 100);
+ };
+
+ scrollElement.addEventListener("scroll", handleScroll);
+ // Check initial state
+ const scrollBottom = scrollElement.scrollTop + scrollElement.clientHeight;
+ const isNearBottom = scrollElement.scrollHeight - scrollBottom < 50;
+ setIsAtBottom(isNearBottom);
+
+ return () => {
+ observer.disconnect();
+ scrollElement.removeEventListener("scroll", handleScroll);
+ if (scrollTimeout) {
+ clearTimeout(scrollTimeout);
+ }
+ };
+ }, [chunks.length, viewMode]);
+
+ // Auto-scroll to bottom when new chunks arrive, if we're at the bottom
+ useEffect(() => {
+ if (isAtBottom && bottomRef.current) {
+ bottomRef.current.scrollIntoView({ behavior: "instant", block: "end" });
+ }
+ }, [chunks, isAtBottom]);
+
+ const firstLineNumber = startIndex ?? 0;
+ const lastLineNumber = firstLineNumber + chunks.length - 1;
+ const maxLineNumberWidth = (chunks.length > 0 ? lastLineNumber : firstLineNumber).toString()
+ .length;
+
+ return (
+
+ {/* Header */}
+
+
+
+
+
+
+ {isConnected ? (
+
+ ) : (
+
+ )}
+
+
+ {isConnected ? "Connected" : "Disconnected"}
+
+
+
+
+ Stream:
+ {streamKey}
+
+
+
+
+ {simplur`${chunks.length} chunk[|s]`}
+
+
+
+
+ setViewMode(viewMode === "list" ? "compact" : "list")}
+ className={cn(
+ "text-text-dimmed transition-colors focus-custom",
+ chunks.length === 0
+ ? "cursor-not-allowed opacity-50"
+ : "hover:cursor-pointer hover:text-text-bright"
+ )}
+ >
+ {viewMode === "list" ? (
+
+ ) : (
+
+ )}
+
+
+ {viewMode === "list" ? "Flow as text" : "View as list"}
+
+
+
+
+
+ setMouseOver(true)}
+ onMouseLeave={() => setMouseOver(false)}
+ className={cn(
+ "transition-colors duration-100 focus-custom",
+ chunks.length === 0
+ ? "cursor-not-allowed opacity-50"
+ : copied
+ ? "text-success hover:cursor-pointer"
+ : "text-text-dimmed hover:cursor-pointer hover:text-text-bright"
+ )}
+ >
+ {copied ? (
+
+ ) : (
+
+ )}
+
+
+ {copied ? "Copied" : "Copy"}
+
+
+
+
+
+ {
+ if (isAtBottom) {
+ scrollRef.current?.scrollTo({ top: 0, behavior: "smooth" });
+ } else {
+ bottomRef.current?.scrollIntoView({ behavior: "smooth", block: "end" });
+ }
+ }}
+ className={cn(
+ "text-text-dimmed transition-colors focus-custom",
+ chunks.length === 0
+ ? "cursor-not-allowed opacity-50"
+ : "hover:cursor-pointer hover:text-text-bright"
+ )}
+ >
+ {isAtBottom ? (
+
+ ) : (
+
+ )}
+
+
+ {isAtBottom ? "Scroll to top" : "Scroll to bottom"}
+
+
+
+
+
+
+
+
+ {/* Content */}
+
+ {error && (
+
+
+ Error: {error.message}
+
+
+ )}
+
+ {chunks.length === 0 && !error && (
+
+ {isConnected ? (
+
+
+
+ Waiting for dataβ¦
+
+
+ ) : (
+
+ No data received
+
+ )}
+
+ )}
+
+ {chunks.length > 0 && viewMode === "list" && (
+
+ {chunks.map((chunk, index) => (
+
+ ))}
+ {/* Sentinel element for IntersectionObserver */}
+
+
+ )}
+
+ {chunks.length > 0 && viewMode === "compact" && (
+
+
+ {/* Sentinel element for IntersectionObserver */}
+
+
+ )}
+
+
{compactText}
;
+}
+
+function StreamChunkLine({
+ chunk,
+ lineNumber,
+ maxLineNumberWidth,
+}: {
+ chunk: StreamChunk;
+ lineNumber: number;
+ maxLineNumberWidth: number;
+}) {
+ const formattedData =
+ typeof chunk.data === "string" ? chunk.data : JSON.stringify(chunk.data, null, 2);
+
+ const date = new Date(chunk.timestamp);
+ const timeString = date.toLocaleTimeString("en-US", {
+ hour12: false,
+ hour: "2-digit",
+ minute: "2-digit",
+ second: "2-digit",
+ });
+ const milliseconds = date.getMilliseconds().toString().padStart(3, "0");
+ const timestamp = `${timeString}.${milliseconds}`;
+
+ return (
+
+ {/* Line number */}
+
+ {lineNumber}
+
+
+ {/* Timestamp */}
+
{timestamp}
+
+ {/* Content */}
+
{formattedData}
+
Error: {error.message}
;
+ *
+ * // Parts array accumulates all chunks
+ * const fullText = parts.join("");
+ *
+ * return {fullText}
;
+ * }
+ * ```
+ *
+ * @example
+ * ```tsx
+ * // With custom options
+ * const { parts, error, stop } = useRealtimeStream(
+ * runId,
+ * "chat-stream",
+ * {
+ * accessToken: publicKey,
+ * timeoutInSeconds: 120,
+ * startIndex: 10, // Start from the 10th chunk
+ * throttleInMs: 50, // Throttle updates to every 50ms
+ * onData: (chunk) => {
+ * console.log("New chunk received:", chunk);
+ * },
+ * }
+ * );
+ *
+ * // Manually stop the subscription
+ *
+ * ```
+ */
+export function useRealtimeStream(
+ runId: string,
+ streamKey: string,
+ options?: UseRealtimeStreamOptions
+): UseRealtimeStreamInstance;
+/**
+ * Hook to subscribe to realtime updates of a stream using the default stream key (`"default"`).
+ *
+ * This is a convenience overload that allows you to subscribe to the default stream without
+ * specifying a stream key. The stream will be accessed with the key `"default"`.
+ *
+ * @template TPart - The type of each chunk/part in the stream
+ * @param runId - The unique identifier of the run to subscribe to
+ * @param options - Optional configuration for the stream subscription
+ * @returns An object containing:
+ * - `parts`: An array of all stream chunks received so far (accumulates over time)
+ * - `error`: Any error that occurred during subscription
+ * - `stop`: A function to manually stop the subscription
+ *
+ * @example
+ * ```tsx
+ * "use client";
+ * import { useRealtimeStream } from "@trigger.dev/react-hooks";
+ *
+ * function DefaultStreamViewer({ runId }: { runId: string }) {
+ * // Subscribe to the default stream
+ * const { parts, error } = useRealtimeStream(runId, {
+ * accessToken: process.env.NEXT_PUBLIC_TRIGGER_PUBLIC_KEY,
+ * });
+ *
+ * if (error) return Error: {error.message}
;
+ *
+ * const fullText = parts.join("");
+ * return {fullText}
;
+ * }
+ * ```
+ *
+ * @example
+ * ```tsx
+ * // Conditionally enable the stream
+ * const { parts } = useRealtimeStream(runId, {
+ * accessToken: publicKey,
+ * enabled: !!runId && isStreaming, // Only subscribe when runId exists and isStreaming is true
+ * });
+ * ```
+ */
+export function useRealtimeStream(
+ runId: string,
+ options?: UseRealtimeStreamOptions
+): UseRealtimeStreamInstance;
+export function useRealtimeStream(
+ runId: string,
+ streamKeyOrOptions?: string | UseRealtimeStreamOptions,
+ options?: UseRealtimeStreamOptions
+): UseRealtimeStreamInstance {
+ // Handle overload: useRealtimeStream(runId, options?) or useRealtimeStream(runId, streamKey, options?)
+ const DEFAULT_STREAM_KEY = "default";
+
+ let streamKey: string;
+ let opts: UseRealtimeStreamOptions | undefined;
+
+ if (typeof streamKeyOrOptions === "string") {
+ // useRealtimeStream(runId, streamKey, options?)
+ streamKey = streamKeyOrOptions;
+ opts = options;
+ } else {
+ // useRealtimeStream(runId, options?)
+ streamKey = DEFAULT_STREAM_KEY;
+ opts = streamKeyOrOptions;
+ }
+ const hookId = useId();
+ const idKey = opts?.id ?? hookId;
+
+ const [initialPartsFallback] = useState([] as Array);
+
+ // Store the streams state in SWR, using the idKey as the key to share states.
+ const { data: parts, mutate: mutateParts } = useSWR>(
+ [idKey, runId, streamKey, "parts"],
+ null,
+ {
+ fallbackData: initialPartsFallback,
+ }
+ );
+
+ // Keep the latest streams in a ref.
+ const partsRef = useRef>(parts ?? ([] as Array));
+ useEffect(() => {
+ partsRef.current = parts || ([] as Array);
+ }, [parts]);
+
+ // Add state to track when the subscription is complete
+ const { data: isComplete = false, mutate: setIsComplete } = useSWR(
+ [idKey, runId, streamKey, "complete"],
+ null
+ );
+
+ const { data: error = undefined, mutate: setError } = useSWR(
+ [idKey, runId, streamKey, "error"],
+ null
+ );
+
+ // Abort controller to cancel the current API call.
+ const abortControllerRef = useRef(null);
+
+ const stop = useCallback(() => {
+ if (abortControllerRef.current) {
+ abortControllerRef.current.abort();
+ abortControllerRef.current = null;
+ }
+ }, []);
+
+ const onData = useCallback(
+ (data: TPart) => {
+ if (opts?.onData) {
+ opts.onData(data);
+ }
+ },
+ [opts?.onData]
+ );
+
+ const apiClient = useApiClient(opts);
+
+ const triggerRequest = useCallback(async () => {
+ try {
+ if (!runId || !apiClient) {
+ return;
+ }
+
+ const abortController = new AbortController();
+ abortControllerRef.current = abortController;
+
+ await processRealtimeStream(
+ runId,
+ streamKey,
+ apiClient,
+ mutateParts,
+ partsRef,
+ setError,
+ onData,
+ abortControllerRef,
+ opts?.timeoutInSeconds,
+ opts?.startIndex,
+ opts?.throttleInMs ?? 16
+ );
+ } catch (err) {
+ // Ignore abort errors as they are expected.
+ if ((err as any).name === "AbortError") {
+ abortControllerRef.current = null;
+ return;
+ }
+
+ setError(err as Error);
+ } finally {
+ if (abortControllerRef.current) {
+ abortControllerRef.current = null;
+ }
+
+ // Mark the subscription as complete
+ setIsComplete(true);
+ }
+ }, [runId, streamKey, mutateParts, partsRef, abortControllerRef, apiClient, setError, onData, opts]);
+
+ useEffect(() => {
+ if (typeof opts?.enabled === "boolean" && !opts.enabled) {
+ return;
+ }
+
+ if (!runId) {
+ return;
+ }
+
+ triggerRequest().finally(() => {});
+
+ return () => {
+ stop();
+ };
+ }, [runId, stop, opts?.enabled, triggerRequest]);
+
+ return { parts: parts ?? initialPartsFallback, error, stop };
+}
+
async function processRealtimeBatch(
batchId: string,
apiClient: ApiClient,
@@ -734,3 +1022,47 @@ async function processRealtimeRun(
mutateRunData(part);
}
}
+
+async function processRealtimeStream(
+ runId: string,
+ streamKey: string,
+ apiClient: ApiClient,
+ mutatePartsData: KeyedMutator>,
+ existingPartsRef: React.MutableRefObject>,
+ onError: (e: Error) => void,
+ onData: (data: TPart) => void,
+ abortControllerRef: React.MutableRefObject,
+ timeoutInSeconds?: number,
+ startIndex?: number,
+ throttleInMs?: number
+) {
+ try {
+ const stream = await apiClient.fetchStream(runId, streamKey, {
+ signal: abortControllerRef.current?.signal,
+ timeoutInSeconds,
+ lastEventId: startIndex ? (startIndex - 1).toString() : undefined,
+ });
+
+ // Throttle the stream
+ const streamQueue = createThrottledQueue(async (parts) => {
+ mutatePartsData([...existingPartsRef.current, ...parts]);
+ }, throttleInMs);
+
+ for await (const part of stream) {
+ onData(part);
+ streamQueue.add(part);
+ }
+ } catch (err) {
+ if ((err as any).name === "AbortError") {
+ return;
+ }
+
+ if (err instanceof Error) {
+ onError(err);
+ } else {
+ onError(new Error(String(err)));
+ }
+
+ throw err;
+ }
+}
diff --git a/packages/trigger-sdk/src/v3/index.ts b/packages/trigger-sdk/src/v3/index.ts
index 77448ae432..dcc258455b 100644
--- a/packages/trigger-sdk/src/v3/index.ts
+++ b/packages/trigger-sdk/src/v3/index.ts
@@ -16,6 +16,7 @@ export * from "./locals.js";
export * from "./otel.js";
export * from "./schemas.js";
export * from "./heartbeats.js";
+export * from "./streams.js";
export type { Context };
import type { Context } from "./shared.js";
diff --git a/packages/trigger-sdk/src/v3/metadata.ts b/packages/trigger-sdk/src/v3/metadata.ts
index b0c321d81d..080c87e345 100644
--- a/packages/trigger-sdk/src/v3/metadata.ts
+++ b/packages/trigger-sdk/src/v3/metadata.ts
@@ -7,6 +7,7 @@ import {
type AsyncIterableStream,
} from "@trigger.dev/core/v3";
import { tracer } from "./tracer.js";
+import { streams } from "./streams.js";
const parentMetadataUpdater: RunMetadataUpdater = runMetadata.parent;
const rootMetadataUpdater: RunMetadataUpdater = runMetadata.root;
@@ -228,12 +229,19 @@ async function refreshMetadata(requestOptions?: ApiRequestOptions): Promise