fix(chat): preserve complete writes and keep detail current

This commit is contained in:
Brennan Benson
2026-10-16 23:31:29 -07:00
parent 7c5006c083
commit f89b4ea989
29 changed files with 772 additions and 421 deletions
@@ -1,15 +1,15 @@
import { useContext, useState } from 'react'
import { useContext, useEffect, useState } from 'react'
import { Pressable, Text, View } from 'react-native'
import {
truncateToolDetail,
MAX_TOOL_DETAIL_LENGTH
} from '../../../src/shared/native-chat-tool-summary'
import { isTextBlock, type NativeChatMessage } from '../../../src/shared/native-chat-types'
import type { NativeChatMessage, NativeChatToolResultBlock } from '../../../src/shared/native-chat-types'
import type { AgentJournalBoundedPayload, AgentJournalItemBody } from '../../../src/shared/agent-session-journal-types'
import { styles } from './mobile-native-chat-message-styles'
import {
MobileNativeChatItemSourceContext,
readMobileNativeChatItem,
type MobileNativeChatItemBody
readMobileNativeChatItem
} from './mobile-native-chat-item-read'
// "Show full output" and "Show full message": a row the phone was sent a head of, read whole from
@@ -41,47 +41,51 @@ function ShowFull({
)
}
function outputOf(body: MobileNativeChatItemBody | null): string | null {
if (body?.kind === 'tool-call' && 'output' in body) {
return body.output?.head ?? null
function outputOf(body: AgentJournalItemBody | null): AgentJournalBoundedPayload | null {
if (body?.kind === 'tool-call') {
return body.output ?? null
}
return body?.kind === 'diff' && 'patch' in body ? body.patch.head : null
return body?.kind === 'diff' ? body.patch : null
}
/** A tool's output in an opened line: its head, and the whole of it on request. */
export function MobileNativeChatToolOutput({
output,
isError,
fullOutputItemId
body
}: {
output: string
isError?: boolean
fullOutputItemId?: string
body: NativeChatToolResultBlock
}): React.JSX.Element {
const source = useContext(MobileNativeChatItemSourceContext)
const [full, setFull] = useState<string | null>(null)
const [load, setLoad] = useState<Load>('idle')
const clipped =
full === null && (fullOutputItemId !== undefined || output.length > MAX_TOOL_DETAIL_LENGTH)
const { output, isError, fullOutputItemId } = body
const client = source?.client, sessionId = source?.sessionId
const [expanded, setExpanded] = useState(false)
const [detail, setDetail] = useState<{
body: typeof body; client: typeof client; sessionId: typeof sessionId;
full?: AgentJournalBoundedPayload; load: Load
} | null>(null)
useEffect(() => () => setDetail((held) => held?.body === body && held.client === client && held.sessionId === sessionId ? null : held), [body, client, sessionId])
const current = detail?.body === body && detail.client === client && detail.sessionId === sessionId ? detail : null
const full = current?.full
const clipped = !full && (fullOutputItemId !== undefined || (!expanded && output.length > MAX_TOOL_DETAIL_LENGTH))
const showFull = (): void => {
if (fullOutputItemId === undefined || source === null) {
setFull(output)
setExpanded(true)
return
}
setLoad('loading')
setDetail({ body, client, sessionId, load: 'loading' })
const finish = (full?: AgentJournalBoundedPayload): void => {
setDetail((held) => held?.body === body && held.client === client && held.sessionId === sessionId ? { body, client, sessionId, full, load: full ? 'idle' : 'failed' } : held)
}
readMobileNativeChatItem(source, fullOutputItemId)
.then((body) => {
setFull(outputOf(body) ?? output)
setLoad('idle')
})
.catch(() => setLoad('failed'))
.then((item) => finish(item?.itemId === fullOutputItemId ? outputOf(item.body) ?? undefined : undefined))
.catch(() => finish())
}
return (
<View>
<View style={[styles.toolResult, isError && styles.toolResultError]}>
<Text style={styles.mono}>{full ?? truncateToolDetail(output)}</Text>
<Text style={styles.mono}>{full ? full.head : expanded ? output : truncateToolDetail(output)}</Text>
</View>
{clipped ? <ShowFull label="Show full output" load={load} onPress={showFull} /> : null}
{full?.truncated ? <Text style={styles.showFullText}>Showing {new TextEncoder().encode(full.head).length} of {full.byteLength} bytes.</Text> : null}
{clipped ? <ShowFull label="Show full output" load={current?.load ?? 'idle'} onPress={showFull} /> : null}
</View>
)
}
@@ -91,35 +95,19 @@ export function useMobileNativeChatFullMessage(message: NativeChatMessage): {
shown: NativeChatMessage
showFull: React.JSX.Element | null
} {
const source = useContext(MobileNativeChatItemSourceContext)
const [texts, setTexts] = useState<(string | undefined)[] | null>(null)
const [load, setLoad] = useState<Load>('idle')
if (texts) {
// Shortening cuts strings only, so the host's blocks line up with the row's one for one.
const blocks = message.blocks.map((block, index) => {
const text = texts[index]
return isTextBlock(block) && text !== undefined ? { ...block, text } : block
})
return { shown: { ...message, blocks }, showFull: null }
}
if (message.shortened !== true || source === null) {
return { shown: message, showFull: null }
}
const detail = useContext(MobileNativeChatItemSourceContext)?.messageDetail
const raw = detail?.raw(message.id)
const [request, setRequest] = useState<{ raw: typeof raw; load: Load } | null>(null)
const load = request?.raw === raw ? request?.load ?? 'idle' : 'idle'
const state = detail?.state(message.id)
const showFull = (): void => {
setLoad('loading')
readMobileNativeChatItem(source, message.id)
.then((body) => {
setTexts(
body?.kind === 'message' && 'blocks' in body
? body.blocks.map((block) => (block.type === 'text' ? block.text : undefined))
: []
)
setLoad('idle')
})
.catch(() => setLoad('failed'))
setRequest({ raw, load: 'loading' })
const finish = (load: Load): void => setRequest((held) => held?.raw === raw ? { raw, load } : held)
void detail?.load(message.id).then(() => finish('idle')).catch(() => finish('failed'))
}
return {
shown: message,
showFull: <ShowFull label="Show full message" load={load} onPress={showFull} />
showFull: state === 'partial' ? <Text style={styles.showFullText}>Showing part of this message.</Text>
: state === 'available' ? <ShowFull label="Show full message" load={load} onPress={showFull} /> : null
}
}
@@ -9,6 +9,7 @@ import { useMobileNativeChatComposerTray } from './use-mobile-native-chat-compos
import { useMobileNativeChatVisualRenderer } from './MobileNativeChatVisual'
import { MobileNativeChatVisualContext } from './mobile-native-chat-visual-context'
import { MobileNativeChatItemSourceContext } from './mobile-native-chat-item-read'
import { useMobileNativeChatMessageDetail } from './use-mobile-native-chat-message-detail'
import {
MobileNativeChatReplyRevealsContext,
useMobileNativeChatReplyReveals
@@ -66,9 +67,10 @@ export function MobileNativeChatOverlay({
const session = controller.nativeChatSession
// The chat is drawn only with a resolved agent; the fallback never reaches the screen.
const agent = controller.nativeChatAgent ?? ''
const fullContent = useMobileNativeChatMessageDetail(session.messages, controller.nativeChatItemSource)
const folded = useMemo(
() => foldMobileNativeChatMessages(session.messages, agent),
[session.messages, agent]
() => foldMobileNativeChatMessages(fullContent.messages, agent),
[fullContent.messages, agent]
)
const streaming = useMobileNativeChatStreamingBubble(
folded,
@@ -102,7 +104,7 @@ export function MobileNativeChatOverlay({
<MobileNativeChatRowOverlaysHost conversation={{ surfaceId: sendSurfaceId, sessionId }}>
<View style={styles.overlay}>
<MobileNativeChatVisualContext.Provider value={visuals}>
<MobileNativeChatItemSourceContext.Provider value={controller.nativeChatItemSource}>
<MobileNativeChatItemSourceContext.Provider value={controller.nativeChatItemSource ? { ...controller.nativeChatItemSource, messageDetail: fullContent.detail } : null}>
<MobileNativeChatReplyRevealsContext.Provider value={replyReveals}>
<MobileNativeChatView
messages={session.messages}
@@ -64,9 +64,7 @@ function ResultBody({
}
return (
<MobileNativeChatToolOutput
output={result.output}
isError={result.isError}
fullOutputItemId={result.fullOutputItemId}
body={result}
/>
)
}
@@ -1,31 +1,16 @@
import { createContext } from 'react'
import { z } from 'zod'
import { isAdmissibleAgentJournalRenderItem } from '../../../src/shared/agent-session-journal-schemas'
import type { AgentJournalRenderItem } from '../../../src/shared/agent-session-journal-types'
import { bindDeferredRpcOperation, defineRpcOperation } from '../transport/rpc-operation'
import { rpcResultVariant } from '../transport/rpc-operation-result-reader'
import type { MobileNativeChatVisualSource } from './mobile-native-chat-visual-read'
import type { MobileNativeChatMessageDetail } from './use-mobile-native-chat-message-detail'
// One chat item whole, for "Show full output" and "Show full message": every other read sends the
// phone a head of long text, and the host keeps the rest.
const Payload = z.object({ head: z.string(), byteLength: z.number(), truncated: z.boolean() })
const itemReplySchema = z.object({
item: z
.object({
body: z.union([
z.object({ kind: z.literal('tool-call'), output: Payload.optional() }),
z.object({ kind: z.literal('diff'), patch: Payload }),
z.object({
kind: z.literal('message'),
blocks: z.array(z.object({ type: z.string(), text: z.string().optional() }))
}),
z.object({ kind: z.string() })
])
})
.nullable()
})
export type MobileNativeChatItemBody = NonNullable<z.output<typeof itemReplySchema>['item']>['body']
const itemReplySchema = z.object({ item: z.custom<AgentJournalRenderItem>(isAdmissibleAgentJournalRenderItem).nullable() })
const nativeChatItemRead = bindDeferredRpcOperation(
defineRpcOperation({
@@ -40,7 +25,7 @@ const nativeChatItemRead = bindDeferredRpcOperation(
const READ_TIMEOUT_MS = 30_000
/** Where whole items are read from; null where the host predates `agentSession.item`. */
export const MobileNativeChatItemSourceContext = createContext<MobileNativeChatVisualSource | null>(
export const MobileNativeChatItemSourceContext = createContext<(MobileNativeChatVisualSource & { messageDetail?: MobileNativeChatMessageDetail | null }) | null>(
null
)
@@ -49,7 +34,7 @@ export const MobileNativeChatItemSourceContext = createContext<MobileNativeChatV
export async function readMobileNativeChatItem(
source: MobileNativeChatVisualSource,
itemId: string
): Promise<MobileNativeChatItemBody | null> {
): Promise<AgentJournalRenderItem | null> {
const response = await nativeChatItemRead.request(
source.client,
{ sessionId: source.sessionId, itemId },
@@ -59,5 +44,5 @@ export async function readMobileNativeChatItem(
if (reply === null) {
throw new Error('The host could not read this item.')
}
return reply.item?.body ?? null
return reply.item
}
@@ -0,0 +1,48 @@
import { createElement } from 'react'
import { act, create, type ReactTestRenderer } from 'react-test-renderer'
import { afterEach, expect, it, vi } from 'vitest'
import type { NativeChatMessage } from '../../../src/shared/native-chat-types'
import type { AgentJournalRenderItem } from '../../../src/shared/agent-session-journal-types'
import type { RpcClient } from '../transport/rpc-client'
import { foldMobileNativeChatMessages } from './mobile-native-chat-render-data'
import { useMobileNativeChatMessageDetail } from './use-mobile-native-chat-message-detail'
const reads = vi.hoisted(() => vi.fn())
vi.mock('./mobile-native-chat-item-read', () => ({ readMobileNativeChatItem: reads }))
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: RPC is mocked; only this client's object identity is used.
const source = { client: {} as RpcClient, sessionId: 'session' }
let latest: ReturnType<typeof useMobileNativeChatMessageDetail> | undefined
let renderer: ReactTestRenderer | undefined
function Probe({ messages }: { messages: NativeChatMessage[] }) {
latest = useMobileNativeChatMessageDetail(messages, source)
return null
}
afterEach(() => { act(() => renderer?.unmount()); reads.mockReset() })
it('prepares full raw user rows with their neighboring images and fences late same-head detail', async () => {
const row = (id: string, text: string): NativeChatMessage => ({ id, role: 'user', source: 'transcript', timestamp: 1, blocks: [{ type: 'text', text }] })
const image = row('image', '[Image: source: /repo/p.png]')
const prompt = { ...row('prompt', '[Image #1] head…'), shortened: true }
const skill = { ...row('skill', '<command-name>/plugin:example</command-name><command-args>head…</command-args>'), shortened: true }
const messages = [image, prompt, skill]
const item = (id: string, text: string, shortened = false): AgentJournalRenderItem => ({ itemId: id, sequence: 1, observedAt: 1, revision: 2, body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }, ...(shortened ? { shortened: true as const } : {}) })
await act(async () => { renderer = create(createElement(Probe, { messages })) })
reads.mockResolvedValueOnce(item('prompt', '[Image #1] head TAIL'))
await act(async () => { await latest?.detail?.load('prompt') })
const folded = foldMobileNativeChatMessages(latest?.messages ?? [], 'claude')
expect(folded[0]?.blocks).toEqual([{ type: 'image-ref', path: '/repo/p.png' }, { type: 'text', text: 'head TAIL' }])
reads.mockResolvedValueOnce(item('skill', '<command-name>/plugin:example</command-name><command-args>full</command-args>', true))
await act(async () => { await latest?.detail?.load('skill') })
expect(foldMobileNativeChatMessages(latest?.messages ?? [], 'claude')[1]?.blocks).toEqual([{ type: 'text', text: '/example full' }])
expect(latest?.detail?.state('skill')).toBe('partial')
const pending = Promise.withResolvers<AgentJournalRenderItem>()
reads.mockReturnValueOnce(pending.promise)
const stale = latest?.detail?.load('prompt')
const newer = { ...prompt }
await act(async () => renderer?.update(createElement(Probe, { messages: [image, newer, skill] })))
await act(async () => { pending.resolve(item('prompt', 'STALE')); await stale })
expect(latest?.messages[1]).toBe(newer)
reads.mockResolvedValueOnce(null)
await expect(latest?.detail?.load('prompt')).rejects.toThrow('Message unavailable')
expect(latest?.detail?.state('prompt')).toBe('available')
})
@@ -0,0 +1,59 @@
import { useEffect, useMemo, useReducer } from 'react'
import { isTextBlock, type NativeChatMessage } from '../../../src/shared/native-chat-types'
import { projectStructuredItemToNativeChat } from '../../../src/shared/structured-agent-session-projection'
import { readMobileNativeChatItem } from './mobile-native-chat-item-read'
import type { MobileNativeChatVisualSource } from './mobile-native-chat-visual-read'
export type MobileNativeChatMessageDetail = {
raw(id: string): NativeChatMessage | undefined
state(id: string): 'available' | 'partial' | 'complete' | null
load(id: string): Promise<void>
}
/** Substitute the original raw row; image, skill and tool preparation then run once as usual. */
export function useMobileNativeChatMessageDetail(
messages: NativeChatMessage[], source: MobileNativeChatVisualSource | null
): { messages: NativeChatMessage[]; detail: MobileNativeChatMessageDetail | null } {
const client = source?.client, sessionId = source?.sessionId
const cache = useMemo(() => ({
live: true, raw: new Set<NativeChatMessage>(),
entries: new Map<NativeChatMessage, NativeChatMessage>()
}), [client, sessionId])
const [revision, refresh] = useReducer((revision: number) => revision + 1, 0)
useEffect(() => {
cache.live = true
return () => { cache.live = false; cache.entries.clear() }
}, [cache])
useEffect(() => {
cache.raw = new Set(messages)
for (const raw of cache.entries.keys()) {
if (!cache.raw.has(raw)) cache.entries.delete(raw)
}
}, [cache, messages])
const byId = useMemo(() => new Map(messages.map((message) => [message.id, message])), [messages])
const raw = (id: string): NativeChatMessage | undefined => byId.get(id)
return {
messages: useMemo(() => messages.map((message) => cache.entries.get(message) ?? message), [cache, messages, revision]),
detail: source === null ? null : {
raw,
state: (id) => {
const original = raw(id)
if (!original?.shortened || !original.blocks.some(isTextBlock)) return null
const loaded = cache.entries.get(original)
return loaded ? (loaded.shortened ? 'partial' : 'complete') : 'available'
},
load: async (id) => {
const original = raw(id)
if (!original?.shortened || !original.blocks.some(isTextBlock)) throw new Error('Message unavailable')
const item = await readMobileNativeChatItem(source, id)
const projected = item?.itemId === id && item.body.kind === 'message'
? projectStructuredItemToNativeChat(item) : null
if (!projected?.blocks.some(isTextBlock)) throw new Error('Message unavailable')
if (cache.live && cache.raw.has(original)) {
cache.entries.set(original, projected)
refresh()
}
}
}
}
}
+25 -6
View File
@@ -89,12 +89,19 @@ export class AcpToolTimeline {
private readonly turns = new BoundedMap<string, string>({
maxEntries: MAX_PROVIDER_TIMELINE_OPEN_ENTRIES
})
/** A running call's whole input and output, outside the budgeted snapshot that holds only their
* preview, so its closing revision keeps them even when the closing update repeats neither. */
private readonly whole = new BoundedMap<string, { input?: unknown; output?: string }>({
maxEntries: MAX_PROVIDER_TIMELINE_OPEN_ENTRIES
// One latest snapshot per live call; the closing event takes ownership. Bytes are a ledger,
// not an RSS ceiling: provider frame limits and the live-entry count bound this finite carry.
private readonly whole = new BoundedMap<string, {
input?: unknown; output?: string; inputBytes: number; outputBytes: number
}>({
maxEntries: MAX_PROVIDER_TIMELINE_OPEN_ENTRIES,
sizeOf: (value, key) => value.inputBytes + value.outputBytes + Buffer.byteLength(key) + 64
})
get retainedToolBytes(): number {
return this.whole.retainedBytes
}
translate(
raw: ToolCallUpdate,
dialect: AcpDialect,
@@ -104,7 +111,8 @@ export class AcpToolTimeline {
const previous = this.tools.get(update.toolCallId)
const held = this.whole.get(update.toolCallId)
const input = update.rawInput === undefined ? held?.input : update.rawInput
const output = outputText(update) ?? held?.output
const suppliedOutput = outputText(update)
const output = suppliedOutput ?? held?.output
const exitCode = rawExitCodeSchema.safeParse(update.rawOutput).data?.exitCode
const state =
update.status === 'completed'
@@ -115,7 +123,18 @@ export class AcpToolTimeline {
? 'running'
: (previous?.body.state ?? 'running')
if (state === 'running') {
this.whole.set(update.toolCallId, { input, output })
if (!held && this.whole.size >= MAX_PROVIDER_TIMELINE_OPEN_ENTRIES) {
throw new Error('ACP tool stream capacity exceeded')
}
if (!this.whole.set(update.toolCallId, {
input, output,
inputBytes: update.rawInput === undefined ? (held?.inputBytes ?? 0)
: Buffer.byteLength(JSON.stringify(input) ?? ''),
outputBytes: suppliedOutput === undefined ? (held?.outputBytes ?? 0)
: Buffer.byteLength(JSON.stringify(output) ?? '')
})) {
throw new Error('ACP tool snapshot could not be retained')
}
} else {
this.whole.delete(update.toolCallId)
}
@@ -467,7 +467,7 @@ describe('codex turn lifecycle rows', () => {
])
})
it('replays a backpressured turn boundary with its original receipt time', async () => {
it('replays an oversized backpressured notification with its original receipt time', async () => {
vi.useFakeTimers()
const connection = {
pauseReading: vi.fn(),
@@ -483,7 +483,7 @@ describe('codex turn lifecycle rows', () => {
})
expect(
retries.handle(SESSION_ID, 'turn/started', { turn: { id: TURN_ID } }, 1_000, -1)
retries.handle(SESSION_ID, 'turn/started', { turn: { id: TURN_ID }, output: 'x'.repeat(9 * 1024 * 1024) }, 1_000, -1)
).toEqual({ accepted: false, reason: 'backpressure' })
await vi.advanceTimersByTimeAsync(50)
@@ -122,7 +122,7 @@ export function createCodexStructuredNotificationRetry(deps: {
state = { connection, events: [], bytes: 0, timer: null, running: false, failed: false }
states.set(sessionId, state)
}
if (state.events.length >= MAX_RETRY_EVENTS || state.bytes + bytes > MAX_RETRY_BYTES) {
if (state.events.length >= MAX_RETRY_EVENTS || (state.events.length > 0 && state.bytes - state.events[0].bytes + bytes > MAX_RETRY_BYTES)) {
// A bounded queue cannot retain more traffic. Fail it truthfully so the
// provider's generation enters host recovery instead of stranding a pause.
fail(sessionId, state, 'notification retry queue overflow')
+16 -22
View File
@@ -81,16 +81,18 @@ describe('Pi final tail before host eviction', () => {
expect(append('lifecycle-refused', true)).toEqual({ accepted: false, reason: 'backpressure' })
const before = journal.snapshot().items.length
vi.useFakeTimers()
const reply = 'Final reply' + 'r'.repeat(2 * 1024 * 1024)
const status = 'Final status' + 's'.repeat(17 * 1024 * 1024)
lane.apply([
{ type: 'text.delta', item: { id: 'reply' }, channel: 'assistant', text: 'Final reply' },
{ type: 'text.delta', item: { id: 'reply' }, channel: 'assistant', text: reply },
{
type: 'item.update',
item: 'tail',
body: { kind: 'status', tone: 'info', text: 'Final status' }
body: { kind: 'status', tone: 'info', text: status }
}
])
lane.finalize()
lane.apply([{ type: 'session.ended', verdict: { state: 'interrupted', completedAt: 2 } }])
lane.finalize({ type: 'session.ended', verdict: { state: 'interrupted', completedAt: 2 } })
lane.apply([{ type: 'item.close', item: 'after-seal', body: { kind: 'status', tone: 'info', text: 'must not land' } }])
let landed = false
const barrier = deferred.drained().then((result) => {
landed = true
@@ -112,14 +114,15 @@ describe('Pi final tail before host eviction', () => {
discardSink: vi.fn()
})
expect(acknowledgeRelease).toHaveBeenCalledOnce()
expect(journal.snapshot().items.some((item) => item.body.kind === 'status' && item.body.text === 'must not land')).toBe(false)
expect(journal.snapshot().items.map((item) => item.body)).toEqual(
expect.arrayContaining([
expect.objectContaining({
kind: 'message',
role: 'assistant',
blocks: [{ type: 'text', text: 'Final reply' }]
blocks: [{ type: 'text', text: reply }]
}),
expect.objectContaining({ kind: 'status', text: 'Final status' }),
expect.objectContaining({ kind: 'status', text: status }),
expect.objectContaining({
kind: 'approval',
resolution: expect.objectContaining({ state: 'cancelled' })
@@ -129,28 +132,19 @@ describe('Pi final tail before host eviction', () => {
)
})
it('fails excessive final output explicitly instead of leaving a refused tail waiting for eviction', async () => {
it('keeps the final transfer operation-count limit', async () => {
const failures: unknown[] = []
const deferred = createDeferredStructuredAgentSessionEventSink({
...testEventSinkLogging(SESSION),
onFailed: (error) => failures.push(error)
})
cleanups.push(() => deferred.close())
expect(
deferred.sink.tryAppendTransition?.({
finalTail: true,
lifecycle: true,
publish: false,
steps: [
{
kind: 'settlement',
settlementId: 'oversize',
reservedBytes: 16 * 1024 * 1024 + 1,
resolve: () => []
}
]
})
).toEqual({ accepted: false, reason: 'failed' })
for (let index = 0; index <= 512; index += 1) {
expect(deferred.sink.tryAppendTransition?.({
finalTail: true, lifecycle: true, publish: false,
steps: [{ kind: 'settlement', settlementId: `tail:${index}`, reservedBytes: 1, resolve: () => [] }]
})).toEqual(index < 512 ? { accepted: true } : { accepted: false, reason: 'failed' })
}
expect(failures).toEqual([
expect.objectContaining({
message: 'structured agent-session final tail exceeded its reserved capacity'
+68 -23
View File
@@ -27,7 +27,9 @@ export type JsonlRpcTimelineLaneDeps = {
export class JsonlRpcTimelineLane {
private readonly assembler: ProviderTimelineAssembler
private readonly identity: ReturnType<typeof createLegacyProviderTimelineIdentityScheme>
private readonly held: { event: ProviderTimelineEvent; bytes: number }[] = []
private readonly held: { events: { event: ProviderTimelineEvent; bytes: number }[]; bytes: number }[] = []
private heldEvents = 0
private sealed = false
private heldBytes = 0
private draining = false
private ended = false
@@ -55,7 +57,7 @@ export class JsonlRpcTimelineLane {
beginCommand(command: StructuredAgentSessionCommandRun): void {
this.retry()
if (this.ended || this.held.length > 0) {
if (this.ended || this.sealed || this.held.length > 0) {
throw new Error('Provider timeline has not drained for this command')
}
this.assembler.beginCommand(command)
@@ -66,19 +68,47 @@ export class JsonlRpcTimelineLane {
}
apply(events: readonly ProviderTimelineEvent[]): void {
if (this.ended) {
if (this.ended || this.sealed) {
return
}
for (const event of events) {
const bytes = Buffer.byteLength(JSON.stringify(event))
if (this.held.length >= MAX_HELD_EVENTS || this.heldBytes + bytes > MAX_HELD_BYTES) {
this.fail('Agent timeline queue capacity exceeded')
return
const pending = events.map((event) => ({ event, bytes: Buffer.byteLength(JSON.stringify(event)) }))
// Try the current frame first; only a refused suffix becomes retained head work.
while (this.held.length === 0 && pending.length > 0 && !this.ended) {
if (!this.consume(pending[0].event)) {
break
}
this.held.push({ event, bytes })
this.heldBytes += bytes
this.retry()
pending.shift()
}
this.notify()
if (pending.length === 0 || this.ended) {
return
}
const bytes = pending.reduce((total, entry) => total + entry.bytes, 0)
if (
this.heldEvents + pending.length > MAX_HELD_EVENTS ||
(this.held.length > 0 && this.heldBytes - this.held[0].bytes + bytes > MAX_HELD_BYTES)
) {
this.fail('Agent timeline queue capacity exceeded')
return
}
this.held.push({ events: pending, bytes })
this.heldEvents += pending.length
this.heldBytes += bytes
this.retry()
}
private consume(event: ProviderTimelineEvent): boolean {
const result = this.assembler.apply(event)
if (!result.admission.accepted) {
if (result.admission.reason !== 'backpressure') {
this.fail(result.admission.reason)
}
return false
}
if (event.type === 'input.accepted' && !result.dropped) {
this.deps.onInputAccepted(event.clientMessageId)
}
return true
}
retry(): void {
@@ -90,29 +120,31 @@ export class JsonlRpcTimelineLane {
this.draining = true
try {
while (this.held.length > 0 && !this.ended) {
const entry = this.held[0]
const result = this.assembler.apply(entry.event)
if (!result.admission.accepted) {
if (result.admission.reason !== 'backpressure') {
this.fail(result.admission.reason)
} else {
const batch = this.held[0]
const entry = batch.events[0]
if (!this.consume(entry.event)) {
if (!this.ended) {
this.deps.pauseReading()
this.retryTimer = setTimeout(() => this.retry(), 250)
this.retryTimer.unref()
}
return
}
this.held.shift()
batch.events.shift()
batch.bytes -= entry.bytes
this.heldBytes -= entry.bytes
if (entry.event.type === 'input.accepted' && !result.dropped) {
this.deps.onInputAccepted(entry.event.clientMessageId)
this.heldEvents -= 1
if (batch.events.length === 0) {
this.held.shift()
}
}
} finally {
this.draining = false
this.notify()
}
this.deps.resumeReading()
if (!this.sealed && !this.ended) {
this.deps.resumeReading()
}
}
whenTurnLeaves(turnId: string): Promise<void> {
@@ -161,11 +193,23 @@ export class JsonlRpcTimelineLane {
this.assembler.flush()
}
finalize(): void {
// Admission must precede the host's exit barrier even when lifecycle writes are full.
finalize(end?: Extract<ProviderTimelineEvent, { type: 'session.ended' }>): void {
if (this.sealed || this.ended) {
return
}
// At most 256 held events + 128 dirty streams + this internal end: below 512 tail slots.
this.sealed = true
this.finalizing = true
this.retry()
if (this.ended) {
return
}
if (this.held.length > 0 || (end && !this.consume(end))) {
this.fail('Agent final timeline could not be transferred')
return
}
this.flush()
this.assembler.dispose()
}
dispose(): void {
@@ -173,6 +217,7 @@ export class JsonlRpcTimelineLane {
clearTimeout(this.retryTimer)
this.held.length = 0
this.heldBytes = 0
this.heldEvents = 0
this.assembler.dispose()
this.notify()
}
@@ -1,27 +0,0 @@
import { describe, expect, it } from 'vitest'
import { journalBodyWithinCeiling } from './journal-body-ceiling'
import { JOURNAL_PAYLOAD_MAX_BYTES } from './journal-payload-bounds'
describe('journal body ceiling', () => {
it('keeps the head and tail of a text past the ceiling, and leaves the rest alone', () => {
const text = `start${'x'.repeat(JOURNAL_PAYLOAD_MAX_BYTES)}end`
const body = {
kind: 'tool-call',
input: { command: 'cat huge.log' },
output: { head: text, byteLength: text.length, digest: 'd', truncated: false }
}
const kept = journalBodyWithinCeiling(body)
expect(kept.input).toBe(body.input)
expect(kept.output.truncated).toBe(true)
expect(kept.output.byteLength).toBe(text.length)
expect(kept.output.head.startsWith('start')).toBe(true)
expect(kept.output.head.endsWith('end')).toBe(true)
expect(Buffer.byteLength(kept.output.head, 'utf8')).toBeLessThan(
JOURNAL_PAYLOAD_MAX_BYTES + 1024
)
const small = { kind: 'message', blocks: [{ type: 'text', text: 'a'.repeat(1024) }] }
expect(journalBodyWithinCeiling(small)).toBe(small)
})
})
@@ -1,63 +0,0 @@
// The one bound every saved item body keeps: no text in it past `JOURNAL_PAYLOAD_MAX_BYTES`.
//
// A single write past the journal write queue's caps stalls the chat, so a larger text keeps its
// head and tail. It is applied where every provider's rows enter that queue
// (structured-agent-session-event-sink.ts), so no translator has to remember it.
import type { AgentJournalBoundedPayload } from '../../../shared/agent-session-journal-types'
import { JOURNAL_PAYLOAD_MAX_BYTES, journalTextWithinCeiling } from './journal-payload-bounds'
function isBoundedPayload(value: object): value is AgentJournalBoundedPayload {
return (
'head' in value &&
typeof value.head === 'string' &&
'byteLength' in value &&
'truncated' in value
)
}
function longestString(value: unknown): number {
if (typeof value === 'string') {
return value.length
}
if (typeof value !== 'object' || value === null) {
return 0
}
let longest = 0
for (const entry of Object.values(value)) {
longest = Math.max(longest, longestString(entry))
}
return longest
}
function ceil(value: unknown): unknown {
if (typeof value === 'string') {
return journalTextWithinCeiling(value)
}
if (typeof value !== 'object' || value === null) {
return value
}
if (!Array.isArray(value) && isBoundedPayload(value)) {
const head = journalTextWithinCeiling(value.head)
return head === value.head ? value : { ...value, head, truncated: true }
}
const entries = Object.entries(value)
const kept = entries.map(([, entry]) => ceil(entry))
if (kept.every((entry, index) => entry === entries[index]![1])) {
return value
}
return Array.isArray(value)
? kept
: Object.fromEntries(entries.map(([key], index) => [key, kept[index]]))
}
/** `body` itself when no text in it is past the ceiling; otherwise a copy with each such text cut
* to its head and tail. */
export function journalBodyWithinCeiling<T>(body: T): T {
// A string's UTF-16 length bounds its UTF-8 bytes from below by a third.
if (longestString(body) * 3 <= JOURNAL_PAYLOAD_MAX_BYTES) {
return body
}
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: `ceil` keeps every key and shape and only shortens strings.
return ceil(body) as T
}
@@ -0,0 +1,67 @@
import { afterEach, expect, it, vi } from 'vitest'
import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types'
import { AcpToolTimeline } from '../../acp/acp-tool-timeline'
import { PiRpcMessages } from '../../pi/rpc-messages'
import { PiRpcContextUsage } from '../../pi/rpc-context-usage'
import { JsonlRpcTimelineLane } from '../../jsonl-rpc/timeline-lane'
import { closeProviderTimelineRigs, openProviderTimelineRig, SESSION, GENERATION, NAMESPACE } from '../agent-session-timeline/provider-timeline-assembler-test-support'
import { estimateStructuredAgentSessionItemBytes } from '../agent-session-wire/structured-agent-session-event-sink-estimate'
import { journalPayload } from './journal-payload-bounds'
const cleanups: (() => void)[] = []
afterEach(async () => { cleanups.splice(0).forEach((cleanup) => cleanup()); await closeProviderTimelineRigs() })
it.each(['direct', 'transition'] as const)('persists complete content through %s admission and SQLite reopen', async (path) => {
const rig = await openProviderTimelineRig()
const gate = Promise.withResolvers<void>()
cleanups.push(() => gate.resolve())
const blocked = rig.journal.readInOrder(() => gate.promise)
const identity = { provider: 'legacy' as const, agent: 'grok', sessionId: SESSION, recordId: 'full' }
const options = { turnScope: AGENT_JOURNAL_THREAD_SCOPE, lifecycle: true }
const text = '\u0001'.repeat(3 * 1024 * 1024) + 'MIDDLE' + 'x'.repeat(5 * 1024 * 1024) + 'TAIL'
const body = { kind: 'tool-call' as const, name: 'bash', input: { large: 'i'.repeat(1024 * 1024) }, state: 'completed', output: journalPayload(text) }
expect(rig.eventSink.tryAppendItem?.({ ...identity, recordId: 'before' }, { kind: 'status', tone: 'info', text: 'before' }, options)).toEqual({ accepted: true })
const append = () => path === 'direct'
? rig.eventSink.tryAppendItem?.(identity, body, options)
: rig.sink.tryAppendTransition({ lifecycle: true, publish: true, steps: [{ kind: 'item', reservedBytes: estimateStructuredAgentSessionItemBytes(identity, body), options, resolve: () => ({ identity, body }) }] })
expect(append()).toEqual({ accepted: false, reason: 'backpressure' })
gate.resolve()
await blocked
await rig.eventSink.written?.()
expect(append()).toEqual({ accepted: true })
await rig.eventSink.written?.()
const reopened = await rig.reopenJournal()
expect(reopened.snapshot().items.map((item) => item.body)).toEqual([
{ kind: 'status', tone: 'info', text: 'before' }, body
])
})
it.each(['acp', 'pi'] as const)('transfers %s full fields at an omitted-field close without saving whole running revisions', async (agent) => {
const rig = await openProviderTimelineRig()
const onFailed = vi.fn()
const lane = new JsonlRpcTimelineLane({ sink: rig.sink, sessionId: SESSION, generation: GENERATION, namespace: NAMESPACE, agent: 'grok', pauseReading: vi.fn(), resumeReading: vi.fn(), onInputAccepted: vi.fn(), onFailed })
cleanups.push(() => lane.dispose())
const acp = new AcpToolTimeline(), pi = new PiRpcMessages(GENERATION, new PiRpcContextUsage())
const size = (agent === 'acp' ? 9 : 2) * 1024 * 1024
const input = { text: 'i'.repeat(size) }, output = 'o'.repeat(size)
for (const id of ['one', 'two']) {
lane.apply(agent === 'acp'
? acp.translate({ toolCallId: id, name: 'bash', status: 'in_progress', rawInput: input }, {}, {})
: pi.tool({ type: 'tool_execution_start', toolCallId: id, toolName: 'bash', args: input }))
await rig.rows(); lane.retry(); await lane.drained()
lane.apply(agent === 'acp' ? acp.translate({ toolCallId: id, rawOutput: output }, {}, {})
: pi.tool({ type: 'tool_execution_update', toolCallId: id, toolName: 'bash', partialResult: { content: [{ type: 'text', text: output }] } }))
await rig.rows(); lane.retry(); await lane.drained()
}
expect(agent === 'acp' ? acp.retainedToolBytes : pi.retainedToolBytes).toBeGreaterThan(4 * size)
expect((await rig.rows()).filter((row) => row.body.kind === 'tool-call').every((row) => JSON.stringify(row.body).length < 40_000)).toBe(true)
for (const id of ['one', 'two']) {
lane.apply(agent === 'acp' ? acp.translate({ toolCallId: id, status: 'completed' }, {}, {})
: pi.tool({ type: 'tool_execution_end', toolCallId: id, toolName: 'bash' }))
await rig.rows(); lane.retry(); await lane.drained()
}
expect(agent === 'acp' ? acp.retainedToolBytes : pi.retainedToolBytes).toBe(0)
expect(onFailed).not.toHaveBeenCalled()
const reopened = await rig.reopenJournal()
expect(reopened.snapshot().items.map((row) => row.body)).toEqual(['one', 'two'].map((callId) => expect.objectContaining({ kind: 'tool-call', callId, input, state: 'completed', output: journalPayload(output) })))
})
@@ -17,6 +17,7 @@ import {
importLegacyTranscriptIntoJournal,
prepareLegacyTranscriptImport
} from './journal-legacy-import'
import { LEGACY_IMPORT_BLOCK_LIMITS } from './journal-legacy-import'
import { openAgentSessionJournal } from './journal-store-factory'
import type { AgentSessionJournal } from './journal-store'
import { openTestJournalHostDatabase } from './journal-host-database-test-support'
@@ -406,8 +407,8 @@ async function importDecodedRoster(agents: RosterAgents): Promise<AgentSessionJo
return journal
}
describe('payloads on import', () => {
it('keeps an imported tool result whole', async () => {
describe('payload bounds on import', () => {
it('marks a clipped tool result and discards the remainder', async () => {
const output = 'y'.repeat(64 * 1024)
const filePath = await writeFixture('claude-big.jsonl', [
{
@@ -429,7 +430,7 @@ describe('payloads on import', () => {
agent: 'claude',
sessionId: CLAUDE_SESSION,
fence: 1,
options: { filePath }
options: { filePath, limits: { ...LEGACY_IMPORT_BLOCK_LIMITS, inlineHeadBytes: 1_024 } }
})
const item = journal.snapshot().items[0]
@@ -437,13 +438,16 @@ describe('payloads on import', () => {
const body = item?.body
const result = body?.kind === 'message' ? body.blocks[0] : undefined
if (result?.type !== 'tool-result') {
throw new Error('expected a tool result')
throw new Error('expected a bounded tool result')
}
expect(result.callId).toBe('toolu_9')
expect(result.output).toBe(output)
expect(result.output.startsWith('y'.repeat(1_024))).toBe(true)
expect(result.output).toContain('65536 bytes')
expect(result.output).toContain('output truncated')
expect(result.output.length).toBeLessThan(1_200)
})
it('bounds an imported subagent roster by entry count and id', async () => {
it('bounds an imported subagent roster by entry count, label and id', async () => {
// The import reads an untrusted file: nothing upstream bounded either string.
const oversized = 'z'.repeat(20 * 1024)
const journal = await importDecodedRoster(
@@ -460,6 +464,7 @@ describe('payloads on import', () => {
throw new Error('expected a subagent-group block')
}
expect(block.agents).toHaveLength(64)
expect(block.agents[0]?.label.length).toBeLessThan(oversized.length)
expect(block.agents[0]?.id.length).toBeLessThan(oversized.length)
expect(block.agents[0]?.id.startsWith('z')).toBe(true)
})
@@ -552,6 +557,55 @@ describe('import failures', () => {
expect(texts.length).toBeLessThan(lines.length)
})
it('bounds oversized legacy tool-call input before journal publication', async () => {
const journalDir = join(root, 'bounded-tool-input-journal')
const limits = { ...LEGACY_IMPORT_BLOCK_LIMITS, inlineHeadBytes: 64 }
const journal = await open('claude', CLAUDE_SESSION, {
database: openTestJournalHostDatabase(journalDir)
})
const filePath = await writeFixture('oversized-tool-input.jsonl', [
{
parentUuid: null,
isSidechain: false,
type: 'assistant',
message: {
role: 'assistant',
content: [
{
type: 'tool_use',
id: 'toolu_large_input',
name: 'Edit',
input: { file_path: 'a.ts', patch: 'x'.repeat(10_000) }
}
]
},
uuid: 'cc11ad00-1111-4222-8333-444455556666',
timestamp: '2026-08-05T10:00:09.000Z',
sessionId: CLAUDE_SESSION
}
])
const result = await importLegacyTranscriptIntoJournal({
journal,
agent: 'claude',
sessionId: CLAUDE_SESSION,
fence: 1,
options: { filePath, limits }
})
expect(result.ok).toBe(true)
const imported = journal.snapshot().items[0]
expect(imported?.body).toMatchObject({
kind: 'tool-call',
input: {
truncated: true,
byteLength: expect.any(Number),
digest: expect.stringMatching(/^[0-9a-f]{64}$/),
head: expect.any(String)
}
})
expect(JSON.stringify(imported?.body)).not.toContain('x'.repeat(1_000))
})
it('reports a missing transcript without touching the journal', async () => {
const journal = await open('claude', CLAUDE_SESSION)
const before = journal.epoch
@@ -610,3 +664,132 @@ describe('import failures', () => {
expect(result).toMatchObject({ ok: false })
})
})
// A tool call is only the SOLE block of its message when the provider wrote it
// that way. Claude interleaves it with narration, Grok hangs `tool_calls` off a
// row that also has text, and omp's execution cells always pair the invocation
// with its output — so the multi-block path carries untrusted tool input too.
describe('multi-block legacy messages', () => {
const limits = { ...LEGACY_IMPORT_BLOCK_LIMITS, inlineHeadBytes: 64 }
const oversized = 'x'.repeat(10_000)
/** The tool-call block of the first imported multi-block message. */
function importedToolCallBlock(journal: AgentSessionJournal): unknown {
for (const entry of journal.snapshot().items) {
if (entry.body.kind !== 'message') {
continue
}
const block = entry.body.blocks.find((candidate) => candidate.type === 'tool-call')
if (block) {
return block.input
}
}
return null
}
it('bounds a Claude tool call that shares its message with narration', async () => {
const journal = await open('claude', CLAUDE_SESSION, {
database: openTestJournalHostDatabase(join(root, 'claude-mixed-journal'))
})
const filePath = await writeFixture('claude-mixed.jsonl', [
{
parentUuid: null,
isSidechain: false,
type: 'assistant',
message: {
role: 'assistant',
content: [
{ type: 'text', text: 'Editing the file.' },
{
type: 'tool_use',
id: 'toolu_mixed',
name: 'Edit',
input: { file_path: 'a.ts', patch: oversized }
}
]
},
uuid: 'dd22be00-1111-4222-8333-444455556666',
timestamp: '2026-08-05T10:00:09.000Z',
sessionId: CLAUDE_SESSION
}
])
const result = await importLegacyTranscriptIntoJournal({
journal,
agent: 'claude',
sessionId: CLAUDE_SESSION,
fence: 1,
options: { filePath, limits }
})
expect(result.ok).toBe(true)
expect(importedToolCallBlock(journal)).toMatchObject({
truncated: true,
byteLength: expect.any(Number),
digest: expect.stringMatching(/^[0-9a-f]{64}$/),
head: expect.any(String)
})
expect(JSON.stringify(journal.snapshot().items)).not.toContain('x'.repeat(1_000))
await journal.close()
})
it('bounds a Grok tool call that shares its row with assistant text', async () => {
const journal = await open('grok', CODEX_SESSION, {
database: openTestJournalHostDatabase(join(root, 'grok-mixed-journal'))
})
const filePath = await writeFixture('grok-mixed.jsonl', [
{
type: 'assistant',
id: 'asst-mixed',
timestamp: '2026-08-05T10:00:09.000Z',
content: [{ type: 'text', text: 'Searching.' }],
tool_calls: [{ id: 'c1', name: 'grep', arguments: JSON.stringify({ pattern: oversized }) }]
}
])
const result = await importLegacyTranscriptIntoJournal({
journal,
agent: 'grok',
sessionId: CODEX_SESSION,
fence: 1,
options: { filePath, limits }
})
expect(result.ok).toBe(true)
expect(importedToolCallBlock(journal)).toMatchObject({ truncated: true })
expect(JSON.stringify(journal.snapshot().items)).not.toContain('x'.repeat(1_000))
await journal.close()
})
it('bounds an omp execution cell, whose invocation always ships with its output', async () => {
const journal = await open('omp', CODEX_SESSION, {
database: openTestJournalHostDatabase(join(root, 'omp-mixed-journal'))
})
const filePath = await writeFixture('omp-mixed.jsonl', [
{
type: 'message',
id: 'omp-mixed-1',
timestamp: '2026-08-05T10:00:09.000Z',
message: {
role: 'bashExecution',
command: `echo ${oversized}`,
output: 'done',
exitCode: 0
}
}
])
const result = await importLegacyTranscriptIntoJournal({
journal,
agent: 'omp',
sessionId: CODEX_SESSION,
fence: 1,
options: { filePath, limits }
})
expect(result.ok).toBe(true)
expect(importedToolCallBlock(journal)).toMatchObject({ truncated: true })
expect(JSON.stringify(journal.snapshot().items)).not.toContain('x'.repeat(1_000))
await journal.close()
})
})
@@ -38,12 +38,21 @@ import { createLegacyIdentityTracker } from './journal-legacy-identity'
import { createPiTranscriptCollector } from './journal-pi-active-branch'
import { createNewestItemWindow } from './journal-newest-item-window'
import type { JournalReplacementItem } from './journal-epoch-replacement'
import {
boundInlineText,
boundToolInput,
type JournalPayloadLimits
} from './journal-payload-bounds'
import type { AgentSessionJournal } from './journal-store'
/** Import preserves the bounded history window; the transcript remains the complete source. */
export const LEGACY_IMPORT_BLOCK_LIMITS: JournalPayloadLimits = { inlineHeadBytes: 16 * 1024 }
export type LegacyImportOptions = ResolveSessionFileOptions & {
/** Resolve directly to this file, skipping path discovery. */
filePath?: string
source?: AdoptedTranscriptSource
limits?: JournalPayloadLimits
decodedMessageIdentities?: true
}
@@ -96,6 +105,7 @@ export async function prepareLegacyTranscriptImport(input: {
options?: LegacyImportOptions
}): Promise<PreparedLegacyImport> {
const options = input.options ?? {}
const limits = options.limits ?? LEGACY_IMPORT_BLOCK_LIMITS
// OMP is a Pi fork that kept Pi's session format, so OMP's decoder reads Pi's sessions too.
const transcriptAgent =
input.agent === 'pi' ? 'omp' : resolveNativeChatTranscriptAgent(input.agent)
@@ -103,7 +113,7 @@ export async function prepareLegacyTranscriptImport(input: {
return { ok: false, error: `Unsupported agent for journal import: ${input.agent}` }
}
if (transcriptAgent === 'opencode') {
return prepareOpenCodeImport(input)
return prepareOpenCodeImport({ ...input, limits })
}
const filePath =
options.filePath ?? (await resolveSessionFilePath(input.agent, input.sessionId, options))
@@ -118,7 +128,8 @@ export async function prepareLegacyTranscriptImport(input: {
agent: input.agent,
sessionId: input.sessionId,
source: options.source,
decodedMessageIdentities: options.decodedMessageIdentities
decodedMessageIdentities: options.decodedMessageIdentities,
limits
})
return { ok: true, items, sourceEmpty: consumedBytes === 0 }
} catch (err) {
@@ -132,6 +143,7 @@ async function prepareOpenCodeImport(input: {
agent: AgentType
sessionId: string
options?: LegacyImportOptions
limits: JournalPayloadLimits
}): Promise<PreparedLegacyImport> {
const dbPath = input.options?.filePath
if (input.options?.source) {
@@ -157,19 +169,20 @@ async function prepareOpenCodeImport(input: {
sessionId: input.sessionId,
recordId: message.id
}))
const items = replacementItems(read.messages, identities)
const items = replacementItems(read.messages, identities, input.limits)
return { ok: true, items, sourceEmpty: read.sourceEmpty === true }
}
function replacementItems(
messages: readonly NativeChatMessage[],
identities: readonly AgentJournalItemIdentity[]
identities: readonly AgentJournalItemIdentity[],
limits: JournalPayloadLimits
): JournalReplacementItem[] {
const replacement: JournalReplacementItem[] = []
for (const [index, message] of messages.entries()) {
const identity = identities[index]
if (identity) {
replacement.push(replacementItem(message, identity))
replacement.push(replacementItem(message, identity, limits))
}
}
return replacement
@@ -177,11 +190,12 @@ function replacementItems(
function replacementItem(
message: NativeChatMessage,
identity: AgentJournalItemIdentity
identity: AgentJournalItemIdentity,
limits: JournalPayloadLimits
): JournalReplacementItem {
return {
identity,
body: legacyItemBody(message),
body: legacyItemBody(message, limits),
observedAt: message.timestamp ?? undefined,
// Imported history is no generation's work: never current, whatever it says still runs.
ownerFence: 0
@@ -195,8 +209,9 @@ const TRANSCRIPT_DECODERS = {
omp: decodeOmpTranscriptLine
} as const
/** Run the real decoder while recording an identity anchor per emitted message. Only the newest
* messages fit the retained budget, so memory tracks what is kept, never the source's size. */
/** Run the real decoder while recording an identity anchor per emitted message. Each message is
* bounded as it is read and only the newest fit the retained budget, so memory tracks what is
* kept, never the source's size. */
async function decodeReplacementItems(input: {
filePath: string
transcriptAgent: keyof typeof TRANSCRIPT_DECODERS
@@ -204,6 +219,7 @@ async function decodeReplacementItems(input: {
sessionId: string
decodedMessageIdentities?: true
source?: AdoptedTranscriptSource
limits: JournalPayloadLimits
}): Promise<{ items: JournalReplacementItem[]; consumedBytes: number }> {
const source = input.source
? adoptedTranscriptSourceObserver({
@@ -220,7 +236,7 @@ async function decodeReplacementItems(input: {
})
const decode = TRANSCRIPT_DECODERS[input.transcriptAgent]
const toItem = (message: NativeChatMessage, identity: AgentJournalItemIdentity) =>
replacementItem(message, identity)
replacementItem(message, identity, input.limits)
const retained = createNewestItemWindow(MAX_LEGACY_IMPORT_RETAINED_BYTES)
const pi =
input.agent === 'pi'
@@ -275,16 +291,22 @@ async function decodeReplacementItems(input: {
/**
* A message whose only content is a tool invocation becomes a tool-call item so
* the reducer renders it as one. Everything else stays a message item. Content is
* kept whole; only the copy sent to clients is bounded.
* the reducer renders it as one. Everything else stays a message item with its
* blocks bounded in place.
*/
function legacyItemBody(message: NativeChatMessage): AgentJournalItemBody {
function legacyItemBody(
message: NativeChatMessage,
limits: JournalPayloadLimits
): AgentJournalItemBody {
const only = message.blocks.length === 1 ? message.blocks[0] : undefined
if (only?.type === 'tool-call') {
// Legacy transcripts are untrusted and can contain arbitrarily large tool
// arguments. Keep them on the same bounded path as live events before the
// replacement epoch is published.
return {
kind: 'tool-call',
name: only.name,
input: only.input,
input: boundToolInput(only.input, limits),
...(only.callId !== undefined ? { callId: only.callId } : {}),
state: 'completed'
}
@@ -292,12 +314,25 @@ function legacyItemBody(message: NativeChatMessage): AgentJournalItemBody {
return {
kind: 'message',
role: message.role,
blocks: message.blocks.map(boundRoster)
blocks: message.blocks.map((block) => boundBlock(block, limits))
}
}
/** A roster's entries are keys, so an imported one keeps a bounded count and bounded ids. */
function boundRoster(block: NativeChatBlock): NativeChatBlock {
/** Every block that can carry untrusted bulk is bounded here, tool calls
* included: a provider decoder is free to put one alongside narration, and the
* sole-block path above never sees those. The remainder is discarded rather
* than stored elsewhere — the marker keeps its digest and byte length, and the
* source transcript remains the full copy. */
function boundBlock(block: NativeChatBlock, limits: JournalPayloadLimits): NativeChatBlock {
if (block.type === 'text') {
return { ...block, text: boundInlineText(block.text, limits).text }
}
if (block.type === 'tool-result') {
return { ...block, output: boundInlineText(block.output, limits).text }
}
if (block.type === 'tool-call') {
return { ...block, input: boundToolInput(block.input, limits) }
}
if (block.type === 'subagent-group') {
return {
...block,
@@ -305,7 +340,8 @@ function boundRoster(block: NativeChatBlock): NativeChatBlock {
...agent,
// The id is the roster key, so it is bounded with a digest rather than
// clipped to a prefix that two distinct children could share.
id: boundSubagentEntryId(agent.id)
id: boundSubagentEntryId(agent.id),
label: boundInlineText(agent.label, limits).text
}))
}
}
@@ -1,11 +1,4 @@
// Journal payloads: what a row keeps of a tool output or diff.
//
// A row keeps the whole value; only what is sent to a client is bounded (the item wire form in
// agent-session-wire), so no writer can turn a transport limit into lost content. Two bounds stay:
// any text past `JOURNAL_PAYLOAD_MAX_BYTES` keeps its head and tail (journal-body-ceiling.ts), and
// a Grok, OpenCode, OMP or Pi call still running keeps a preview that its closing revision
// replaces. The byte clip below also serves keys and the few fields with their own named limit.
// Completed content is canonical; running revisions keep previews until the full close.
import { createHash } from 'node:crypto'
import type { AgentJournalBoundedPayload } from '../../../shared/agent-session-journal-types'
@@ -14,10 +7,6 @@ export type JournalPayloadLimits = {
inlineHeadBytes: number
}
/** Past this one write outgrows the journal write queue (16 MiB per lifecycle write), which stalls
* the chat, so a larger text keeps its head and tail, where a log's command and errors are. */
export const JOURNAL_PAYLOAD_MAX_BYTES = 4 * 1024 * 1024
/** What a revision of a call still running keeps where every update is a row and the open budget
* holds it (provider-timeline-budget.ts); its closing revision keeps the whole value. */
export const JOURNAL_RUNNING_PREVIEW_LIMITS: JournalPayloadLimits = {
@@ -53,23 +42,6 @@ export function journalPayload(payload: string): AgentJournalBoundedPayload {
}
}
/** `text` itself within `JOURNAL_PAYLOAD_MAX_BYTES`; past it, its head and tail. */
export function journalTextWithinCeiling(text: string): string {
// UTF-8 spends at most three bytes per UTF-16 unit, so most text needs no encoding.
if (text.length * 3 <= JOURNAL_PAYLOAD_MAX_BYTES) {
return text
}
const byteLength = Buffer.byteLength(text, 'utf8')
if (byteLength <= JOURNAL_PAYLOAD_MAX_BYTES) {
return text
}
const keep = JOURNAL_PAYLOAD_MAX_BYTES / 2
const head = utf8Head(text, keep)
const tail = utf8Tail(text, keep)
const omitted = byteLength - Buffer.byteLength(head, 'utf8') - Buffer.byteLength(tail, 'utf8')
return `${head}\n… [Orca kept the start and end; ${omitted} bytes in between were not saved] …\n${tail}`
}
/** Whole once `running` is false; a preview while the call still runs. */
export function journalToolPayload(payload: string, running: boolean): AgentJournalBoundedPayload {
return running ? boundPayload(payload, JOURNAL_RUNNING_PREVIEW_LIMITS) : journalPayload(payload)
@@ -137,16 +109,6 @@ export function boundToolInput(input: unknown, limits: JournalPayloadLimits): un
: input
}
/** The last `maxBytes` of `text` as UTF-8, never splitting a character. */
function utf8Tail(text: string, maxBytes: number): string {
const buffer = Buffer.from(text, 'utf8')
let start = Math.max(0, buffer.length - maxBytes)
while (start < buffer.length && (buffer[start] & 0b1100_0000) === 0b1000_0000) {
start += 1
}
return buffer.subarray(start).toString('utf8')
}
/** Slice at a byte budget without splitting a multi-byte character. */
function clipUtf8(buffer: Buffer, maxBytes: number): string {
let end = maxBytes
@@ -0,0 +1,19 @@
import { expect, it } from 'vitest'
import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types'
import { journalPayload } from '../agent-session-journal/journal-payload-bounds'
import { HISTORY_PAGE_CONTENT_BUDGET_BYTES } from './agent-session-history-page-bounds'
import { agentJournalItemForWire, shortenedAgentJournalItem } from './agent-session-item-wire-form'
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)
})
it.each([73_728, HISTORY_PAGE_CONTENT_BUDGET_BYTES, 4 * 1024 * 1024])('keeps tool identity and output when cardinality exceeds %i bytes', (budget) => {
const item: AgentJournalRenderItem = { itemId: 'call', revision: 1, sequence: 1, observedAt: 1,
body: { kind: 'tool-call', callId: 'call', name: 'query', state: 'completed', input: { values: Array(Math.ceil(budget / 2) + 1_000).fill(0) }, output: journalPayload('done') } }
const sent = shortenedAgentJournalItem(item, budget)
expect(sent).toMatchObject({ itemId: 'call', shortened: true, body: { kind: 'tool-call', name: 'query', input: { truncated: true }, output: { head: 'done' } } })
expect(Buffer.byteLength(JSON.stringify(sent))).toBeLessThanOrEqual(budget)
})
@@ -2,7 +2,7 @@
//
// The journal keeps tool output and diffs whole. A client is sent a head of each and the item is
// marked `shortened`; it fetches the whole item with `agentSession.item` when the user asks. Every
// other field is sent as kept. An item still past half a page is shortened to fit and marked
// other field is sent as kept. An item still past a page is shortened to fit and marked
// rather than replaced, so its row and its "show full" stay readable.
import type {
@@ -10,7 +10,7 @@ import type {
AgentJournalItemBody,
AgentJournalRenderItem
} from '../../../shared/agent-session-journal-types'
import { utf8Head } from '../agent-session-journal/journal-payload-bounds'
import { boundToolInput, JOURNAL_RUNNING_PREVIEW_LIMITS, utf8Head } from '../agent-session-journal/journal-payload-bounds'
import {
HISTORY_PAGE_CONTENT_BUDGET_BYTES,
type HistoryReader
@@ -20,8 +20,8 @@ import { jsonBytes, shortenJsonStringsToFit } from './json-string-shortening'
/** Each tool output or diff as sent: the head older clients have always been sent. */
export const AGENT_SESSION_WIRE_PAYLOAD_HEAD_BYTES = 16 * 1024
/** Half a page, so an item and the submission it carries always fit one. */
const CLIENT_ITEM_MAX_BYTES = HISTORY_PAGE_CONTENT_BUDGET_BYTES / 2
/** Preserve every ordinary message that already fits a desktop page. */
const CLIENT_ITEM_MAX_BYTES = HISTORY_PAGE_CONTENT_BUDGET_BYTES
/** The payload as sent; itself when its head already fits. */
function wirePayload(payload: AgentJournalBoundedPayload): AgentJournalBoundedPayload {
@@ -60,10 +60,14 @@ export function shortenedAgentJournalItem(
if (jsonBytes(item) <= maxBytes) {
return item
}
const body = shortenJsonStringsToFit(
item.body,
maxBytes - (jsonBytes({ ...item, body: null, shortened: true }) - jsonBytes(null))
)
const budget = maxBytes - (jsonBytes({ ...item, body: null, shortened: true }) - jsonBytes(null))
let body = shortenJsonStringsToFit(item.body, budget)
if (body === null && item.body.kind === 'tool-call') {
body = shortenJsonStringsToFit(
{ ...item.body, input: boundToolInput(item.body.input, JOURNAL_RUNNING_PREVIEW_LIMITS) },
budget
)
}
return body === null ? null : { ...item, body, shortened: true }
}
@@ -176,14 +176,16 @@ export class StructuredAgentSessionSinkQueue {
this.lifecycleQueuedBytes - (replaced ? lifecycleCost(replaced) : 0) + lifecycleBytes
const nextLifecycleOperations =
this.lifecycleQueuedOperations - (replaced?.lifecycle ? 1 : 0) + (lifecycle ? 1 : 0)
// A byte watermark paces waiting work; an empty budget can own one complete operation.
const exceedsOrdinary =
!finalTail &&
!lifecycle &&
(nextBytes > this.deps.watermarks.maxQueuedBytes ||
((this.queuedOperations > 0 && nextBytes > this.deps.watermarks.maxQueuedBytes) ||
nextOperations > this.deps.watermarks.maxQueuedOperations)
const exceedsLifecycle =
lifecycle &&
(nextLifecycleBytes > this.deps.watermarks.maxLifecycleQueuedBytes ||
((this.lifecycleQueuedOperations > 0 &&
nextLifecycleBytes > this.deps.watermarks.maxLifecycleQueuedBytes) ||
nextLifecycleOperations > this.deps.watermarks.maxLifecycleQueuedOperations)
if (exceedsOrdinary || exceedsLifecycle) {
this.setBackpressure(true)
@@ -8,8 +8,9 @@ import type {
import type { AgentSessionTurnActivity } from '../../../shared/agent-session-wire'
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
import type { JournalLifecycleIdentityMutationInput } from '../agent-session-journal/journal-row-builders'
import { estimateStructuredAgentSessionItemBytes } from './structured-agent-session-event-sink-estimate'
import { StructuredAgentSessionSinkQueue } from './structured-agent-session-event-sink-queue'
import { createStructuredAgentSessionItemAppend } from './structured-agent-session-item-append'
import { structuredAgentSessionJournalAppendOptions } from './structured-agent-session-journal-append-options'
import { createStructuredAgentSessionResolvedAppend } from './structured-agent-session-resolved-append'
import {
createStructuredAgentSessionTransitionMembers,
@@ -231,7 +232,27 @@ export function createDeferredStructuredAgentSessionEventSink(deps: {
...(deps.onBackpressureChange ? { onBackpressureChange: deps.onBackpressureChange } : {})
})
const resolvedAppend = createStructuredAgentSessionResolvedAppend(queue)
const { appendItem, appendLifecycleBatch } = createStructuredAgentSessionItemAppend(queue)
const appendLifecycleBatch = (
settlementId: string,
mutations: readonly JournalLifecycleIdentityMutationInput[],
options: StructuredAgentSessionAppendOptions = {}
): StructuredAgentSessionSinkAdmission =>
queue.submit(
{
bytes: Buffer.byteLength(JSON.stringify({ settlementId, mutations }), 'utf8') + 512,
run: (bound) =>
bound.journal.appendLifecycleBatch({
settlementId,
mutations,
// No row-level linkage: each mutation names its own (see the batch row builder).
fence: bound.fence,
// The provider's own observation: its child's generation produced every item it creates.
ownerFence: bound.fence
})
},
{ ...options, lifecycle: true }
)
const publish = (
options: StructuredAgentSessionPublishOptions = {}
@@ -245,6 +266,24 @@ export function createDeferredStructuredAgentSessionEventSink(deps: {
options
)
const appendItem: NonNullable<StructuredAgentSessionEventSink['tryAppendItem']> = (
identity,
body,
options
) =>
queue.submit(
{
bytes: estimateStructuredAgentSessionItemBytes(identity, body),
run: (bound) =>
bound.journal.appendItem(
identity,
body,
structuredAgentSessionJournalAppendOptions(bound.fence, options)
)
},
options
)
return {
sink: {
appendItem: (identity, body, options) => {
@@ -1,7 +1,6 @@
const MAX_FINAL_TAIL_BYTES = 16 * 1024 * 1024
const MAX_FINAL_TAIL_OPERATIONS = 512
/** Finalized providers can admit their bounded tail before the host takes its exit barrier. */
/** A sealed lane transfers already-observed work before the host takes its exit barrier. */
export class StructuredAgentSessionFinalTailReservation {
private bytes = 0
private operations = 0
@@ -12,7 +11,6 @@ export class StructuredAgentSessionFinalTailReservation {
this.closed ||
!Number.isSafeInteger(bytes) ||
bytes < 0 ||
this.bytes + bytes > MAX_FINAL_TAIL_BYTES ||
this.operations >= MAX_FINAL_TAIL_OPERATIONS
) {
return null
@@ -1,69 +0,0 @@
// Plain item and lifecycle-batch appends. Each body is held to the journal body ceiling as it
// enters the write queue, so no one write outgrows the queue's caps and stalls the chat.
import { journalBodyWithinCeiling } from '../agent-session-journal/journal-body-ceiling'
import type { JournalLifecycleIdentityMutationInput } from '../agent-session-journal/journal-row-builders'
import type {
StructuredAgentSessionAppendOptions,
StructuredAgentSessionEventSink,
StructuredAgentSessionSinkAdmission
} from './structured-agent-session-event-sink'
import { estimateStructuredAgentSessionItemBytes } from './structured-agent-session-event-sink-estimate'
import type { StructuredAgentSessionSinkQueue } from './structured-agent-session-event-sink-queue'
import { structuredAgentSessionJournalAppendOptions } from './structured-agent-session-journal-append-options'
function mutationWithinCeiling(
mutation: JournalLifecycleIdentityMutationInput
): JournalLifecycleIdentityMutationInput {
if (mutation.kind !== 'item') {
return mutation
}
const body = journalBodyWithinCeiling(mutation.body)
return body === mutation.body ? mutation : { ...mutation, body }
}
export function createStructuredAgentSessionItemAppend(queue: StructuredAgentSessionSinkQueue): {
appendItem: NonNullable<StructuredAgentSessionEventSink['tryAppendItem']>
appendLifecycleBatch: (
settlementId: string,
mutations: readonly JournalLifecycleIdentityMutationInput[],
options?: StructuredAgentSessionAppendOptions
) => StructuredAgentSessionSinkAdmission
} {
return {
appendItem: (identity, written, options) => {
const body = journalBodyWithinCeiling(written)
return queue.submit(
{
bytes: estimateStructuredAgentSessionItemBytes(identity, body),
run: (bound) =>
bound.journal.appendItem(
identity,
body,
structuredAgentSessionJournalAppendOptions(bound.fence, options)
)
},
options
)
},
appendLifecycleBatch: (settlementId, written, options = {}) => {
const mutations = written.map(mutationWithinCeiling)
return queue.submit(
{
bytes: Buffer.byteLength(JSON.stringify({ settlementId, mutations }), 'utf8') + 512,
run: (bound) =>
bound.journal.appendLifecycleBatch({
settlementId,
mutations,
// No row-level linkage: each mutation names its own (see the batch row builder).
fence: bound.fence,
// The provider's own observation: its child's generation produced every item it
// creates.
ownerFence: bound.fence
})
},
{ ...options, lifecycle: true }
)
}
}
}
@@ -3,7 +3,6 @@ import type {
AgentJournalItemIdentity
} from '../../../shared/agent-session-journal-types'
import { estimateStructuredAgentSessionItemBytes } from './structured-agent-session-event-sink-estimate'
import { journalBodyWithinCeiling } from '../agent-session-journal/journal-body-ceiling'
import type {
StructuredAgentSessionItemAppendOptions,
StructuredAgentSessionEventSink,
@@ -63,9 +62,8 @@ export function createStructuredAgentSessionResolvedAppend(
)
const ITEM_OVERFLOW = 'structured agent-session resolved item exceeded its reserved size'
const identityOnly = (publish: boolean) =>
((identitySizeBound, written, resolveIdentity, options) => {
const body = journalBodyWithinCeiling(written)
return submit(
((identitySizeBound, body, resolveIdentity, options) =>
submit(
estimateStructuredAgentSessionItemBytes(identitySizeBound, body) + (publish ? 1 : 0),
(journal) => {
const identity = resolveIdentity(journal)
@@ -73,8 +71,7 @@ export function createStructuredAgentSessionResolvedAppend(
},
options,
{ publish, overflow: ITEM_OVERFLOW }
)
}) satisfies NonNullable<StructuredAgentSessionEventSink['tryAppendResolvedItem']>
)) satisfies NonNullable<StructuredAgentSessionEventSink['tryAppendResolvedItem']>
return {
tryAppendResolvedItem: identityOnly(false),
tryAppendResolvedItemAndPublish: identityOnly(true),
+25 -4
View File
@@ -1,4 +1,6 @@
import { z } from 'zod'
import { BoundedMap } from '../../shared/bounded-map'
import { MAX_PROVIDER_TIMELINE_OPEN_ENTRIES } from '../native-chat/agent-session-timeline/provider-timeline-budget'
import type { AgentJournalToolCallItem } from '../../shared/agent-session-journal-types'
import {
boundToolInput,
@@ -19,9 +21,17 @@ export class PiRpcMessages {
private active?: number
private readonly text = new Map<string, { index: number; channel: 'assistant' | 'reasoning' }>()
private readonly tools = new Map<string, AgentJournalToolCallItem>()
/** A running call's whole input and output, apart from `tools`, which holds the previews the open
* budget counts, so its closing revision keeps them though Pi's end event repeats no args. */
private readonly whole = new Map<string, { input?: unknown; output?: string }>()
// Closing events take these latest full fields; running journal rows retain only previews.
private readonly whole = new BoundedMap<string, {
input?: unknown; output?: string; inputBytes: number; outputBytes: number
}>({
maxEntries: MAX_PROVIDER_TIMELINE_OPEN_ENTRIES,
sizeOf: (value, key) => value.inputBytes + value.outputBytes + Buffer.byteLength(key) + 64
})
get retainedToolBytes(): number {
return this.whole.retainedBytes
}
constructor(
private readonly generation: string,
@@ -141,7 +151,18 @@ export class PiRpcMessages {
throw new Error('Pi tool stream capacity exceeded')
}
this.tools.set(event.toolCallId, body)
this.whole.set(event.toolCallId, { input, output })
if (!held && this.whole.size >= MAX_PROVIDER_TIMELINE_OPEN_ENTRIES) {
throw new Error('Pi tool stream capacity exceeded')
}
if (!this.whole.set(event.toolCallId, {
input, output,
inputBytes: event.args === undefined ? (held?.inputBytes ?? 0)
: Buffer.byteLength(JSON.stringify(input) ?? ''),
outputBytes: result === undefined ? (held?.outputBytes ?? 0)
: Buffer.byteLength(JSON.stringify(output) ?? '')
})) {
throw new Error('Pi tool snapshot could not be retained')
}
if (
[...this.tools.values()].reduce(
(bytes, tool) => bytes + Buffer.byteLength(JSON.stringify(tool)),
+3 -5
View File
@@ -258,11 +258,9 @@ export class PiRpcSession {
this.publishedExit = true
this.dialogs.cancelAll()
this.turns.end()
this.lane.finalize()
this.lane.apply([
{ type: 'session.ended', verdict: { state: 'interrupted', completedAt: Date.now() } }
])
this.lane.flush()
this.lane.finalize({
type: 'session.ended', verdict: { state: 'interrupted', completedAt: Date.now() }
})
let timer: ReturnType<typeof setTimeout> | undefined
try {
await Promise.race([
@@ -0,0 +1,42 @@
// @vitest-environment happy-dom
import { act } from 'react'
import { createRoot } from 'react-dom/client'
import { afterEach, expect, it, vi } from 'vitest'
import type { AgentJournalRenderItem } from '../../../../shared/agent-session-journal-types'
import type { NativeChatToolResultBlock } from '../../../../shared/native-chat-types'
import { NativeChatToolOutput } from './NativeChatToolOutput'
const mocks = vi.hoisted(() => ({ read: vi.fn(), owner: { target: { kind: 'local' }, sessionId: 'session', worktreeId: 'workspace' } }))
vi.mock('./native-chat-visual-owner', () => ({ useNativeChatVisualOwner: () => mocks.owner }))
vi.mock('@/runtime/structured-agent-session-client', () => ({ supportsStructuredAgentSessionItemDetail: async () => true, readStructuredAgentSessionItem: mocks.read }))
const container = document.createElement('div')
const root = createRoot(container)
afterEach(() => { act(() => root.unmount()); mocks.read.mockReset() })
it('fences same-head live revisions and late replies, and keeps null/partial results truthful', async () => {
const body: NativeChatToolResultBlock = { type: 'tool-result', output: 'x'.repeat(16_384), fullOutputItemId: 'call' }
const response = (text: string, truncated = false): AgentJournalRenderItem => ({ itemId: 'call', revision: 2, sequence: 1, observedAt: 1, body: { kind: 'tool-call', name: 'bash', input: null, state: 'running', output: { head: text, byteLength: 100_000, digest: 'd', truncated } } })
const render = (block: NativeChatToolResultBlock) => act(() => root.render(<NativeChatToolOutput body={block} />))
const click = async () => { await act(async () => container.querySelector('button')?.click()) }
render(body)
mocks.read.mockResolvedValueOnce(response('OLD TAIL'))
await click()
expect(container.textContent).toContain('OLD TAIL')
render({ ...body })
expect(container.textContent).not.toContain('OLD TAIL')
expect(container.querySelector('button')).not.toBeNull()
const pending = Promise.withResolvers<AgentJournalRenderItem>()
mocks.read.mockReturnValueOnce(pending.promise)
await click()
render({ ...body })
await act(async () => pending.resolve(response('STALE TAIL')))
expect(container.textContent).not.toContain('STALE TAIL')
mocks.read.mockResolvedValueOnce(null)
await click()
expect(container.textContent).toContain("Couldn't load the full output.")
expect(container.querySelector('button')).not.toBeNull()
mocks.read.mockResolvedValueOnce(response('PARTIAL', true))
await click()
expect(container.textContent).toContain('Showing')
expect(container.textContent).toContain('PARTIAL')
})
@@ -1,4 +1,4 @@
import { useState } from 'react'
import { useEffect, useState } from 'react'
import { Button } from '@/components/ui/button'
import { cn } from '@/lib/utils'
import { translate } from '@/i18n/i18n'
@@ -16,8 +16,7 @@ import { MAX_TOOL_DETAIL_LENGTH } from '../../../../shared/native-chat-tool-summ
import { useNativeChatVisualOwner } from './native-chat-visual-owner'
import { truncateToolDetail } from './native-chat-tool-summary'
/** `partial`: Orca kept only part of this output (past its storage ceiling) or the item was more
* than one reply carries. */
/** A partial detail response retains its original byte count. */
type FullOutput = { text: string; partial: { shown: number; total: number } | null }
function fullPayload(body: AgentJournalItemBody): AgentJournalBoundedPayload | null {
@@ -35,44 +34,49 @@ export function NativeChatToolOutput({
body: NativeChatToolResultBlock
}): React.JSX.Element {
const owner = useNativeChatVisualOwner()
const [full, setFull] = useState<FullOutput | null>(null)
const [loading, setLoading] = useState(false)
const [failed, setFailed] = useState(false)
const [expanded, setExpanded] = useState(false)
const [detail, setDetail] = useState<{
body: NativeChatToolResultBlock; owner: typeof owner; full?: FullOutput;
loading?: boolean; failed?: boolean
} | null>(null)
useEffect(() => () => setDetail((held) => held?.body === body && held.owner === owner ? null : held), [body, owner])
const current = detail?.body === body && detail.owner === owner ? detail : null
const full = current?.full
const loading = current?.loading === true
const failed = current?.failed === true
const clipped =
full === null &&
(body.fullOutputItemId !== undefined || body.output.length > MAX_TOOL_DETAIL_LENGTH)
!full &&
(body.fullOutputItemId !== undefined || (!expanded && body.output.length > MAX_TOOL_DETAIL_LENGTH))
const showFull = async (): Promise<void> => {
const itemId = body.fullOutputItemId
if (itemId === undefined || owner === null) {
setFull({ text: body.output, partial: null })
setExpanded(true)
return
}
setLoading(true)
setFailed(false)
setDetail({ body, owner, loading: true })
const finish = (result: { full?: FullOutput; failed?: boolean }): void => {
setDetail((held) => held?.body === body && held.owner === owner ? { body, owner, ...result } : held)
}
try {
// An older host keeps only the head it already sent.
const item = (await supportsStructuredAgentSessionItemDetail(owner.target))
? await readStructuredAgentSessionItem(owner.target, owner.sessionId, itemId)
: null
const payload = item ? fullPayload(item.body) : null
setFull(
payload
? {
text: payload.head,
partial: payload.truncated
? {
shown: new TextEncoder().encode(payload.head).length,
total: payload.byteLength
}
: null
}
: { text: body.output, partial: null }
)
if (!(await supportsStructuredAgentSessionItemDetail(owner.target))) {
setExpanded(true)
finish({})
return
}
const item = await readStructuredAgentSessionItem(owner.target, owner.sessionId, itemId)
const payload = item?.itemId === itemId ? fullPayload(item.body) : null
if (!payload) {
throw new Error('Output is unavailable')
}
finish({ full: {
text: payload.head,
partial: payload.truncated ? {
shown: new TextEncoder().encode(payload.head).length, total: payload.byteLength
} : null
} })
} catch {
setFailed(true)
} finally {
setLoading(false)
finish({ failed: true })
}
}
@@ -85,7 +89,7 @@ export function NativeChatToolOutput({
body.isError ? 'text-destructive' : 'text-chat-foreground'
)}
>
{full ? full.text : truncateToolDetail(body.output)}
{full ? full.text : expanded ? body.output : truncateToolDetail(body.output)}
</pre>
{full?.partial ? (
<p className="text-xs text-chat-foreground-faint">
@@ -209,7 +209,7 @@ export function projectStructuredItemToNativeChat(
...lifecycle,
...(command ? { command } : {}),
...(from && projected.role === 'user' ? { from } : {}),
...(item.shortened ? { shortened: true } : {})
...(item.shortened && item.body.kind === 'message' ? { shortened: true } : {})
}
: null
projectedItems.set(item, message)