From b78e2fb0a1ea87e8df10de77244cd14cf33ae718 Mon Sep 17 00:00:00 2001 From: pdelfan Date: Thu, 15 Oct 2026 11:17:49 -0700 Subject: [PATCH] A reply's stream is reduced in batches of the events that arrive within one animation frame, so a reply streaming a word at a time renders, and Streamdown completes and splits its whole text, at most once a frame rather than once a word. The status event is passed on at once with whatever came before it, and a dropped or unreadable stream still keeps every event that arrived first. A thought or tool step renders its body only once it is first opened, and keeps it once closed, so a closed thought no longer parses every token streaming into it and a closed tool step no longer holds all the tool returned. The body still unfolds, since it arrives while the step grows from nothing. The web UI spec and the architecture overview say so, the stream tests cover the batches, and the tool view tests open each step before reading it. --- .../features/conversation/lib/reply-stream.ts | 32 ++++++++--- .../features/conversation/reply-follower.ts | 10 ++-- apps/web/src/parts/WorkSteps.tsx | 56 ++++++++++++++----- .../components/MessageView.test.tsx | 25 +++++++++ .../conversation/lib/reply-stream.test.ts | 38 ++++++++++++- docs/architecture.md | 2 +- specs/web-ui.md | 4 +- 7 files changed, 137 insertions(+), 30 deletions(-) diff --git a/apps/web/src/features/conversation/lib/reply-stream.ts b/apps/web/src/features/conversation/lib/reply-stream.ts index 4f0ca69..c77116c 100644 --- a/apps/web/src/features/conversation/lib/reply-stream.ts +++ b/apps/web/src/features/conversation/lib/reply-stream.ts @@ -6,32 +6,46 @@ import type { Part } from '../../../parts/types.ts' export type ReplyEvent = HubEvent /** - * A reply's events from its stream endpoint. It ends after the status event, and throws if the - * connection drops or sends an event it can't parse first. The connection closes when it ends or - * the signal aborts. + * A reply's events from its stream endpoint, in batches of those that arrived within one animation + * frame, so a reply streaming a word at a time rerenders at most once a frame. It ends after the + * status event, which is sent with its batch at once, and throws if the connection drops or sends + * an event it can't parse first. The connection closes when it ends or the signal aborts. */ -export async function* replyEvents(url: string, signal: AbortSignal): AsyncGenerator { +export async function* replyEvents(url: string, signal: AbortSignal): AsyncGenerator { const source = new EventSource(url) let failure: unknown - const events = new ReadableStream({ + let batch: ReplyEvent[] = [] + let frame: number | null = null + const events = new ReadableStream({ start(controller) { + const flush = () => { + if (frame !== null) cancelAnimationFrame(frame) + frame = null + if (batch.length) controller.enqueue(batch) + batch = [] + } // A failure closes the stream rather than erroring it, which would drop the events not yet read. const end = (error?: unknown) => { failure = error source.close() + flush() controller.close() } for (const type of ['part-start', 'delta', 'part', 'queued', 'status'] as const) source.addEventListener(type, (event) => { try { - controller.enqueue({ type, ...JSON.parse((event as MessageEvent).data) } as ReplyEvent) + batch.push({ type, ...JSON.parse((event as MessageEvent).data) } as ReplyEvent) } catch (error) { return end(error) } - if (type === 'status') end() + if (type === 'status') return end() + frame ??= requestAnimationFrame(flush) }) source.onerror = () => end(new Error('The reply stream dropped')) }, + cancel() { + if (frame !== null) cancelAnimationFrame(frame) + }, }) const reader = events.getReader() const cancel = () => void reader.cancel() @@ -78,6 +92,10 @@ export function addEvent(reply: ReplyParts, event: ReplyEvent): ReplyParts { } } +/** The reply with a batch of events added. */ +export const addEvents = (reply: ReplyParts, events: ReplyEvent[]): ReplyParts => + events.reduce(addEvent, reply) + /** The reply with one part replaced. A part arriving means the turn has left the queue. */ function setPart(reply: ReplyParts, index: number, part: Part): ReplyParts { const parts = [...reply.parts] diff --git a/apps/web/src/features/conversation/reply-follower.ts b/apps/web/src/features/conversation/reply-follower.ts index 8d3dcc3..b28a1e2 100644 --- a/apps/web/src/features/conversation/reply-follower.ts +++ b/apps/web/src/features/conversation/reply-follower.ts @@ -8,7 +8,7 @@ import { import { useState } from 'react' import { refreshConversation, useRefreshConversation } from '../../store/react.tsx' import type { BranchMessage } from './lib/branch.ts' -import { addEvent, noParts, type ReplyParts, replyEvents } from './lib/reply-stream.ts' +import { addEvents, noParts, type ReplyParts, replyEvents } from './lib/reply-stream.ts' /* * The reply follower: it follows each pending reply in the open conversation until its final @@ -30,13 +30,13 @@ const streamQuery = (skey: string, rkey: string) => queryFn: streamedQuery({ streamFn: async function* ({ client, signal }) { const url = `/api/conversations/${skey}/messages/${rkey}/stream` - for await (const event of replyEvents(url, signal)) { - yield event - if (event.type === 'status' && FINISHED.has(event.status)) + for await (const events of replyEvents(url, signal)) { + yield events + if (events.some((event) => event.type === 'status' && FINISHED.has(event.status))) void refreshConversation(client, skey) } }, - reducer: addEvent, + reducer: addEvents, initialValue: noParts, }), // A stream runs once. Leaving the conversation drops it, and coming back follows again. diff --git a/apps/web/src/parts/WorkSteps.tsx b/apps/web/src/parts/WorkSteps.tsx index 9293a3b..0deaa49 100644 --- a/apps/web/src/parts/WorkSteps.tsx +++ b/apps/web/src/parts/WorkSteps.tsx @@ -1,4 +1,4 @@ -import { memo, type ReactNode } from 'react' +import { memo, type ReactNode, type SyntheticEvent, useState } from 'react' import { cn } from '../shared/cn.ts' import { ChevronDownIcon, ChevronRightIcon } from '../shared/icons.tsx' import { action } from '../shared/ui/actions.ts' @@ -282,6 +282,22 @@ const StepView = memo( sameStep(prev.step, next.step), ) +/** + * Whether a step has been opened, so its body renders only once someone looks: a closed thought + * would otherwise parse every token streaming into it, and a closed tool step would hold all a tool + * returned. The body still unfolds, since it arrives while the step grows from nothing. Once + * opened, it stays, so closing folds it away rather than emptying it. + */ +function useOpened() { + const [opened, setOpened] = useState(false) + return { + opened, + onToggle: (event: SyntheticEvent) => { + if (event.currentTarget.open) setOpened(true) + }, + } +} + interface ThoughtProps { text: string /** Whether the model is still thinking it. */ @@ -298,11 +314,13 @@ interface ThoughtProps { * goes. */ function Thought(props: ThoughtProps) { + const step = useOpened() return (
@@ -312,11 +330,13 @@ function Thought(props: ThoughtProps) { {props.elapsed !== undefined && } - + {step.opened && ( + + )}
) } @@ -355,8 +375,14 @@ function ToolStep(props: ToolStepProps) { const { call, result } = props.step const entry = toolEntry(call.tool as string) const status = props.status + const step = useOpened() return ( -
+
{entry.code ? ( @@ -385,13 +411,15 @@ function ToolStep(props: ToolStepProps) { )} {props.elapsed !== undefined && } - + {step.opened && ( + + )}
) } diff --git a/apps/web/tests/features/conversation/components/MessageView.test.tsx b/apps/web/tests/features/conversation/components/MessageView.test.tsx index 54c695d..1924471 100644 --- a/apps/web/tests/features/conversation/components/MessageView.test.tsx +++ b/apps/web/tests/features/conversation/components/MessageView.test.tsx @@ -65,12 +65,29 @@ describe('MessageView', () => { expect(within(thought).queryByRole('button')).toBeNull() }) + it("renders a step's body only once it has been opened, and keeps it once closed", async () => { + const { container } = render() + await userEvent.click(screen.getByRole('button', { name: 'Thought · lookup' })) + const thought = container.querySelector('[data-step=reasoning]') as HTMLDetailsElement + const tool = container.querySelector('[data-step=tool]') as HTMLDetailsElement + expect(thought.children).toHaveLength(1) + expect(tool.querySelector('pre')).toBeNull() + const row = tool.querySelector('summary') as HTMLElement + await userEvent.click(row) + expect(tool.open).toBe(true) + expect(tool.querySelectorAll('pre')).toHaveLength(2) + await userEvent.click(row) + expect(tool.open).toBe(false) + expect(tool.querySelectorAll('pre')).toHaveLength(2) + }) + it('pairs a tool call with its result, showing the input at a glance and both in full', async () => { const { container } = render() await userEvent.click(screen.getByRole('button', { name: 'Thought · lookup' })) const tool = container.querySelector('[data-step=tool]') as HTMLDetailsElement expect(tool.querySelector('summary')).toHaveTextContent('lookup "tiles"') expect(tool.dataset.state).toBe('done') + await openToolSteps(container) const shown = Array.from(tool.querySelectorAll('pre')).map((pre) => pre.textContent) expect(shown).toEqual(['{\n "query": "tiles"\n}', '{\n "hits": 2\n}']) }) @@ -487,11 +504,18 @@ async function openSteps(parts: object[], status = 'complete') { const view = render() const footer = view.container.querySelector('footer') as HTMLElement await userEvent.click(within(footer).getByRole('button')) + await openToolSteps(view.container) return view } const toolSteps = (container: HTMLElement) => Array.from(container.querySelectorAll('[data-step=tool]')) +/** Open each tool step, whose view renders only once it has been opened. */ +async function openToolSteps(container: HTMLElement) { + for (const step of toolSteps(container)) + await userEvent.click(step.querySelector('summary') as HTMLElement) +} + describe('tool views', () => { it('shows every call still waiting for its result in a streaming reply as running', () => { const parts = [call('a', 'lookup', { q: 'one' }), call('b', 'lookup', { q: 'two' })] @@ -604,6 +628,7 @@ describe('tool views', () => { const live = render( , ) + await openToolSteps(live.container) expect(toolSteps(live.container)[0]).toHaveTextContent('Searching…') live.unmount() const { container } = await openSteps([search, result('a', { results: [] })]) diff --git a/apps/web/tests/features/conversation/lib/reply-stream.test.ts b/apps/web/tests/features/conversation/lib/reply-stream.test.ts index 6a49dde..3c15c6c 100644 --- a/apps/web/tests/features/conversation/lib/reply-stream.test.ts +++ b/apps/web/tests/features/conversation/lib/reply-stream.test.ts @@ -1,9 +1,12 @@ -import { describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { addEvent, + addEvents, noParts, type ReplyEvent, + replyEvents, } from '../../../../src/features/conversation/lib/reply-stream.ts' +import { lastSource, stubEventSource } from '../../../helpers/event-source.ts' const d = (name: string) => `network.sharedcomputer.chat.defs#${name}` @@ -37,3 +40,36 @@ describe('addEvent', () => { expect(addEvent(noParts, { type: 'status', status: 'error' }).status).toBe('error') }) }) + +describe('replyEvents', () => { + beforeEach(stubEventSource) + afterEach(() => vi.unstubAllGlobals()) + + it('sends the events that arrive within a frame as one batch, and the status at once', async () => { + const stream = replyEvents('/stream', new AbortController().signal) + const first = stream.next() + lastSource().emit('part-start', { index: 0, partType: 'textPart' }) + for (const letter of 'Hello') lastSource().emit('delta', { index: 0, text: letter }) + const batch = (await first).value as ReplyEvent[] + expect(batch).toHaveLength(6) + expect(addEvents(noParts, batch).parts[0]?.text).toBe('Hello') + const second = stream.next() + lastSource().emit('delta', { index: 0, text: '!' }) + lastSource().emit('status', { status: 'complete' }) + expect((await second).value).toEqual([ + { type: 'delta', index: 0, text: '!' }, + { type: 'status', status: 'complete' }, + ]) + expect((await stream.next()).done).toBe(true) + expect(lastSource().closed).toBe(true) + }) + + it('sends what arrived before the stream dropped, then throws', async () => { + const stream = replyEvents('/stream', new AbortController().signal) + const first = stream.next() + lastSource().emit('part-start', { index: 0, partType: 'textPart' }) + lastSource().onerror?.() + expect((await first).value).toEqual([{ type: 'part-start', index: 0, partType: 'textPart' }]) + await expect(stream.next()).rejects.toThrow('The reply stream dropped') + }) +}) diff --git a/docs/architecture.md b/docs/architecture.md index d606eb2..7545867 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -12,7 +12,7 @@ The web app uses Hono server, listening on port 3000 by default. It runs [TanSta Code is grouped by feature in `src/features/`, and `src/routes/` maps each URL to a feature's page. -Each route's loader fetches the server data its page needs, and the page reads it with `useSuspenseQuery`, so a page renders with its data and a failed load shows inside the page's layout with a Retry. Chats are the exception: they come from the browser's local copy in `src/store/`, not from loaders. Features reach that copy only through the store's hooks, such as `useRefreshConversation` and `useSyncIndex`, never its worker. The open conversation is `useChat` in `features/conversation/chat.ts`, shaped like the AI SDK's `useChat` but without a message list of its own: its messages come from the local copy, and the reply follower in `reply-follower.ts` shows each pending reply's stream until its record arrives. Message parts are drawn by the views in `src/parts/`, looked up by part type, and tool calls by the views in `src/parts/tools/`, looked up by tool name, and `src/shared/` holds what features share, like the form kit in `form.tsx` that every form is built with. +Each route's loader fetches the server data its page needs, and the page reads it with `useSuspenseQuery`, so a page renders with its data and a failed load shows inside the page's layout with a Retry. Chats are the exception: they come from the browser's local copy in `src/store/`, not from loaders. Features reach that copy only through the store's hooks, such as `useRefreshConversation` and `useSyncIndex`, never its worker. The open conversation is `useChat` in `features/conversation/chat.ts`, shaped like the AI SDK's `useChat` but without a message list of its own: its messages come from the local copy, and the reply follower in `reply-follower.ts` shows each pending reply's stream until its record arrives, rendering at most once a frame however fast the words come. Message parts are drawn by the views in `src/parts/`, looked up by part type, and tool calls by the views in `src/parts/tools/`, looked up by tool name, and `src/shared/` holds what features share, like the form kit in `form.tsx` that every form is built with. The look comes from one file of design tokens, `src/styles/theme.css`, and each component styles itself with Tailwind utilities that can only name those tokens, since the theme clears Tailwind's own palette and scales. A look used in more than one place is a component in `src/shared/ui/`, such as `Button` or the settings page kit. Light and dark follow the system, or a choice kept on the device in Preferences. Replies render through Streamdown, with code highlighted in the theme's colors. The styling spec has the rules. diff --git a/specs/web-ui.md b/specs/web-ui.md index ef49f8f..f378ef3 100644 --- a/specs/web-ui.md +++ b/specs/web-ui.md @@ -83,7 +83,7 @@ The root route's `beforeLoad` asks `/api/me` who is signed in, through the query - **Message.** A message shows its work as glass, not a black box: what the model did is visible as it happens, collapsed to a line by default, and every step can be opened down to its exact input and output. - Its header names its author: "You", or the model that wrote the reply, from the record's `model`. - Assistant text parts and reasoning are rendered as GitHub-flavored Markdown with Streamdown, as ADR 0005 explains. While a part streams, Streamdown completes its unfinished Markdown, so a `**` not yet closed already shows bold and a link still being typed shows as its text, and renders it block by block, so the blocks already written don't render again. Fenced code is highlighted by Shiki in the theme's colors, as it streams too, and a language Shiki doesn't know shows as plain text. Math between `$$` renders with KaTeX, loaded the first time a text holds some, and emphasis beside Chinese, Japanese, and Korean punctuation works. Raw HTML is not rendered and shows as the text it is, links open in a new tab, a fenced code block has buttons in its corner that copy its code and download it as a file named for its language, as `code.py`, once its fence closes, a table has menus in its corner that copy it as Markdown, CSV, or TSV and download it as CSV or Markdown, through Streamdown's table helpers with our own buttons and menu, and these show on hover and always on a touch screen, and images show their alt text, full URL, and a load button instead of loading, so injected content cannot leak the chat through an image URL without the user choosing to load it. User text is shown as plain text, except that runs of lines starting with `>`, as a quote from a reply sends them, show as quotes without their markers, in the same bubble as the text, through `splitQuotes` in `parts/quotes.ts`. Full Markdown for user text is planned. - - Parts show in the order the model produced them. Text, sources, images, and files show through `partViews` in `src/parts/index.ts`, and each run of reasoning and tool parts is one block of steps, drawn by `StepsView`, a `
` whose summary names the steps, such as "Thought · web_search ×2". A block of one step shows that step alone. A block shows only while the model is working in it, so once the reply's text starts streaming the work gives way to it. The block stays mounted but hidden, so its steps' times are still recorded. A finished reply keeps its steps behind a button at the end of its footer, named by their summary across every block, as "Thought for 12s · web_search", which opens them as one numbered list under the footer and keeps the footer shown while open. A reasoning step is a row like a tool step's, naming it, "Thought for 12s", or "Thinking" while it runs, with a muted glimpse of its first line of text, its heading, quote, and list marks and inline Markdown left out. It opens onto the whole reasoning as Markdown, and the glimpse goes while it is open. A tool call and its result, matched by `callId`, are one step whose summary shows the tool and a glimpse of its input, which goes while the step is open, and whose body shows the input and output pretty-printed when they are JSON, with a link to `outputBlob` when there is one. A failed tool or a call with no result is marked as such. + - Parts show in the order the model produced them. Text, sources, images, and files show through `partViews` in `src/parts/index.ts`, and each run of reasoning and tool parts is one block of steps, drawn by `StepsView`, a `
` whose summary names the steps, such as "Thought · web_search ×2". A block of one step shows that step alone. A block shows only while the model is working in it, so once the reply's text starts streaming the work gives way to it. The block stays mounted but hidden, so its steps' times are still recorded. A finished reply keeps its steps behind a button at the end of its footer, named by their summary across every block, as "Thought for 12s · web_search", which opens them as one numbered list under the footer and keeps the footer shown while open. A reasoning step is a row like a tool step's, naming it, "Thought for 12s", or "Thinking" while it runs, with a muted glimpse of its first line of text, its heading, quote, and list marks and inline Markdown left out. It opens onto the whole reasoning as Markdown, and the glimpse goes while it is open. A tool call and its result, matched by `callId`, are one step whose summary shows the tool and a glimpse of its input, which goes while the step is open, and whose body shows the input and output pretty-printed when they are JSON, with a link to `outputBlob` when there is one. A failed tool or a call with no result is marked as such. A step renders its body only once it is first opened, and keeps it once closed, so a closed thought doesn't parse every token streaming into it and a closed tool step doesn't hold all the tool returned. - The footer's steps and each step start closed, and the reader opens what they want to see. - Sources are shown after the parts as a numbered list under their count, such as "3 sources", so a reader sees at a glance what the reply drew on. Each is a link named by its title, followed by its site, or named by its site alone when it has no title, with the full URL on hover. A streaming reply's sources join the list as they arrive. Past seven sources, the rest fold behind a "+3 more" badge at the end of the list, which shows them in place and becomes "Show less", while the count still says how many there are in all. A list that would hide only one source shows all eight instead. Opening it moves focus to the first link it showed, and the list starts folded each time it is drawn. The search results in a web search step fold the same way, through the same `useFold` in `src/parts/fold.ts`. - Images are shown inline, and files as their names. @@ -121,7 +121,7 @@ The root route's `beforeLoad` asks `/api/me` who is signed in, through the query The reply follower, in `features/conversation/reply-follower.ts`, follows each pending reply in the open conversation until its final record arrives. The conversation follows its replies with `useReplyFollower(skey, messages)`, which follows every message whose record is pending, and returns `follow` for a reply just sent or regenerated, which it follows at once, before its pending record reaches the browser store. -Each followed reply has a stream query over the reply's stream endpoint, private to the follower. It reduces the stream's events into the reply's parts in the record's own shape, by index: `part-start` starts a text or reasoning part, `delta` appends to it, and `part` sets a finished tool call, tool result, or source. A `queued` event marks the reply as waiting until its first part arrives. The query ends after the `status` event. Since streamed parts have the record's shape, a pending reply and a finished one render through the same view. A finished status (`complete`, `error`, or `cancelled`) refreshes the conversation from the PDS, which brings the final record into the browser store. A stream runs once while the conversation is open. Leaving the conversation drops it, and coming back follows the reply again. +Each followed reply has a stream query over the reply's stream endpoint, private to the follower. It reduces the stream's events into the reply's parts in the record's own shape, by index: `part-start` starts a text or reasoning part, `delta` appends to it, and `part` sets a finished tool call, tool result, or source. A `queued` event marks the reply as waiting until its first part arrives. The query ends after the `status` event. Since streamed parts have the record's shape, a pending reply and a finished one render through the same view. A finished status (`complete`, `error`, or `cancelled`) refreshes the conversation from the PDS, which brings the final record into the browser store. The events that arrive within one animation frame are reduced as one batch, so a reply streaming a word at a time renders at most once a frame, and the `status` event is passed on at once with whatever came before it. A stream runs once while the conversation is open. Leaving the conversation drops it, and coming back follows the reply again. A reply's views read it from the follower without ever starting a stream, so each delta re-renders only the views that show it, never the conversation. `ReplyStream`, which `ConversationView` puts in `MessageView`'s `pending` slot, reads `useFollowedReply(skey, rkey)`: what has streamed so far, and why the reply is no longer live, if it isn't. `LiveAuthor` reads `useReplyStarted(skey, rkey)`, which re-renders the header only once the reply starts. A message with nothing in the slot, as on the shared page, shows "Thinking..." while pending. -- knotmirror 0.0.1