mirror of
https://github.com/stablyai/orca.git
synced 2026-10-06 08:02:28 +00:00
perf: reuse unchanged history page byte totals
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, expect, it } from 'vitest'
|
||||
import type {
|
||||
AgentJournalItemBody,
|
||||
AgentJournalItemIdentity,
|
||||
AgentSessionJournalIdentity
|
||||
} from '../../../shared/agent-session-journal-types'
|
||||
import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open'
|
||||
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
|
||||
import { readAgentSessionHistory } from './agent-session-history-page'
|
||||
|
||||
const IDENTITY: AgentSessionJournalIdentity = {
|
||||
sessionId: 'session-1',
|
||||
workspaceId: 'ws-1',
|
||||
hostId: 'host-1',
|
||||
agent: 'codex',
|
||||
providerHandle: { kind: 'codex', threadId: 'thread-1' }
|
||||
}
|
||||
|
||||
const journals = createTrackedJournalOpener()
|
||||
let root: string
|
||||
let clock = 1_000
|
||||
let epochs = 0
|
||||
let journal: AgentSessionJournal
|
||||
|
||||
function tick(): number {
|
||||
clock += 1
|
||||
return clock
|
||||
}
|
||||
|
||||
function item(ordinal: number): AgentJournalItemIdentity {
|
||||
return { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal }
|
||||
}
|
||||
|
||||
function body(text: string): AgentJournalItemBody {
|
||||
return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text }] }
|
||||
}
|
||||
|
||||
async function appendItems(count: number, text: string): Promise<void> {
|
||||
for (let ordinal = 1; ordinal <= count; ordinal += 1) {
|
||||
await journal.appendItem(item(ordinal), body(`${text}-${ordinal}`), { fence: 1 })
|
||||
}
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-wire-history-'))
|
||||
clock = 1_000
|
||||
epochs = 0
|
||||
journal = await journals.open({
|
||||
identity: IDENTITY,
|
||||
journalDir: root,
|
||||
now: tick,
|
||||
mintEpoch: () => {
|
||||
epochs += 1
|
||||
return `epoch-${epochs}`
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await journals.closeAll()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
it.each([1, 100, 200])('serializes each of %i unchanged forward page items once', async (count) => {
|
||||
const cursor = journal.cursor()
|
||||
await appendItems(count, 'x'.repeat(8_000))
|
||||
const snapshot = journal.snapshot()
|
||||
const stringify = JSON.stringify
|
||||
let itemSerializations = 0
|
||||
JSON.stringify = ((value: unknown, ...args: unknown[]) => {
|
||||
if (value && typeof value === 'object' && 'itemId' in value && 'body' in value) {
|
||||
itemSerializations++
|
||||
}
|
||||
return Reflect.apply(stringify, JSON, [value, ...args])
|
||||
}) as typeof JSON.stringify
|
||||
try {
|
||||
const result = readAgentSessionHistory(
|
||||
journal,
|
||||
{
|
||||
sessionId: 'session-1',
|
||||
direction: 'after',
|
||||
limit: count,
|
||||
cursor
|
||||
},
|
||||
snapshot
|
||||
)
|
||||
expect(result.ok).toBe(true)
|
||||
expect(result.page.items).toHaveLength(count)
|
||||
expect(itemSerializations).toBe(count)
|
||||
} finally {
|
||||
JSON.stringify = stringify
|
||||
}
|
||||
})
|
||||
@@ -196,11 +196,8 @@ function readForward(
|
||||
if (!projected.ok) {
|
||||
return historyReset(snapshot, projected.reset)
|
||||
}
|
||||
while (
|
||||
rows.length > 1 &&
|
||||
pageContentBytes(projected.batch.items, projected.batch.removedItemIds) >
|
||||
HISTORY_PAGE_CONTENT_BUDGET_BYTES
|
||||
) {
|
||||
let contentBytes = pageContentBytes(projected.batch.items, projected.batch.removedItemIds)
|
||||
while (rows.length > 1 && contentBytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES) {
|
||||
rows = rows.slice(0, Math.ceil(rows.length / 2))
|
||||
const shrunk = projectJournalBatch({
|
||||
rows,
|
||||
@@ -212,19 +209,18 @@ function readForward(
|
||||
return historyReset(snapshot, shrunk.reset)
|
||||
}
|
||||
projected = shrunk
|
||||
contentBytes = pageContentBytes(projected.batch.items, projected.batch.removedItemIds)
|
||||
}
|
||||
// One row can still touch an over-budget item; degrade it visibly.
|
||||
const items =
|
||||
pageContentBytes(projected.batch.items, projected.batch.removedItemIds) >
|
||||
HISTORY_PAGE_CONTENT_BUDGET_BYTES
|
||||
? projected.batch.items.map((item) => {
|
||||
const bytes = historyEntryBytes(item, submissionBytes)
|
||||
return bytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES
|
||||
? oversizedHistoryItem(item, bytes)
|
||||
: item
|
||||
})
|
||||
: projected.batch.items
|
||||
if (pageContentBytes(items, projected.batch.removedItemIds) > HISTORY_PAGE_CONTENT_BUDGET_BYTES) {
|
||||
let items = projected.batch.items
|
||||
if (contentBytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES) {
|
||||
items = items.map((item) => {
|
||||
const bytes = historyEntryBytes(item, submissionBytes)
|
||||
return bytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES ? oversizedHistoryItem(item, bytes) : item
|
||||
})
|
||||
contentBytes = pageContentBytes(items, projected.batch.removedItemIds)
|
||||
}
|
||||
if (contentBytes > HISTORY_PAGE_CONTENT_BUDGET_BYTES) {
|
||||
// A single row's semantic payload — in practice a pre-bounding oversized
|
||||
// removal id — can never fit any page, and truncating a removal id would
|
||||
// break the client's keying. A bounded tail replaces the client's state
|
||||
|
||||
Reference in New Issue
Block a user