Better span experience, show input streams on dashboard, cleanup the data format

ericallam · ericallam · commit 96908e2d0599 · 2026-02-27T14:01:41.000Z
diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts
@@ -629,6 +629,41 @@ export class SpanPresenter extends BasePresenter {
           },
         };
       }
+      case "input-stream": {
+        if (!span.entity.id) {
+          logger.error(`SpanPresenter: No input stream id`, {
+            spanId,
+            inputStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        const [runId, streamId] = span.entity.id.split(":");
+
+        if (!runId || !streamId) {
+          logger.error(`SpanPresenter: Invalid input stream id`, {
+            spanId,
+            inputStreamId: span.entity.id,
+          });
+          return { ...data, entity: null };
+        }
+
+        // Translate user-facing stream ID to internal S2 stream name
+        const s2StreamKey = `$trigger.input:${streamId}`;
+
+        return {
+          ...data,
+          entity: {
+            type: "realtime-stream" as const,
+            object: {
+              runId,
+              streamKey: s2StreamKey,
+              displayName: streamId,
+              metadata: undefined,
+            },
+          },
+        };
+      }
       default:
         return { ...data, entity: null };
     }
diff --git a/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts b/apps/webapp/app/routes/api.v1.runs.$runFriendlyId.input-streams.wait.ts
@@ -110,24 +110,19 @@ const { action, loader } = createActionApiRoute(
 
             if (records.length > 0) {
               const record = records[0]!;
-              try {
-                const parsed = JSON.parse(record.data) as { data: unknown };
-
-                // Data exists — complete the waitpoint immediately
-                await engine.completeWaitpoint({
-                  id: result.waitpoint.id,
-                  output: {
-                    value: JSON.stringify(parsed.data),
-                    type: "application/json",
-                    isError: false,
-                  },
-                });
-
-                // Clean up the Redis cache since we completed it ourselves
-                await deleteInputStreamWaitpoint(run.friendlyId, body.streamId);
-              } catch {
-                // Skip malformed records
-              }
+
+              // Record data is the raw user payload — no wrapper to unwrap
+              await engine.completeWaitpoint({
+                id: result.waitpoint.id,
+                output: {
+                  value: record.data,
+                  type: "application/json",
+                  isError: false,
+                },
+              });
+
+              // Clean up the Redis cache since we completed it ourselves
+              await deleteInputStreamWaitpoint(run.friendlyId, body.streamId);
             }
           }
         } catch {
diff --git a/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts b/apps/webapp/app/routes/realtime.v1.streams.$runId.input.$streamId.ts
@@ -67,13 +67,9 @@ const { action } = createActionApiRoute(
       run.realtimeStreamsVersion
     );
 
-    // Build the input stream record
+    // Build the input stream record (raw user data, no wrapper)
     const recordId = `inp_${crypto.randomUUID().replace(/-/g, "").slice(0, 12)}`;
-    const record = JSON.stringify({
-      data: body.data.data,
-      ts: Date.now(),
-      id: recordId,
-    });
+    const record = JSON.stringify(body.data.data);
 
     // Append the record to the per-stream S2 stream (auto-creates on first write)
     await realtimeStream.appendPart(
diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx
@@ -1348,6 +1348,7 @@ function SpanEntity({ span }: { span: Span }) {
           runId={span.entity.object.runId}
           streamKey={span.entity.object.streamKey}
           metadata={span.entity.object.metadata}
+          displayName={span.entity.object.displayName}
         />
       );
     }
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
@@ -98,10 +98,12 @@ export function RealtimeStreamViewer({
   runId,
   streamKey,
   metadata,
+  displayName,
 }: {
   runId: string;
   streamKey: string;
   metadata: Record<string, unknown> | undefined;
+  displayName?: string;
 }) {
   const organization = useOrganization();
   const project = useProject();
@@ -244,8 +246,8 @@ export function RealtimeStreamViewer({
               variant="small/bright"
               className="mb-0 flex min-w-0 items-center gap-1 truncate whitespace-nowrap"
             >
-              <span>Stream:</span>
-              <span className="truncate font-mono text-text-dimmed">{streamKey}</span>
+              <span>{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">
@@ -487,6 +489,9 @@ function useRealtimeStream(resourcePath: string, startIndex?: number) {
   const [isConnected, setIsConnected] = useState(false);
 
   useEffect(() => {
+    setChunks([]);
+    setError(null);
+
     const abortController = new AbortController();
     let reader: ReadableStreamDefaultReader<SSEStreamPart<unknown>> | null = null;
 
diff --git a/packages/core/src/v3/inputStreams/manager.ts b/packages/core/src/v3/inputStreams/manager.ts
@@ -10,14 +10,6 @@ type OnceWaiter = {
   timeoutHandle?: ReturnType<typeof setTimeout>;
 };
 
-/**
- * InputStreamRecord is the shape of records on a per-stream S2 stream.
- */
-interface InputStreamRecord {
-  data: unknown;
-  ts: number;
-  id: string;
-}
 
 type TailState = {
   abortController: AbortController;
@@ -195,7 +187,7 @@ export class StandardInputStreamManager implements InputStreamManager {
 
   async #runTail(runId: string, streamId: string, signal: AbortSignal): Promise<void> {
     try {
-      const stream = await this.apiClient.fetchStream<InputStreamRecord>(
+      const stream = await this.apiClient.fetchStream<unknown>(
         runId,
         `input/${streamId}`,
         {
@@ -225,21 +217,19 @@ export class StandardInputStreamManager implements InputStreamManager {
       for await (const record of stream) {
         if (signal.aborted) break;
 
-        // S2 SSE returns record bodies as JSON strings; parse into InputStreamRecord
-        let parsed: InputStreamRecord;
+        // S2 SSE returns record bodies as JSON strings; parse if needed
+        let data: unknown;
         if (typeof record === "string") {
           try {
-            parsed = JSON.parse(record) as InputStreamRecord;
+            data = JSON.parse(record);
           } catch {
-            continue;
+            data = record;
           }
-        } else if (record.data !== undefined) {
-          parsed = record;
         } else {
-          continue;
+          data = record;
         }
 
-        this.#dispatch(streamId, parsed.data);
+        this.#dispatch(streamId, data);
       }
     } catch (error) {
       // AbortError is expected when disconnecting
diff --git a/packages/trigger-sdk/src/v3/streams.ts b/packages/trigger-sdk/src/v3/streams.ts
@@ -703,7 +703,29 @@ function input<TData>(opts: { id: string }): RealtimeDefinedInputStream<TData> {
       );
     },
     once(options) {
-      return inputStreams.once(opts.id, options) as Promise<TData>;
+      const ctx = taskContext.ctx;
+      const runId = ctx?.run.id;
+
+      return tracer.startActiveSpan(
+        `inputStream.once()`,
+        async () => {
+          return inputStreams.once(opts.id, options) as Promise<TData>;
+        },
+        {
+          attributes: {
+            [SemanticInternalAttributes.STYLE_ICON]: "streams",
+            [SemanticInternalAttributes.ENTITY_TYPE]: "input-stream",
+            ...(runId
+              ? { [SemanticInternalAttributes.ENTITY_ID]: `${runId}:${opts.id}` }
+              : {}),
+            streamId: opts.id,
+            ...accessoryAttributes({
+              items: [{ text: opts.id, variant: "normal" }],
+              style: "codepath",
+            }),
+          },
+        }
+      );
     },
     peek() {
       return inputStreams.peek(opts.id) as TData | undefined;
@@ -732,6 +754,9 @@ function input<TData>(opts: { id: string }): RealtimeDefinedInputStream<TData> {
                 lastSeqNum: inputStreams.lastSeqNum(opts.id),
               });
 
+              // Set the entity ID now that we have the waitpoint ID
+              span.setAttribute(SemanticInternalAttributes.ENTITY_ID, response.waitpointId);
+
               // 2. Block the run on the waitpoint
               const waitResponse = await apiClient.waitForWaitpointToken({
                 runFriendlyId: ctx.run.id,
@@ -775,7 +800,7 @@ function input<TData>(opts: { id: string }): RealtimeDefinedInputStream<TData> {
                 ...accessoryAttributes({
                   items: [
                     {
-                      text: `input:${opts.id}`,
+                      text: opts.id,
                       variant: "normal",
                     },
                   ],
@@ -792,8 +817,26 @@ function input<TData>(opts: { id: string }): RealtimeDefinedInputStream<TData> {
       });
     },
     async send(runId, data, options) {
-      const apiClient = apiClientManager.clientOrThrow();
-      await apiClient.sendInputStream(runId, opts.id, data, options?.requestOptions);
+      return tracer.startActiveSpan(
+        `inputStream.send()`,
+        async () => {
+          const apiClient = apiClientManager.clientOrThrow();
+          await apiClient.sendInputStream(runId, opts.id, data, options?.requestOptions);
+        },
+        {
+          attributes: {
+            [SemanticInternalAttributes.STYLE_ICON]: "streams",
+            [SemanticInternalAttributes.ENTITY_TYPE]: "input-stream",
+            [SemanticInternalAttributes.ENTITY_ID]: `${runId}:${opts.id}`,
+            streamId: opts.id,
+            runId,
+            ...accessoryAttributes({
+              items: [{ text: opts.id, variant: "normal" }],
+              style: "codepath",
+            }),
+          },
+        }
+      );
     },
   };
 }

Original file line number	Diff line number	Diff line change
`@@ -1348,6 +1348,7 @@ function SpanEntity({ span }: { span: Span }) {`
`1348`	`1348`	`runId={span.entity.object.runId}`
`1349`	`1349`	`streamKey={span.entity.object.streamKey}`
`1350`	`1350`	`metadata={span.entity.object.metadata}`
	`1351`	`+ displayName={span.entity.object.displayName}`
`1351`	`1352`	`/>`
`1352`	`1353`	`);`
`1353`	`1354`	`}`