diff --git a/.changeset/stream-tail-web-cursors.md b/.changeset/stream-tail-web-cursors.md new file mode 100644 index 0000000000..99ab36ed38 --- /dev/null +++ b/.changeset/stream-tail-web-cursors.md @@ -0,0 +1,5 @@ +--- +'@workflow/web': patch +--- + +Resume dashboard stream readers from the last delivered chunk without replaying the current tail. diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 18b0650099..8d98f3039a 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1056,6 +1056,7 @@ describe('e2e', () => { const streamName = `${run.runId.replace('wrun_', 'strm_')}_user`; const paginatedChunks: Uint8Array[] = []; let cursor: string | null = null; + let hasMore: boolean; do { const page = await world.streams.getChunks(run.runId, streamName, { limit: 1, // small page size to exercise pagination @@ -1065,10 +1066,11 @@ describe('e2e', () => { paginatedChunks.push(chunk.data); } cursor = page.cursor; - if (!page.hasMore) { + hasMore = page.hasMore; + if (!hasMore) { expect(page.done).toBe(true); } - } while (cursor); + } while (hasMore); // Both methods should return the same number of chunks expect(paginatedChunks).toHaveLength(streamChunks.length); diff --git a/packages/web/app/components/run-detail-view.tsx b/packages/web/app/components/run-detail-view.tsx index 97d182f7e7..c882d23f38 100644 --- a/packages/web/app/components/run-detail-view.tsx +++ b/packages/web/app/components/run-detail-view.tsx @@ -430,7 +430,7 @@ export function RunDetailView({ isLive: streamIsLive, isInitialLoading: streamIsInitialLoading, error: streamError, - } = useStreamReader(env, selectedStreamId, runId, encryptionKey, run.status); + } = useStreamReader(selectedStreamId, runId, encryptionKey, run.status); const handleCancelClick = () => { setShowCancelDialog(true); diff --git a/packages/web/app/lib/client/workflow-streams.test.ts b/packages/web/app/lib/client/workflow-streams.test.ts index 866a49d91e..ec49efa63a 100644 --- a/packages/web/app/lib/client/workflow-streams.test.ts +++ b/packages/web/app/lib/client/workflow-streams.test.ts @@ -76,7 +76,7 @@ describe('readStream', () => { headers: { 'X-Stream-Cursor': 'abc123', 'X-Stream-Done': 'false' }, }); - const result = await readStream(env, 'stream-1', 'run-1'); + const result = await readStream('stream-1', 'run-1', null); expect(result.body).toBe(mockBody); expect(result.cursor).toBe('abc123'); expect(result.done).toBe(false); @@ -93,13 +93,7 @@ describe('readStream', () => { headers: { 'X-Stream-Done': 'true' }, }); - const result = await readStream( - env, - 'stream-1', - 'run-1', - undefined, - 'cur_xyz' - ); + const result = await readStream('stream-1', 'run-1', 'cur_xyz'); expect(result.done).toBe(true); expect(result.cursor).toBeNull(); expect(globalThis.fetch).toHaveBeenCalledWith( @@ -111,7 +105,7 @@ describe('readStream', () => { it('throws WorkflowWebAPIError when response is not ok', async () => { mockFetchResponse({ ok: false, status: 500 }); - await expect(readStream(env, 'stream-1', 'run-1')).rejects.toThrow( + await expect(readStream('stream-1', 'run-1', null)).rejects.toThrow( 'Failed to read stream: 500' ); }); @@ -129,7 +123,7 @@ describe('readStream', () => { }), }); - await expect(readStream(env, 'stream-1', 'run-1')).rejects.toThrow( + await expect(readStream('stream-1', 'run-1', null)).rejects.toThrow( 'invalid stream' ); }); @@ -137,7 +131,7 @@ describe('readStream', () => { it('throws WorkflowWebAPIError when body is null', async () => { mockFetchResponse({ ok: true, body: null }); - await expect(readStream(env, 'stream-1', 'run-1')).rejects.toThrow( + await expect(readStream('stream-1', 'run-1', null)).rejects.toThrow( 'Failed to read stream: no body' ); }); @@ -145,7 +139,7 @@ describe('readStream', () => { it('wraps non-WorkflowWebAPIError in WorkflowWebAPIError', async () => { globalThis.fetch = vi.fn().mockRejectedValue(new TypeError('network fail')); - const err = await readStream(env, 'stream-1', 'run-1').catch((e) => e); + const err = await readStream('stream-1', 'run-1', null).catch((e) => e); expect(err).toBeInstanceOf(WorkflowWebAPIError); expect(err.message).toBe('Failed to read stream'); expect(err.cause).toBeInstanceOf(TypeError); diff --git a/packages/web/app/lib/client/workflow-streams.ts b/packages/web/app/lib/client/workflow-streams.ts index 77d0a072dd..8a573f85a3 100644 --- a/packages/web/app/lib/client/workflow-streams.ts +++ b/packages/web/app/lib/client/workflow-streams.ts @@ -23,11 +23,10 @@ export interface StreamResponse { } export async function readStream( - _env: EnvMap, streamId: string, runId: string, - signal?: AbortSignal, - cursor?: string | null + cursor: string | null, + signal?: AbortSignal ): Promise { try { const params = new URLSearchParams({ runId }); diff --git a/packages/web/app/lib/hooks/use-stream-reader.test.ts b/packages/web/app/lib/hooks/use-stream-reader.test.ts index 0d3592d84c..8dd40760dc 100644 --- a/packages/web/app/lib/hooks/use-stream-reader.test.ts +++ b/packages/web/app/lib/hooks/use-stream-reader.test.ts @@ -8,8 +8,6 @@ vi.mock('~/lib/workflow-api-client', () => ({ readStream: vi.fn(), })); -const env = {}; - function emptyStreamResponse(done: boolean): StreamResponse { return { body: new ReadableStream({ @@ -36,7 +34,7 @@ describe('useStreamReader status', () => { ); const { result, unmount } = renderHook(() => - useStreamReader(env, 'stream-1', 'run-1', null, 'running') + useStreamReader('stream-1', 'run-1', null, 'running') ); expect(result.current.isInitialLoading).toBe(true); @@ -56,7 +54,7 @@ describe('useStreamReader status', () => { vi.mocked(readStream).mockResolvedValue(emptyStreamResponse(true)); const { result } = renderHook(() => - useStreamReader(env, 'stream-1', 'run-1', null, 'running') + useStreamReader('stream-1', 'run-1', null, 'running') ); await waitFor(() => { @@ -69,7 +67,7 @@ describe('useStreamReader status', () => { vi.mocked(readStream).mockResolvedValue(emptyStreamResponse(false)); const { result } = renderHook(() => - useStreamReader(env, 'stream-1', 'run-1', null, 'completed') + useStreamReader('stream-1', 'run-1', null, 'completed') ); await waitFor(() => { diff --git a/packages/web/app/lib/hooks/use-stream-reader.ts b/packages/web/app/lib/hooks/use-stream-reader.ts index 7e6bbfd19a..19d164787f 100644 --- a/packages/web/app/lib/hooks/use-stream-reader.ts +++ b/packages/web/app/lib/hooks/use-stream-reader.ts @@ -8,7 +8,6 @@ import { import { getWebRevivers } from '@workflow/web-shared'; import type { WorkflowRunStatus } from '@workflow/world'; import { useCallback, useEffect, useRef, useState } from 'react'; -import type { EnvMap } from '~/lib/types'; import { readStream } from '~/lib/workflow-api-client'; export interface StreamChunk { @@ -21,7 +20,7 @@ const FRAME_HEADER_SIZE = 4; const ENCRYPTED_PLACEHOLDER = '[Encrypted]'; const POLL_INTERVAL_MS = 3000; -function isRunActive(status?: WorkflowRunStatus): boolean { +function isRunActive(status: WorkflowRunStatus): boolean { return status === 'pending' || status === 'running'; } @@ -49,21 +48,16 @@ function detectEncoding(data: Uint8Array): StreamEncoding { } export function useStreamReader( - env: EnvMap, streamId: string | null, - runId?: string, - encryptionKey?: Uint8Array | null, - runStatus?: WorkflowRunStatus + runId: string, + encryptionKey: Uint8Array | null, + runStatus: WorkflowRunStatus ) { const [chunks, setChunks] = useState([]); const [isLive, setIsLive] = useState(false); - const [isInitialLoading, setIsInitialLoading] = useState( - Boolean(streamId && runId) - ); + const [isInitialLoading, setIsInitialLoading] = useState(Boolean(streamId)); const [error, setError] = useState(null); - const abortControllerRef = useRef(null); const chunkIdRef = useRef(0); - const frameCountRef = useRef(0); const pollTimerRef = useRef | null>(null); const runStatusRef = useRef(runStatus); runStatusRef.current = runStatus; @@ -132,9 +126,8 @@ export function useStreamReader( setChunks([]); setError(null); setIsLive(false); - setIsInitialLoading(Boolean(streamId && runId)); + setIsInitialLoading(Boolean(streamId)); chunkIdRef.current = 0; - frameCountRef.current = 0; serverCursorRef.current = null; if (pollTimerRef.current) { @@ -142,14 +135,13 @@ export function useStreamReader( pollTimerRef.current = null; } - if (!streamId || !runId) { + if (!streamId) { setIsInitialLoading(false); return; } let mounted = true; const abortController = new AbortController(); - abortControllerRef.current = abortController; const revivers = getWebRevivers(); @@ -168,37 +160,33 @@ export function useStreamReader( * Fetch stream data and parse frames. * * When `cursor` is provided, the server only returns chunks after that - * position (incremental fetch). `skipFrames` skips N frames from the - * response to handle the overlap from cursor-based pagination. + * position (incremental fetch). */ const fetchAndParse = async ( - targetBuffer: StreamChunk[], cryptoKey: PayloadKey | undefined, - options?: { skipFrames?: number; cursor?: string | null } + cursor: string | null ): Promise< | { encrypted: true } | { encrypted: false; - frameCount: number; + chunks: StreamChunk[]; cursor: string | null; done: boolean; } > => { const streamResponse = await readStream( - env, streamId, runId, - abortController.signal, - options?.cursor + cursor, + abortController.signal ); - const skipFrames = options?.skipFrames ?? 0; + const chunks: StreamChunk[] = []; const reader = streamResponse.body.getReader(); const decoder = new TextDecoder(); let buffer = new Uint8Array(0); let encoding: StreamEncoding | null = null; let textRemainder = ''; - let frameIndex = 0; const appendToBuffer = (data: Uint8Array) => { const newBuffer = new Uint8Array(buffer.length + data.length); @@ -213,10 +201,7 @@ export function useStreamReader( const { value, done } = await reader.read(); if (done) { if (encoding === 'legacy' && textRemainder.trim()) { - frameIndex++; - if (frameIndex > skipFrames) { - targetBuffer.push(parseLegacyLine(textRemainder.trim())); - } + chunks.push(parseLegacyLine(textRemainder.trim())); textRemainder = ''; } break; @@ -243,10 +228,7 @@ export function useStreamReader( for (const line of lines) { const trimmed = line.trim(); if (trimmed) { - frameIndex++; - if (frameIndex > skipFrames) { - targetBuffer.push(parseLegacyLine(trimmed)); - } + chunks.push(parseLegacyLine(trimmed)); } } continue; @@ -277,17 +259,12 @@ export function useStreamReader( ); offset += FRAME_HEADER_SIZE + frameLength; - frameIndex++; - if (frameIndex <= skipFrames) { - continue; - } - const result = await processFrame(frameData, cryptoKey, revivers); if (result.encrypted) { reader.cancel().catch(() => {}); return { encrypted: true }; } - targetBuffer.push(result.chunk); + chunks.push(result.chunk); framesInBatch++; if (framesInBatch % YIELD_EVERY_N_FRAMES === 0) { @@ -301,7 +278,7 @@ export function useStreamReader( return { encrypted: false, - frameCount: frameIndex, + chunks, cursor: streamResponse.cursor, done: streamResponse.done, }; @@ -315,8 +292,7 @@ export function useStreamReader( ? await deriveRunPayloadKeys(encryptionKey) : undefined; - const initialChunks: StreamChunk[] = []; - const result = await fetchAndParse(initialChunks, cryptoKey); + const result = await fetchAndParse(cryptoKey, null); if (result.encrypted) { if (mounted) { @@ -327,12 +303,11 @@ export function useStreamReader( return; } - frameCountRef.current = result.frameCount; serverCursorRef.current = result.cursor; if (!mounted || abortController.signal.aborted) return; - setChunks(initialChunks); + setChunks(result.chunks); setIsInitialLoading(false); // If the stream itself is done, no need to poll regardless of run status @@ -346,18 +321,15 @@ export function useStreamReader( const poll = async () => { if (!mounted || abortController.signal.aborted) return; try { - const newChunks: StreamChunk[] = []; - const pollResult = await fetchAndParse(newChunks, cryptoKey, { - cursor: serverCursorRef.current, - skipFrames: frameCountRef.current, - }); + const pollResult = await fetchAndParse( + cryptoKey, + serverCursorRef.current + ); if (!pollResult.encrypted) { - frameCountRef.current = pollResult.frameCount; - if (pollResult.cursor) { - serverCursorRef.current = pollResult.cursor; - } - if (newChunks.length > 0 && mounted) { - setChunks((prev) => [...prev, ...newChunks]); + serverCursorRef.current = + pollResult.cursor ?? serverCursorRef.current; + if (pollResult.chunks.length > 0 && mounted) { + setChunks((prev) => [...prev, ...pollResult.chunks]); } if (pollResult.done) { setIsLive(false); @@ -396,7 +368,7 @@ export function useStreamReader( } }; // eslint-disable-next-line react-hooks/exhaustive-deps - }, [env, streamId, runId, encryptionKey, processFrame]); + }, [streamId, runId, encryptionKey, processFrame]); // When run finishes, stop polling useEffect(() => { diff --git a/packages/web/app/routes/api.stream.$streamId.tsx b/packages/web/app/routes/api.stream.$streamId.tsx index 8fde995ccb..83a560b034 100644 --- a/packages/web/app/routes/api.stream.$streamId.tsx +++ b/packages/web/app/routes/api.stream.$streamId.tsx @@ -38,12 +38,7 @@ export async function loader({ params, request }: Route.LoaderArgs) { const cursor = url.searchParams.get('cursor') ?? undefined; try { - const result = await readStreamChunksServerAction( - {}, - streamId, - runId, - cursor - ); + const result = await readStreamChunksServerAction(streamId, runId, cursor); if (!('buffer' in result)) { return Response.json(result, { status: 500 }); diff --git a/packages/web/app/server/workflow-server-actions.server.ts b/packages/web/app/server/workflow-server-actions.server.ts index 1387768f86..2780ee2528 100644 --- a/packages/web/app/server/workflow-server-actions.server.ts +++ b/packages/web/app/server/workflow-server-actions.server.ts @@ -1267,26 +1267,21 @@ export interface StreamChunksResult { * the next request. */ export async function readStreamChunksServerAction( - env: EnvMap, streamId: string, runId: string, - startCursor?: string + startCursor: string | undefined ): Promise { try { - const world = await getWorldFromEnv(env); + const world = await getWorldFromEnv({}); const allChunks: Uint8Array[] = []; - let pageCursor: string | undefined = startCursor; + let cursor: string | undefined = startCursor; let streamDone = false; - // Track the last non-null cursor so we can resume from the start of - // the final page on the next poll. When getChunks returns - // cursor=null we've exhausted all pages, but this saved cursor lets - // the client re-fetch only the last page + any new chunks. - let resumeCursor: string | null = startCursor ?? null; + let hasMore: boolean; do { const result = await world.streams.getChunks(runId, streamId, { limit: CHUNKS_PAGE_SIZE, - cursor: pageCursor, + cursor, }); for (const chunk of result.data) { @@ -1294,25 +1289,18 @@ export async function readStreamChunksServerAction( } streamDone = result.done; - if (result.cursor) { - resumeCursor = result.cursor; + hasMore = result.hasMore; + if (hasMore && !result.cursor) { + throw new Error('Stream chunk page with more data is missing a cursor'); } - pageCursor = result.cursor ?? undefined; - } while (pageCursor); - - let totalSize = 0; - for (const chunk of allChunks) { - totalSize += chunk.length; - } + cursor = result.cursor ?? cursor; + } while (hasMore); - const body = new Uint8Array(totalSize); - let offset = 0; - for (const chunk of allChunks) { - body.set(chunk, offset); - offset += chunk.length; - } - - return { buffer: body, cursor: resumeCursor, done: streamDone }; + return { + buffer: Buffer.concat(allChunks), + cursor: streamDone ? null : (cursor ?? null), + done: streamDone, + }; } catch (error) { const actionError = createServerActionError( error,