diff --git a/docs/performance.md b/docs/performance.md index 904924dea..e7efbca1d 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -327,7 +327,12 @@ when the core advertises `chunkedAnswers`. - A first visit with a saved reading position (an anchor message, not pinned to the bottom, no cached visit) asks `threads.get` for the page `around` that message on a `readingPages` core: 20 messages before it and 20 from - it, when more than 40 follow it. `messagesAfter` is the cursor below that + it, when more than 40 follow it. Both halves share the 12 MiB page budget; + larger windows drop their farthest messages while retaining the anchor + and cursors in both directions. Each cut cursor names the retained edge; + a single complete anchor may exceed the byte budget. The requested message + count also applies to the combined window, including a limit of one. + `messagesAfter` is the cursor below that window; `messages.list` with `after` pages down as the reader nears the bottom, and "Jump to latest" replaces the window with the last page. Live messages wait for the pages below, except a prompt, which brings the last diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 4dba563b0..6079985db 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -1335,6 +1335,7 @@ export const MESSAGE_PAGE = 120; export const INITIAL_MESSAGE_PAGE = 40; export { previewToolOutputs, previewToolPart, inputPreview, longerThan, TOOL_OUTPUT_INLINE_CHARS, TOOL_OUTPUT_PREVIEW_CHARS, TOOL_INPUT_INLINE_CHARS, TOOL_INPUT_PREVIEW_CHARS, TOOL_DOCUMENTS_INLINE_CHARS } from './message-preview'; export { forTransport, projectMessage, MESSAGE_SENT_MAX_BYTES, type TransportOptions } from './transport.ts'; +export { boundedMessageWindow } from './message-window.ts'; export { lastAgentText, notificationExcerpt, requestExcerpt, NOTIFICATION_TEXT_CHARS } from './notification-text'; /** * The core's generic notification body, named so a phone can show it in the @@ -1356,7 +1357,8 @@ export interface PushPayload { export const MESSAGE_PAGE_MAX = 200; /** * Serialized UTF-8 bytes of a page's message array, including its brackets and - * commas. One complete message may exceed this budget to advance pagination, + * commas, shared by both halves of a page around a saved reading position. + * One complete message may exceed this budget to advance pagination, * provided the full RPC response still fits `RPC_MAX_FRAME_BYTES`. A reconnect * tail must fit this budget in full; otherwise `threads.get` returns a page. */ diff --git a/packages/contracts/src/message-window.ts b/packages/contracts/src/message-window.ts new file mode 100644 index 000000000..342763e47 --- /dev/null +++ b/packages/contracts/src/message-window.ts @@ -0,0 +1,23 @@ +import type { Message } from './index.ts'; +import { utf8Bytes } from './transport.ts'; + +/** + * A contiguous slice around the anchor, measured as the client receives it. + * Drop the farthest edge until the two halves share one page's byte and + * message budgets. Keep a single complete anchor even when it alone exceeds + * the page budget, as ordinary pages do. Callers provide the index of an + * existing anchor and preserve cursors for either edge that was cut. + */ +export function boundedMessageWindow(messages: Message[], anchorIndex: number, maxBytes: number, maxMessages: number): { start: number; end: number } { + let start = 0, end = messages.length; + const sizes = messages.map(message => utf8Bytes(JSON.stringify(message))); + let bytes = 2 + sizes.reduce((sum, size) => sum + size, 0) + Math.max(0, end - 1); + while ((bytes > maxBytes || end - start > maxMessages) && end - start > 1) { + const older = anchorIndex - start; + const newer = end - 1 - anchorIndex; + const removed = older > newer ? start++ : --end; + // Removing an edge from at least two messages removes exactly one comma. + bytes -= sizes[removed]! + 1; + } + return { start, end }; +} diff --git a/packages/core/src/threads/snapshot.ts b/packages/core/src/threads/snapshot.ts index 064eebf39..57d41cae6 100644 --- a/packages/core/src/threads/snapshot.ts +++ b/packages/core/src/threads/snapshot.ts @@ -1,5 +1,5 @@ import { createHash } from 'node:crypto'; -import { MESSAGE_PAGE, MESSAGE_PAGE_MAX, resumeAnchor, snapshotOptionsProblem } from '@boite/contracts'; +import { MESSAGE_PAGE, MESSAGE_PAGE_MAX, MESSAGE_PAGE_MAX_BYTES, boundedMessageWindow, resumeAnchor, snapshotOptionsProblem } from '@boite/contracts'; import type { Message, RpcParams, Thread, ThreadSummary, Turn } from '@boite/contracts'; import type { Core } from '../core'; import { invalidParams, refused } from '../errors'; @@ -69,6 +69,8 @@ export function threadSnapshot(core: Core, thread: ThreadSummary, after: string * The last page, unless `around` names a message with more than `limit` * written after it: then half a page before it and half from it on. A count * on the index decides, so the far case reads no row it does not send. + * Both halves share MESSAGE_PAGE_MAX_BYTES and the requested count; cuts + * retain the anchor and rewrite cursors to the retained edges. */ function pageAround(core: Core, threadId: string, around: string | undefined, limit: number, project: Projection, toolPreviews: boolean): { messages: Message[]; sent: Message[]; before: string | null; after: string | null } { const journal = core.journal; @@ -77,7 +79,13 @@ function pageAround(core: Core, threadId: string, around: string | undefined, li const half = Math.max(1, Math.floor(limit / 2)); const older = journal.listMessagePage(threadId, { beforeRowid: rowid, limit: half, project, toolPreviews }); const newer = journal.listMessagesForward(threadId, rowid, Math.max(1, limit - half), project, toolPreviews); - return { messages: [...older.messages, ...newer.messages], sent: [...older.sent, ...newer.sent], before: older.before, after: newer.after }; + const messages = [...older.messages, ...newer.messages], sent = [...older.sent, ...newer.sent]; + const { start, end } = boundedMessageWindow(sent, older.messages.length, MESSAGE_PAGE_MAX_BYTES, limit); + return { + messages: messages.slice(start, end), sent: sent.slice(start, end), + before: start > 0 ? messages[start]!.id : older.before, + after: end < messages.length ? messages[end - 1]!.id : newer.after, + }; } /** A page of `messages.list`, with the turns its messages belong to. */ diff --git a/packages/core/test/message-window.test.ts b/packages/core/test/message-window.test.ts new file mode 100644 index 000000000..30d791918 --- /dev/null +++ b/packages/core/test/message-window.test.ts @@ -0,0 +1,16 @@ +import { expect, test } from 'bun:test'; +import { boundedMessageWindow, type Message } from '@boite/contracts'; + +test('a centred page counts empty object syntax in tool inputs against its byte budget', () => { + const messages: Message[] = ['older', 'anchor', 'newer'].map(id => ({ + id, threadId: 'thread', turnId: 'turn', role: 'assistant', state: 'complete', createdAt: 0, + parts: [{ type: 'tool', toolId: id, name: 'Read', status: 'done', output: null, input: Array.from({ length: 5000 }, () => ({})) }], + })); + const budget = 40_000; + expect(Buffer.byteLength(JSON.stringify(messages))).toBeGreaterThan(budget); + const { start, end } = boundedMessageWindow(messages, 1, budget, messages.length); + const page = messages.slice(start, end); + expect(Buffer.byteLength(JSON.stringify(page))).toBeLessThanOrEqual(budget); + expect(page.map(message => message.id)).toContain('anchor'); + expect(page).toHaveLength(2); +}); diff --git a/packages/core/test/messages.test.ts b/packages/core/test/messages.test.ts index 1e843c111..487c2d430 100644 --- a/packages/core/test/messages.test.ts +++ b/packages/core/test/messages.test.ts @@ -312,6 +312,11 @@ describe('opening where the reader was', () => { expect(thread.messagesBefore).toBe(ids(thread.messages)[0]!); expect(thread.messagesAfter).toBe(ids(thread.messages).at(-1)!); + const single = await client.call('threads.get', { threadId, around: 'msg_0100', limit: 1 }); + expect(ids(single.messages)).toEqual(['msg_0100']); + expect(single.messagesBefore).toBe('msg_0100'); + expect(single.messagesAfter).toBe('msg_0100'); + const walked = [...ids(thread.messages)]; let cursor = thread.messagesAfter ?? null; while (cursor !== null) { diff --git a/packages/core/test/thread-history.test.ts b/packages/core/test/thread-history.test.ts index 070fd48f4..cd696a46d 100644 --- a/packages/core/test/thread-history.test.ts +++ b/packages/core/test/thread-history.test.ts @@ -1,7 +1,74 @@ import { expect, test } from 'bun:test'; -import { INITIAL_MESSAGE_PAGE, MESSAGE_SENT_MAX_BYTES, RPC_MAX_FRAME_BYTES, TOOL_OUTPUT_PREVIEW_CHARS, type Message } from '@boite/contracts'; +import { INITIAL_MESSAGE_PAGE, MESSAGE_PAGE_MAX_BYTES, MESSAGE_SENT_MAX_BYTES, RPC_MAX_FRAME_BYTES, TOOL_OUTPUT_PREVIEW_CHARS, type Message } from '@boite/contracts'; import { echoThread, startTestCore } from './harness'; +test('a saved reading position shares one byte budget across both halves and keeps every history cursor', async () => { + const harness = await startTestCore(); + try { + const client = await harness.connect(); + const { threadId } = await echoThread(harness, client); + const text = '\n'.repeat(2 * 1024 * 1024); + const ids = Array.from({ length: 12 }, (_, index) => `heavy-${index}`); + harness.core.journal.append({ type: 'message.started', threadId, version: 1, payload: {} }, () => { + for (const [index, id] of ids.entries()) harness.core.journal.putMessage({ + id, threadId, turnId: 'history-turn', role: 'assistant', state: 'complete', createdAt: index, + parts: [{ type: 'text', text }], + }); + }); + const options = { compactToolParts: true, limit: 6 }; + const opened = await client.call('threads.get', { threadId, around: 'heavy-5', ...options, open: {} }); + expect(opened.messages.some(message => message.id === 'heavy-5')).toBe(true); + expect(Buffer.byteLength(JSON.stringify(opened.messages))).toBeLessThanOrEqual(MESSAGE_PAGE_MAX_BYTES); + expect(opened.messagesBefore).not.toBeNull(); + expect(opened.messagesAfter).toBeDefined(); + const complete = (messages: Message[]) => { + for (const message of messages) expect(message.parts[0]).toEqual({ type: 'text', text }); + }; + complete(opened.messages); + const walked = opened.messages.map(message => message.id); + let pages = 0; + for (let before = opened.messagesBefore; before;) { + const page = await client.call('messages.list', { threadId, before, ...options }); + expect(page.messages.length).toBeGreaterThan(0); + complete(page.messages); + expect(page.before).not.toBe(before); + expect(++pages).toBeLessThan(ids.length); + walked.unshift(...page.messages.map(message => message.id)); + before = page.before; + } + for (let after = opened.messagesAfter; after;) { + const page = await client.call('messages.list', { threadId, after, ...options }); + expect(page.messages.length).toBeGreaterThan(0); + complete(page.messages); + expect(page.after).not.toBe(after); + expect(++pages).toBeLessThan(ids.length); + walked.push(...page.messages.map(message => message.id)); + after = page.after ?? undefined; + } + expect(walked).toEqual(ids); + expect(harness.core.journal.getMessage('heavy-5')!.parts[0]).toEqual({ type: 'text', text }); + + // Ordinary pages allow one legal message above 12 MiB. A centred page + // must keep that anchor alone rather than dropping it or adding a neighbour. + const singleThread = await echoThread(harness, client); + const largeText = 'x'.repeat(13 * 1024 * 1024); + harness.core.journal.append({ type: 'message.started', threadId: singleThread.threadId, version: 1, payload: {} }, () => { + for (let index = 0; index < 12; index++) harness.core.journal.putMessage({ + id: `single-${index}`, threadId: singleThread.threadId, turnId: 'single-turn', role: 'assistant', state: 'complete', createdAt: index, + parts: [{ type: 'text', text: index === 5 ? largeText : 'neighbour' }], + }); + }); + const single = await client.call('threads.get', { threadId: singleThread.threadId, around: 'single-5', limit: 6 }); + expect(single.messages.map(message => message.id)).toEqual(['single-5']); + expect(single.messagesBefore).toBe('single-5'); + expect(single.messagesAfter).toBe('single-5'); + expect(single.messages[0]!.parts[0]).toEqual({ type: 'text', text: largeText }); + const bytes = Buffer.byteLength(JSON.stringify(single.messages)); + expect(bytes).toBeGreaterThan(MESSAGE_PAGE_MAX_BYTES); + expect(bytes).toBeLessThan(RPC_MAX_FRAME_BYTES); + } finally { await harness.stop(); } +}); + test('a compact first page keeps complete history and retrieves tool output on demand over RPC', async () => { const harness = await startTestCore(); try { diff --git a/packages/ui/src/lib/fake-client.test.ts b/packages/ui/src/lib/fake-client.test.ts index cfca1796a..bb6b308f5 100644 --- a/packages/ui/src/lib/fake-client.test.ts +++ b/packages/ui/src/lib/fake-client.test.ts @@ -4,6 +4,8 @@ import type { FakeClient } from './fake-client'; import { ATTACHMENT_MAX_BYTES, DEFAULT_DELEGATION_CONFIG, MESSAGE_PAGE_MAX_BYTES, MESSAGE_SENT_MAX_BYTES, RPC_MAX_FRAME_BYTES, RpcErrorCode, SPEECH_DEFAULT_MODEL, TODO_TEXT_MAX, type RpcEvents, type RpcMethodName, type Turn } from '@boite/contracts'; import { FAKE_AUTO_COMPACT_SETTLE_MS } from './fake-client/turns'; import { FakeContext } from './fake-client/context'; +import { seed } from './fake-client/seed'; +import { threadMethods } from './fake-client/threads'; test('done expiry mirrors the core while manual archives and restored history remain recoverable', async ({ createClient }) => { const client = await createClient({ delayMs: 0 }); @@ -117,11 +119,65 @@ test('fake byte-bounded pages walk complete escaped UTF-8 messages and fall back } expect(walked).toEqual(expected); expect(pages).toBeGreaterThan(1); + + // A saved position combines two independently bounded halves. The result + // must share their budget, and loading its newer cursor must do the same. + expect(expected).toHaveLength(8); + const around = await client.call('threads.get', { threadId: thread.id, around: expected[3], limit: 4 }); + expect(around.messages.map(message => message.id)).toContain(expected[3]); + expect(new TextEncoder().encode(JSON.stringify(around.messages)).byteLength).toBeLessThanOrEqual(MESSAGE_PAGE_MAX_BYTES); + const window = around.messages.map(message => message.id); + pages = 0; + for (let before = around.messagesBefore; before;) { + const page = await client.call('messages.list', { threadId: thread.id, before, limit: 4 }); + expect(page.messages.length).toBeGreaterThan(0); + expect(new TextEncoder().encode(JSON.stringify(page.messages)).byteLength).toBeLessThanOrEqual(MESSAGE_PAGE_MAX_BYTES); + expect(page.before).not.toBe(before); + expect(++pages).toBeLessThan(10); + window.unshift(...page.messages.map(message => message.id)); + before = page.before; + } + for (let after = around.messagesAfter; after;) { + const page = await client.call('messages.list', { threadId: thread.id, after, limit: 4 }); + expect(page.messages.length).toBeGreaterThan(0); + expect(new TextEncoder().encode(JSON.stringify(page.messages)).byteLength).toBeLessThanOrEqual(MESSAGE_PAGE_MAX_BYTES); + expect(page.after).not.toBe(after); + expect(++pages).toBeLessThan(10); + window.push(...page.messages.map(message => message.id)); + after = page.after ?? undefined; + } + expect(window).toEqual(expected); + const single = await client.call('threads.get', { threadId: thread.id, around: expected[3], limit: 1 }); + expect(single.messages.map(message => message.id)).toEqual([expected[3]]); + expect(single.messagesBefore).toBe(expected[3]); + expect(single.messagesAfter).toBe(expected[3]); +}); + +test('fake history refuses an oversized next message before the page byte boundary, like the core', async () => { + const ctx = new FakeContext({ delayMs: 0 }); + seed(ctx); + const thread = ctx.thread('t-parser'); + const base = thread.messages[0]!; + thread.turns = []; + const part = (id: string, text: string) => ({ ...base, id, parts: [{ type: 'text' as const, text }] }); + const cursor = part('cursor', 'cursor'); + const small = part('small', 'x'.repeat(7 * 1024 * 1024)); + const oversized = part('oversized', 'x'.repeat(RPC_MAX_FRAME_BYTES)); + const methods = threadMethods(ctx); + thread.messages = [cursor, small, oversized]; + await expect(methods['messages.list']({ threadId: thread.id, after: cursor.id, limit: 2 }).then(() => null, error => error)) + .resolves.toMatchObject({ code: RpcErrorCode.Refused, data: { field: 'messages', messageId: oversized.id } }); + thread.messages = [oversized, small, cursor]; + await expect(methods['messages.list']({ threadId: thread.id, before: cursor.id, limit: 2 }).then(() => null, error => error)) + .resolves.toMatchObject({ code: RpcErrorCode.Refused, data: { field: 'messages', messageId: oversized.id } }); }); test('fake pages keep a complete legal attachment bundle above the byte budget and refuse an oversized message', async ({ createClient }) => { const client = await createClient({ delayMs: 0 }); const thread = await client.call('threads.create', { projectId: 'p-boite', providerId: 'echo', accountId: 'a-echo', title: 'Single message' }); + await client.call('turns.start', { threadId: thread.id, prompt: 'Before the bundle' }); + await client.settled(); + const prefix = await client.call('threads.get', { threadId: thread.id }); const data = 'A'.repeat(Math.ceil(5 * 1024 * 1024 / 3) * 4 - 1) + '='; await client.call('turns.start', { threadId: thread.id, prompt: 'Review the bundle', attachments: ['first.bin', 'second.bin'].map(name => ({ kind: 'file', name, mimeType: 'application/octet-stream', data })) }); await client.settled(); @@ -132,8 +188,21 @@ test('fake pages keep a complete legal attachment bundle above the byte budget a const bytes = new TextEncoder().encode(JSON.stringify(page.messages)).byteLength; expect(bytes).toBeGreaterThan(MESSAGE_PAGE_MAX_BYTES); expect(bytes).toBeLessThan(RPC_MAX_FRAME_BYTES); - expect(page.before).toBeNull(); + expect(page.before).toBe(page.messages[0]!.id); + const older = await client.call('messages.list', { threadId: thread.id, before: page.before! }); + expect(older.messages.map(message => message.id)).toEqual(prefix.messages.map(message => message.id)); + expect(older.before).toBeNull(); expect(page.messages[0]?.parts.filter(part => part.type === 'file').every(part => part.data === data)).toBe(true); + for (let index = 0; index < 4; index++) { + await client.call('turns.start', { threadId: thread.id, prompt: `After the bundle ${index}` }); + await client.settled(); + } + const anchor = page.messages[0]!.id; + const centred = await client.call('threads.get', { threadId: thread.id, around: anchor, limit: 4 }); + expect(centred.messages.map(message => message.id)).toEqual([anchor]); + expect(centred.messages[0]!.parts).toEqual(page.messages[0]!.parts); + expect(centred.messagesBefore).toBe(anchor); + expect(centred.messagesAfter).toBe(anchor); await client.call('turns.start', { threadId: thread.id, prompt: 'x'.repeat(RPC_MAX_FRAME_BYTES) }); await client.settled(); await expect(client.call('threads.get', { threadId: thread.id })).rejects.toMatchObject({ code: RpcErrorCode.Refused, data: { field: 'messages', expected: `a complete message below ${RPC_MAX_FRAME_BYTES} serialized UTF-8 bytes` } }); diff --git a/packages/ui/src/lib/fake-client/threads.ts b/packages/ui/src/lib/fake-client/threads.ts index eaaf34354..664c92205 100644 --- a/packages/ui/src/lib/fake-client/threads.ts +++ b/packages/ui/src/lib/fake-client/threads.ts @@ -15,7 +15,7 @@ import { modelsOf, checkSpeed, discoverSelection } from './provider-catalog'; import { delegationConfig, stopDelegation } from './delegation'; import type { FakeContext, FakeMethods } from './context'; import { registerFakeWorktree, requireFakeCwd } from './worktrees'; -import { forTransport, projectMessage, resumeAnchor, snapshotOptionsProblem, type TransportOptions } from '@boite/contracts'; +import { boundedMessageWindow, projectMessage, resumeAnchor, snapshotOptionsProblem, type TransportOptions } from '@boite/contracts'; import { dropWaitingMove, fakeMoveNote } from './thread-move'; import { cancelFamilySideQuestions, cancelSide, sideQuestionMethods } from './side-questions'; @@ -83,12 +83,12 @@ function pageOf( const message = messages[start - 1]!; const size = sentBytes(message, project); const next = bytes + size + (start < end ? 1 : 0); - if (start < end && next > MESSAGE_PAGE_MAX_BYTES) break; if (size >= RPC_MAX_FRAME_BYTES) { throw new RpcFailure({ code: RpcErrorCode.Refused, message: `message ${message.id} is ${size} serialized UTF-8 bytes; expected a complete message below ${RPC_MAX_FRAME_BYTES} bytes`, data: { threadId: message.threadId, messageId: message.id, field: 'messages', bytes: size, max: RPC_MAX_FRAME_BYTES, expected: `a complete message below ${RPC_MAX_FRAME_BYTES} serialized UTF-8 bytes` } }); } + if (start < end && next > MESSAGE_PAGE_MAX_BYTES) break; bytes = next; start -= 1; } @@ -96,27 +96,50 @@ function pageOf( return { messages: page, before: start > 0 ? (page[0]?.id ?? null) : null }; } -/** The core's page from `from` on, at most `limit`, and the cursor below it while more follow. */ -function forwardOf(messages: Message[], from: number, limit: number): { messages: Message[]; after: string | null } { - const page = messages.slice(from, from + limit); - return { messages: page, after: from + limit < messages.length ? (page.at(-1)?.id ?? null) : null }; +/** + * The core's forward page: at most `limit` within the byte budget, except for + * one complete legal message. The cursor names its retained edge while more follow. + */ +function forwardOf(messages: Message[], from: number, limit: number, project: Projection): { messages: Message[]; after: string | null } { + // Reuse the backward page's byte and single-message rules, iterating from + // the first newer message instead of from the last older one. + const candidates = messages.slice(from, from + limit).reverse(); + const page = pageOf(candidates, candidates.length, limit, project).messages.reverse(); + return { messages: page, after: from + page.length < messages.length ? (page.at(-1)?.id ?? null) : null }; } -/** The core's page around a reading position: half a page above `at`, half from it on, once more than a page follows it. */ +/** + * Half a page on either side of the anchor when more than a page follows it. + * Both halves share the byte and count budgets, retaining a complete single + * anchor when needed; cuts rewrite cursors to the retained edges. + */ function aroundOf(messages: Message[], around: string | undefined, limit: number, project: Projection): { messages: Message[]; before: string | null; after: string | null } | null { const at = around === undefined ? -1 : messages.findIndex(message => message.id === around); if (at < 0 || messages.length - at <= limit) return null; const half = Math.max(1, Math.floor(limit / 2)); const older = pageOf(messages, at, half, project); - const newer = forwardOf(messages, at, Math.max(1, limit - half)); - return { messages: [...older.messages, ...newer.messages], before: older.before, after: newer.after }; + const newer = forwardOf(messages, at, Math.max(1, limit - half), project); + const candidates = [...older.messages, ...newer.messages]; + const { start, end } = boundedMessageWindow(candidates.map(project), older.messages.length, MESSAGE_PAGE_MAX_BYTES, limit); + return { + messages: candidates.slice(start, end), + before: start > 0 ? candidates[start]!.id : older.before, + after: end < candidates.length ? candidates[end - 1]!.id : newer.after, + }; } type Projection = (message: Message) => Message; /** The core's rule: page budgets measure a message as the client receives it. */ function projection(options: TransportOptions): Projection { - return message => projectMessage(message, options); + const cache = new WeakMap(); + return message => { + const held = cache.get(message); + if (held !== undefined) return held; + const sent = projectMessage(message, options); + cache.set(message, sent); + return sent; + }; } /** As the core counts it: a read with no projection is internal and never measured. */ @@ -353,7 +376,7 @@ export function threadMethods(ctx: FakeContext) { const proof = anchor === null ? undefined : { from: anchor, hash: snapshotHash(page.messages.slice(page.messages.findIndex(message => message.id === anchor)), params) }; const known = params.sync && params.sync !== true ? params.sync : undefined; const unchanged = tail !== null && params.after !== undefined && known?.from === params.after && known.hash === (proof?.from === params.after ? proof.hash : snapshotHash(tail, params)); - const snapshot = { ...thread, turns, messages: unchanged ? [] : forTransport(page.messages, params), messagesBefore: page.before, + const snapshot = { ...thread, turns, messages: unchanged ? [] : page.messages.map(project), messagesBefore: page.before, ...(page.after === null ? {} : { messagesAfter: page.after }), ...(tail !== null ? { messagesFrom: params.after } : {}), ...(proof ? { messagesSync: proof } : {}), ...(unchanged ? { messagesUnchanged: true as const } : {}) }; if (!params.open) return pagingReply(snapshot); @@ -382,8 +405,9 @@ export function threadMethods(ctx: FakeContext) { } const asked = params.limit ?? MESSAGE_PAGE; const limit = Math.min(Math.max(1, Math.trunc(asked)), MESSAGE_PAGE_MAX); - const page = params.before === undefined ? { ...forwardOf(thread.messages, at + 1, limit), before: null } : pageOf(thread.messages, at, limit, projection(params)); - page.messages = forTransport(page.messages, params); + const project = projection(params); + const page = params.before === undefined ? { ...forwardOf(thread.messages, at + 1, limit, project), before: null } : pageOf(thread.messages, at, limit, project); + page.messages = page.messages.map(project); const turns = new Set(page.messages.map((message) => message.turnId)); return pagingReply({ ...page, turns: thread.turns.filter((turn) => turns.has(turn.id)) }); },