Reuse measured chat item sizes in history pages

This commit is contained in:
Brennan Benson
2026-10-17 02:47:04 -07:00
parent 5455975195
commit 9f1f88aee7
8 changed files with 62 additions and 34 deletions
@@ -72,7 +72,7 @@ describe('journal payload allocation bounds', () => {
body,
sequence: 1,
observedAt: 1
})
}).item
const encodedStringBytes = from.mock.calls.reduce(
(largest, [value]) =>
typeof value === 'string' ? Math.max(largest, Buffer.byteLength(value, 'utf8')) : largest,
@@ -5,6 +5,8 @@ import type { AgentJournalRenderItem } from '../../../shared/agent-session-journ
import { boundJournalKeyComponent } from '../../../shared/agent-session-journal-item-key'
import {
BOUNDED_FRAME_CONTENT_BUDGET_BYTES,
measureHistoryItem,
type HistoryItemForm,
oversizedHistoryItem
} from './agent-session-history-page-bounds'
import { shortenedAgentJournalItem } from './agent-session-item-wire-form'
@@ -36,17 +38,16 @@ function oversizedFrameItem(item: AgentJournalRenderItem): AgentJournalRenderIte
}
}
function boundFrameItem(item: AgentJournalRenderItem): AgentJournalRenderItem {
return shortenedAgentJournalItem(item, BOUNDED_FRAME_ROW_BYTES) ?? oversizedFrameItem(item)
}
// Items are immutable per revision, so a catch-up or a second subscriber reuses the bound form.
const bounded = new WeakMap<AgentJournalRenderItem, AgentJournalRenderItem>()
const bounded = new WeakMap<AgentJournalRenderItem, HistoryItemForm>()
export function boundedFrameItem(item: AgentJournalRenderItem): AgentJournalRenderItem {
export function boundedFrameItem(formed: HistoryItemForm): HistoryItemForm {
const { item, bytes } = formed
let result = bounded.get(item)
if (result === undefined) {
result = boundFrameItem(item)
const fitted =
shortenedAgentJournalItem(item, BOUNDED_FRAME_ROW_BYTES, bytes) ?? oversizedFrameItem(item)
result = fitted === item ? formed : measureHistoryItem(fitted)
bounded.set(item, result)
}
return result
@@ -132,13 +132,15 @@ describe('a bounded row', () => {
})
]
for (const row of rows) {
expect(jsonBytes(boundedFrameItem(row))).toBeLessThanOrEqual(BOUNDED_FRAME_ROW_BYTES)
const formed = boundedFrameItem({ item: row, bytes: jsonBytes(row) })
expect(formed.bytes).toBe(jsonBytes(formed.item))
expect(formed.bytes).toBeLessThanOrEqual(BOUNDED_FRAME_ROW_BYTES)
}
})
it('is the stored row when it already fits', () => {
const row = rendered({ body: message('hello') })
expect(boundedFrameItem(row)).toBe(row)
expect(boundedFrameItem({ item: row, bytes: jsonBytes(row) }).item).toBe(row)
})
})
@@ -65,10 +65,19 @@ export function oversizedHistoryItem(item: AgentJournalRenderItem): AgentJournal
}
}
export type HistoryItemForm = {
item: AgentJournalRenderItem
bytes: number
}
export function measureHistoryItem(item: AgentJournalRenderItem): HistoryItemForm {
return { item, bytes: Buffer.byteLength(JSON.stringify(item), 'utf8') }
}
/** How a reader receives each item, and the bytes its page may hold of them. A page is measured
* in that form, so it carries what it will actually send. */
export type HistoryReader = {
form: (item: AgentJournalRenderItem) => AgentJournalRenderItem
form: (item: AgentJournalRenderItem) => HistoryItemForm
budgetBytes: number
}
@@ -78,14 +87,18 @@ export function boundHistoryItemsByBytes(
keep: 'newest' | 'oldest',
submissionBytes: ReadonlyMap<string, number>,
maxBytes: number,
form: HistoryReader['form'] = (item) => item
form: HistoryReader['form'] = measureHistoryItem
): { items: AgentJournalRenderItem[]; dropped: number } {
const ordered = groupItemsBySequence(items, keep)
const kept: AgentJournalRenderItem[][] = []
let total = 0
for (const stored of ordered) {
const group = stored.map(form)
const bytes = group.reduce((sum, item) => sum + historyEntryBytes(item, submissionBytes), 0)
const formed = stored.map(form)
const group = formed.map(({ item }) => item)
const bytes = formed.reduce(
(sum, { item, bytes }) => sum + bytes + (submissionBytes.get(item.itemId) ?? 0),
0
)
if (kept.length === 0 && bytes > maxBytes) {
kept.push(group.map((item) => oversizedHistoryItem(item)))
break
@@ -32,7 +32,8 @@ import { CLIENT_HISTORY_READER } from './agent-session-item-wire-form'
import {
boundHistoryItemsByBytes,
type HistoryReader,
historyEntryBytes,
type HistoryItemForm,
measureHistoryItem,
newestWholeSequenceGroups,
wholeSubagentGroupsStart,
oversizedHistoryItem,
@@ -235,10 +236,13 @@ function readForward(
// submissions AND removal ids — a legal pre-bounding tombstone id can dwarf
// every item on the page.
const pageContentBytes = (
items: readonly AgentJournalRenderItem[],
items: readonly HistoryItemForm[],
removedItemIds: readonly string[]
): number =>
items.reduce((total, item) => total + historyEntryBytes(item, submissionBytes), 0) +
items.reduce(
(total, { item, bytes }) => total + bytes + (submissionBytes.get(item.itemId) ?? 0),
0
) +
removedItemIds.reduce(
(total, itemId) => total + Buffer.byteLength(JSON.stringify(itemId), 'utf8') + 1,
0
@@ -247,7 +251,7 @@ function readForward(
// clipping projected items: dropping an item while advancing the cursor past
// the rows that touched it would lose that revision for good.
let rows = since.rows.slice(0, limit)
const project = (window: typeof rows): ReturnType<typeof projectJournalBatch> => {
const project = (window: typeof rows) => {
const result = projectJournalBatch({
rows: window,
snapshot,
@@ -275,9 +279,11 @@ function readForward(
// One row can still touch an over-budget item; degrade it visibly.
let items = projected.batch.items
if (contentBytes > reader.budgetBytes) {
items = items.map((item) => {
const bytes = historyEntryBytes(item, submissionBytes)
return bytes > reader.budgetBytes ? oversizedHistoryItem(item) : item
items = items.map((formed) => {
const bytes = formed.bytes + (submissionBytes.get(formed.item.itemId) ?? 0)
return bytes > reader.budgetBytes
? measureHistoryItem(oversizedHistoryItem(formed.item))
: formed
})
contentBytes = pageContentBytes(items, projected.batch.removedItemIds)
}
@@ -295,7 +301,7 @@ function readForward(
page: buildPage({
snapshot,
direction: 'after',
items,
items: items.map(({ item }) => item),
removedItemIds: projected.batch.removedItemIds,
// Reading after a position means there is something before it.
hasOlder: cursor.sequence > 0,
@@ -7,7 +7,9 @@ import { agentJournalItemForWire, shortenedAgentJournalItem } from './agent-sess
it('preserves the full previously readable desktop message range', () => {
const item: AgentJournalRenderItem = { itemId: 'message', revision: 1, sequence: 1, observedAt: 1,
body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: `${'x'.repeat(1_200_000)}TAIL` }] } }
expect(agentJournalItemForWire(item)).toBe(item)
const formed = agentJournalItemForWire(item)
expect(formed.item).toBe(item)
expect(formed.bytes).toBe(Buffer.byteLength(JSON.stringify(item), 'utf8'))
})
it.each([73_728, HISTORY_PAGE_CONTENT_BUDGET_BYTES, 4 * 1024 * 1024])('keeps tool identity and output when cardinality exceeds %i bytes', (budget) => {
@@ -13,6 +13,7 @@ import type {
import { boundToolInput, JOURNAL_RUNNING_PREVIEW_LIMITS, utf8Head } from '../agent-session-journal/journal-payload-bounds'
import {
HISTORY_PAGE_CONTENT_BUDGET_BYTES,
type HistoryItemForm,
type HistoryReader
} from './agent-session-history-page-bounds'
import { jsonBytes, shortenJsonStringsToFit } from './json-string-shortening'
@@ -55,9 +56,10 @@ function wireBody(body: AgentJournalItemBody): AgentJournalItemBody {
* that cannot fit, which the page answers with its too-large marker. */
export function shortenedAgentJournalItem(
item: AgentJournalRenderItem,
maxBytes: number
maxBytes: number,
bytes = jsonBytes(item)
): AgentJournalRenderItem | null {
if (jsonBytes(item) <= maxBytes) {
if (bytes <= maxBytes) {
return item
}
const budget = maxBytes - (jsonBytes({ ...item, body: null, shortened: true }) - jsonBytes(null))
@@ -71,17 +73,19 @@ export function shortenedAgentJournalItem(
return body === null ? null : { ...item, body, shortened: true }
}
function wireItem(item: AgentJournalRenderItem): AgentJournalRenderItem {
function wireItem(item: AgentJournalRenderItem): HistoryItemForm {
const body = wireBody(item.body)
const sent = body === item.body ? item : { ...item, body, shortened: true as const }
return shortenedAgentJournalItem(sent, CLIENT_ITEM_MAX_BYTES) ?? sent
const bytes = jsonBytes(sent)
const fitted = shortenedAgentJournalItem(sent, CLIENT_ITEM_MAX_BYTES, bytes) ?? sent
return { item: fitted, bytes: fitted === sent ? bytes : jsonBytes(fitted) }
}
// Items are immutable per revision, so a catch-up or a second subscriber reuses the sent form.
const sentForms = new WeakMap<AgentJournalRenderItem, AgentJournalRenderItem>()
const sentForms = new WeakMap<AgentJournalRenderItem, HistoryItemForm>()
/** What a client is sent of `item`. */
export function agentJournalItemForWire(item: AgentJournalRenderItem): AgentJournalRenderItem {
export function agentJournalItemForWire(item: AgentJournalRenderItem): HistoryItemForm {
let sent = sentForms.get(item)
if (sent === undefined) {
sent = wireItem(item)
@@ -1,7 +1,7 @@
import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types'
import { AGENT_SESSION_HISTORY_MAX_LIMIT } from '../../../shared/agent-session-wire'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import { oversizedHistoryItem } from './agent-session-history-page-bounds'
import { measureHistoryItem, oversizedHistoryItem } from './agent-session-history-page-bounds'
import { emptyAgentSessionBatch } from './agent-session-empty-batch'
import { subscriberHistoryReader } from './agent-session-bounded-frames'
import {
@@ -69,16 +69,16 @@ export function refreshDerivedStopNotes(
continue
}
const formed = form(item)
const projectedBytes = Buffer.byteLength(JSON.stringify(formed), 'utf8')
const bounded = projectedBytes > budgetBytes ? oversizedHistoryItem(formed) : formed
const itemBytes = Buffer.byteLength(JSON.stringify(bounded), 'utf8') + 1
const bounded =
formed.bytes > budgetBytes ? measureHistoryItem(oversizedHistoryItem(formed.item)) : formed
const itemBytes = bounded.bytes + 1
if (
pending.length > 0 &&
(pending.length === AGENT_SESSION_HISTORY_MAX_LIMIT || bytes + itemBytes > budgetBytes)
) {
emit()
}
pending.push(bounded)
pending.push(bounded.item)
bytes += itemBytes
}
if (pending.length > 0 && port.isActive(subscriber)) {