diff --git a/CHANGELOG.md b/CHANGELOG.md index 859426b..e370998 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ What changed in each release of SCN Chat, for the people who use it, run it, or ## [Unreleased] +### Fixed + +- A chat left open for a long time no longer keeps a copy of every reply it streamed in memory once the reply is saved. + ### Changed - For forkers: a tab reaches the browser store through `StoreClient.calls`, which has every method of the store's logic, rather than `StoreClient.worker`. Web tests run components over the real store through `testStore` in `apps/web/tests/helpers/store.ts`, rather than hand-built fakes. diff --git a/apps/web/src/features/conversation/reply-follower.ts b/apps/web/src/features/conversation/reply-follower.ts index de80b10..15640d0 100644 --- a/apps/web/src/features/conversation/reply-follower.ts +++ b/apps/web/src/features/conversation/reply-follower.ts @@ -58,7 +58,10 @@ export function useReplyFollower(skey: string, messages: BranchMessage[]) { const [followed, setFollowed] = useState([]) const follow = (rkey: string) => setFollowed((current) => (current.includes(rkey) ? current : [...current, rkey])) - const rkeys = [...new Set([...pending, ...followed])] + // A reply followed since it was sent, until the local copy has its final record, which the + // conversation shows instead. Its stream then goes, and what streamed with it. + const final = new Set(messages.filter((m) => !isPending(m.record)).map((m) => m.rkey)) + const rkeys = [...new Set([...pending, ...followed.filter((rkey) => !final.has(rkey))])] // Only whether each stream has ended, so a delta doesn't rerender the conversation. const ended = useQueries({ queries: rkeys.map((rkey) => streamQuery(skey, rkey)), diff --git a/apps/web/tests/features/conversation/reply-follower.test.tsx b/apps/web/tests/features/conversation/reply-follower.test.tsx index 8b90602..271de54 100644 --- a/apps/web/tests/features/conversation/reply-follower.test.tsx +++ b/apps/web/tests/features/conversation/reply-follower.test.tsx @@ -1,42 +1,33 @@ import { act, screen } from '@testing-library/react' -import { useState } from 'react' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import type { BranchMessage } from '../../../src/features/conversation/lib/branch.ts' +import { useChat } from '../../../src/features/conversation/chat.ts' import { useFollowedReply, - useReplyFollower, useReplyStarted, } from '../../../src/features/conversation/reply-follower.ts' -import { StoreProvider, useConversation } from '../../../src/store/react.tsx' +import { isPending } from '../../../src/shared/message.ts' +import { StoreProvider } from '../../../src/store/react.tsx' import { FakeEventSource, lastSource, stubEventSource } from '../../helpers/event-source.ts' +import { stubFetch } from '../../helpers/fetch.ts' import { messageRecord } from '../../helpers/records.ts' import { renderAt } from '../../helpers/router.tsx' -import { testStore } from '../../helpers/store.ts' +import { type HeldMessage, type TestStore, testStore } from '../../helpers/store.ts' -const user: BranchMessage = { rkey: 'u', record: messageRecord({ role: 'user' }) } -const reply = (status: string): BranchMessage => ({ - rkey: 'u.r0', +const user: HeldMessage = { rkey: 'u', record: messageRecord({ role: 'user' }) } +const reply = (status: string, rkey = 'u.r0'): HeldMessage => ({ + rkey, record: messageRecord({ parent: 'u', status }), }) -let follow: (rkey: string) => void -let setMessages: (messages: BranchMessage[]) => void +let chat: ReturnType const renders = { conversation: 0, header: 0 } -interface ConversationProps { - messages: BranchMessage[] -} - -/** The open conversation as the follower sees it: its replies, each with a header and its stream. */ -function Conversation(props: ConversationProps) { +/** The open conversation as the follower sees it: its pending replies, each with a header and its stream. */ +function Conversation() { renders.conversation++ - const [messages, set] = useState(props.messages) - setMessages = set - // The conversation's own refresh, which a finished reply sets off. - useConversation('s1') - follow = useReplyFollower('s1', messages).follow - return messages - .filter((message) => message.record.status === 'pending') + chat = useChat('s1') + return chat.messages + .filter((message) => isPending(message.record)) .map((message) => (
@@ -64,13 +55,15 @@ function Stream(props: ReplyProps) { ) } -async function open(messages: BranchMessage[]) { - // A store that holds no chats, the conversation's messages coming from the test. +/** Open chat `s1`, holding `messages` on this device and the PDS. */ +async function open(messages: HeldMessage[]) { const store = testStore() + store.held('s1', { messages }) const view = await renderAt( - + , + '/chat/s1', ) // The conversation has loaded and refreshed, so only the follower can rerender it now. await vi.waitFor(() => expect(store.api.fetchKeys).toHaveBeenCalled()) @@ -78,6 +71,10 @@ async function open(messages: BranchMessage[]) { return { store, view } } +/** The PDS now holds the reply as finished, for the next refresh to bring in. */ +const finish = (store: TestStore, status = 'complete', rkey = 'u.r0') => + store.remote('s1', { messages: [user, reply(status, rkey)] }) + /** Stream a text part saying `text`, a delta at a time. */ function streamText(text: string) { const source = lastSource() @@ -98,7 +95,7 @@ afterEach(() => { }) describe('reply follower', () => { - it("follows a pending reply's stream, and refreshes the conversation once it finishes", async () => { + it("follows a pending reply's stream, and brings in its record once it finishes", async () => { const { store } = await open([user, reply('pending')]) await vi.waitFor(() => expect(lastSource()).toBeDefined()) expect(lastSource().url).toBe('/api/conversations/s1/messages/u.r0/stream') @@ -106,11 +103,11 @@ describe('reply follower', () => { streamText('Hello') expect(await screen.findByText('Hello')).toBeInTheDocument() expect(screen.getByRole('heading')).toHaveTextContent('Started') - store.api.fetchConversation.mockClear() + finish(store) act(() => lastSource().emit('status', { status: 'complete' })) - await vi.waitFor(() => expect(store.api.fetchConversation).toHaveBeenCalled()) + // The refresh the finish sets off brings the final record, which ends the pending reply. + await vi.waitFor(() => expect(screen.queryByRole('heading')).toBeNull()) expect(lastSource().closed).toBe(true) - expect(screen.queryByText(/Not live/)).toBeNull() }) it('rerenders neither the conversation nor the header for the deltas after the reply starts', async () => { @@ -126,13 +123,31 @@ describe('reply follower', () => { expect(renders).toEqual(before) }) - it('follows a reply just sent, before its pending record arrives', async () => { - await open([user]) + it('follows a reply just sent, before its pending record arrives, until its record is final', async () => { + stubFetch(async () => + Response.json({ rkey: 'v', replyRkey: 'v.r0', status: 'claimed' }, { status: 201 }), + ) + const { store, view } = await open([user, reply('complete')]) expect(lastSource()).toBeUndefined() - act(() => follow('u.r0')) + await act(() => chat.send({ parent: 'u.r0', parts: [], generation: {} })) await vi.waitFor(() => - expect(lastSource().url).toBe('/api/conversations/s1/messages/u.r0/stream'), + expect(lastSource().url).toBe('/api/conversations/s1/messages/v.r0/stream'), + ) + const stream = () => + view.queryClient.getQueryCache().find({ queryKey: ['reply-stream', 's1', 'v.r0'] }) + expect(stream()).toBeDefined() + // Its record arrives finished, so what streamed of it goes. + act(() => + store.held('s1', { + messages: [ + user, + reply('complete'), + { rkey: 'v', record: messageRecord({ role: 'user', parent: 'u.r0' }) }, + { rkey: 'v.r0', record: messageRecord({ parent: 'v', status: 'complete' }) }, + ], + }), ) + await vi.waitFor(() => expect(stream()).toBeUndefined()) }) it('keeps what streamed when the stream drops, and polls with a backoff until the reply finishes', async () => { @@ -147,9 +162,13 @@ describe('reply follower', () => { // Two seconds, then four more. await act(async () => vi.advanceTimersByTimeAsync(6_100)) expect(store.api.fetchConversation).toHaveBeenCalledTimes(2) - act(() => setMessages([user, reply('complete')])) + // The next poll, eight seconds on, finds the reply finished, and polling stops. + finish(store) + await act(async () => vi.advanceTimersByTimeAsync(8_100)) + expect(store.api.fetchConversation).toHaveBeenCalledTimes(3) + expect(screen.queryByRole('heading')).toBeNull() await act(async () => vi.advanceTimersByTimeAsync(30_000)) - expect(store.api.fetchConversation).toHaveBeenCalledTimes(2) + expect(store.api.fetchConversation).toHaveBeenCalledTimes(3) }) it('treats an event it cannot read as a dropped stream', async () => { diff --git a/specs/web-ui.md b/specs/web-ui.md index 214c22b..6c3348d 100644 --- a/specs/web-ui.md +++ b/specs/web-ui.md @@ -43,7 +43,7 @@ The server process runs Hono, and Hono runs Start, so the server and plugins sta ### Data - **Server data.** Query wraps every typed-client call, keyed by route, such as `['models']` or `['preferences']`, from the factories in each feature's `queries.ts`. Writes are mutations that invalidate the keys they change. A 401 from any call ends the session as before. Queries do not retry or refetch on window focus, so failures show at once, as they did before Query. Admin data is keyed under `['admin']`, such as `['admin', 'roles']`, and the users list is an infinite query, `['admin', 'users', q]`, paged by the server's cursor. -- **Chats.** The chat list is the query `['conversations']`, a conversation is `['conversation', skey]`, and a search is `['search', q]`, all read from the browser store's worker. `StoreProvider` turns the store's change events into invalidations: an `index` change invalidates the chat list, a `conversation` change invalidates that conversation, and any change invalidates searches. Queries wait while the store is held by another tab. Opening a conversation also checks it against the PDS through the query `['conversation', skey, 'refresh']`, which fetches what changed, then reconciles its record keys so nothing missed stays missing, and whose changes come back as store events. `useConversation` gives the conversation with `waiting`, while its messages are on their way, from the local copy or, when it has none, from the PDS, and `empty`, once the PDS has confirmed it has none. The store's React module keeps its query keys and its worker to itself, and offers intents: `useRefreshConversation` checks an open conversation against the PDS again, as opening it does, and is what sending, a conversation action, and the reply follower use, and `useSyncIndex` syncs the chat index, for the chat list and the sync status. Only the store's own queries use the `conversation` prefix. +- **Chats.** The chat list is the query `['conversations']`, a conversation is `['conversation', skey]`, and a search is `['search', q]`, all read from the browser store's worker. `StoreProvider` turns the store's change events into invalidations: an `index` change invalidates the chat list, a `conversation` change invalidates that conversation, and any change invalidates searches. Queries wait while the store is held by another tab. Opening a conversation also checks it against the PDS through the query `['conversation', skey, 'refresh']`, which fetches what changed, then reconciles its record keys so nothing missed stays missing, and whose changes come back as store events. `useConversation` gives the conversation with `waiting`, while its messages are on their way, from the local copy or, when it has none, from the PDS, and `empty`, once the PDS has confirmed it has none. The store's React module keeps its query keys and its worker to itself, and offers intents: `useRefreshConversation` checks an open conversation against the PDS again, as opening it does, and is what sending, a conversation action, and the reply follower use, and `useSyncIndex` syncs the chat index, for the chat list and the sync status. The reply follower refreshes through `useRefreshConversation` rather than a query of its own, so the store's hooks know nothing of reply streams, and only a conversation that is open is checked; keep it that way rather than merging the follower into the store's hooks. Only the store's own queries use the `conversation` prefix. - **URL state.** On a conversation, and on a shared conversation, `m` names a focused message, and the branch on screen is its ancestors, then the newest sibling at each level below it. Choosing a sibling, or starting a turn by sending, editing, or regenerating, replaces `m` with the chosen or new message, so the URL always names the branch on screen, and reloading or sharing the link keeps it. On every chat route, `q` holds the sidebar search: the box starts from it, and the search goes back into it once typing pauses. The chat layout validates `q` and keeps it on every navigation between chat routes with the router's `retainSearchParams`, so opening a result, another chat, or a new chat keeps the search until it is cleared. On `/admin/users`, `q` holds the users search, set when the search is submitted. Changes to `m` and `q` replace the history entry rather than adding one, and the routes validate them in `src/shared/search-params.ts`. ### Routes @@ -123,7 +123,7 @@ The reply follower, in `features/conversation/reply-follower.ts`, follows each p 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. A pending reply is drawn by `PendingReply.tsx`, which `MessageView` uses for its header and body while the record is `pending`, and which finds the conversation to follow through the `FollowedConversation` context that `ConversationView` provides. Its body reads `useFollowedReply(skey, rkey)`: what has streamed so far, and why the reply is no longer live, if it isn't. Its header reads `useReplyStarted(skey, rkey)`, which re-renders the header only once the reply starts. The stream's part views get the pending record itself. Without that context, as on the shared page, a pending reply shows only its model's name breathing. A change to the conversation, such as a message arriving or a reply finishing, renders again only the messages it changed: `BranchView` gives each message's row only its record, which the local copy keeps the same object while unchanged, its place among its versions, and functions that stay the same, and the reply running at the end of the branch reaches the Stop actions through the `RunningReply` context. +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. A pending reply is drawn by `PendingReply.tsx`, which `MessageView` uses for its header and body while the record is `pending`, and which finds the conversation to follow through the `FollowedConversation` context that `ConversationView` provides. Its body reads `useFollowedReply(skey, rkey)`: what has streamed so far, and why the reply is no longer live, if it isn't. Its header reads `useReplyStarted(skey, rkey)`, which re-renders the header only once the reply starts. The stream's part views get the pending record itself. Without that context, as on the shared page, a pending reply shows only its model's name breathing. A reply followed since it was sent stops being followed once the local copy holds its final record, so what streamed of it isn't kept beside the record. A change to the conversation, such as a message arriving or a reply finishing, renders again only the messages it changed: `BranchView` gives each message's row only its record, which the local copy keeps the same object while unchanged, its place among its versions, and functions that stay the same, and the reply running at the end of the branch reaches the Stop actions through the `RunningReply` context. If a reply is still pending after its stream ends, because the stream dropped or the status says another server runs it, the follower polls, checking the conversation against the PDS as opening it does, with a backoff from two seconds to thirty until the reply leaves `pending`. The first poll comes one interval after the stream ends, and polling continues while the tab is in the background. Meanwhile the pending reply keeps what streamed and says why the reply is no longer live: "Lost the live stream. Checking for the reply…" when the stream dropped or sent an event it couldn't read, and "Not streaming on this server. Checking for the reply…" when its status was not final.