From 16a6fab790abf5fb0df64c940e64d0f1809cbba4 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Mon, 23 Feb 2026 13:50:44 +0000 Subject: [PATCH] Better span experience, show input streams on dashboard, cleanup the data format --- .../app/presenters/v3/SpanPresenter.server.ts | 35 +++++++++++++ ....runs.$runFriendlyId.input-streams.wait.ts | 27 ++++------ ...ltime.v1.streams.$runId.input.$streamId.ts | 8 +-- .../route.tsx | 1 + .../route.tsx | 9 +++- packages/core/src/v3/inputStreams/manager.ts | 24 +++------ packages/trigger-sdk/src/v3/streams.ts | 51 +++++++++++++++++-- 7 files changed, 110 insertions(+), 45 deletions(-) diff --git a/apps/webapp/app/presenters/v3/SpanPresenter.server.ts b/apps/webapp/app/presenters/v3/SpanPresenter.server.ts index a85d8b20d..ce83c2e24 100644 --- 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 index 3ad87a949..eeb6673a4 100644 --- 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, - }, - }); + // 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 { - // Skip malformed records - } + // 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 index df3417ac9..c33ad89f3 100644 --- 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 index a78c95d60..e46eaa514 100644 --- 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 index a1d941597..b5763bb4e 100644 --- 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 | 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" > - Stream: - {streamKey} + {displayName ? "Input stream:" : "Stream:"} + {displayName ?? streamKey}
@@ -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> | null = null; diff --git a/packages/core/src/v3/inputStreams/manager.ts b/packages/core/src/v3/inputStreams/manager.ts index e1481852a..7347f76d1 100644 --- a/packages/core/src/v3/inputStreams/manager.ts +++ b/packages/core/src/v3/inputStreams/manager.ts @@ -10,14 +10,6 @@ type OnceWaiter = { timeoutHandle?: ReturnType; }; -/** - * 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 { try { - const stream = await this.apiClient.fetchStream( + const stream = await this.apiClient.fetchStream( 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 index 2790e2929..2cfbfaa40 100644 --- a/packages/trigger-sdk/src/v3/streams.ts +++ b/packages/trigger-sdk/src/v3/streams.ts @@ -703,7 +703,29 @@ function input(opts: { id: string }): RealtimeDefinedInputStream { ); }, once(options) { - return inputStreams.once(opts.id, options) as Promise; + const ctx = taskContext.ctx; + const runId = ctx?.run.id; + + return tracer.startActiveSpan( + `inputStream.once()`, + async () => { + return inputStreams.once(opts.id, options) as Promise; + }, + { + 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(opts: { id: string }): RealtimeDefinedInputStream { 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(opts: { id: string }): RealtimeDefinedInputStream { ...accessoryAttributes({ items: [ { - text: `input:${opts.id}`, + text: opts.id, variant: "normal", }, ], @@ -792,8 +817,26 @@ function input(opts: { id: string }): RealtimeDefinedInputStream { }); }, 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", + }), + }, + } + ); }, }; }