diff --git a/mobile/src/session/MobileNativeChatFullContent.tsx b/mobile/src/session/MobileNativeChatFullContent.tsx index 22a5f366bf3..b93323859da 100644 --- a/mobile/src/session/MobileNativeChatFullContent.tsx +++ b/mobile/src/session/MobileNativeChatFullContent.tsx @@ -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(null) - const [load, setLoad] = useState('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 ( - {full ?? truncateToolDetail(output)} + {full ? full.head : expanded ? output : truncateToolDetail(output)} - {clipped ? : null} + {full?.truncated ? Showing {new TextEncoder().encode(full.head).length} of {full.byteLength} bytes. : null} + {clipped ? : null} ) } @@ -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('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: state === 'partial' ? Showing part of this message. + : state === 'available' ? : null } } diff --git a/mobile/src/session/MobileNativeChatOverlay.tsx b/mobile/src/session/MobileNativeChatOverlay.tsx index 0c627e58f37..5838df57dbe 100644 --- a/mobile/src/session/MobileNativeChatOverlay.tsx +++ b/mobile/src/session/MobileNativeChatOverlay.tsx @@ -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({ - + ) } diff --git a/mobile/src/session/mobile-native-chat-item-read.ts b/mobile/src/session/mobile-native-chat-item-read.ts index 0ffd7afc29b..bd3b02164de 100644 --- a/mobile/src/session/mobile-native-chat-item-read.ts +++ b/mobile/src/session/mobile-native-chat-item-read.ts @@ -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['item']>['body'] +const itemReplySchema = z.object({ item: z.custom(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( +export const MobileNativeChatItemSourceContext = createContext<(MobileNativeChatVisualSource & { messageDetail?: MobileNativeChatMessageDetail | null }) | null>( null ) @@ -49,7 +34,7 @@ export const MobileNativeChatItemSourceContext = createContext { +): Promise { 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 } diff --git a/mobile/src/session/use-mobile-native-chat-message-detail.test.tsx b/mobile/src/session/use-mobile-native-chat-message-detail.test.tsx new file mode 100644 index 00000000000..7f979981ca4 --- /dev/null +++ b/mobile/src/session/use-mobile-native-chat-message-detail.test.tsx @@ -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 | 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', '/plugin:examplehead…'), 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', '/plugin:examplefull', 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() + 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') +}) diff --git a/mobile/src/session/use-mobile-native-chat-message-detail.ts b/mobile/src/session/use-mobile-native-chat-message-detail.ts new file mode 100644 index 00000000000..bea8722342b --- /dev/null +++ b/mobile/src/session/use-mobile-native-chat-message-detail.ts @@ -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 +} + +/** 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(), + entries: new Map() + }), [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() + } + } + } + } +} diff --git a/src/main/acp/acp-tool-timeline.ts b/src/main/acp/acp-tool-timeline.ts index 9e561949ede..d75e0c6838d 100644 --- a/src/main/acp/acp-tool-timeline.ts +++ b/src/main/acp/acp-tool-timeline.ts @@ -89,12 +89,19 @@ export class AcpToolTimeline { private readonly turns = new BoundedMap({ 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({ - 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({ + 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) } diff --git a/src/main/codex/codex-structured-journal-translation-turn-lifecycle.test.ts b/src/main/codex/codex-structured-journal-translation-turn-lifecycle.test.ts index 0f51961b64d..784e1b148ab 100644 --- a/src/main/codex/codex-structured-journal-translation-turn-lifecycle.test.ts +++ b/src/main/codex/codex-structured-journal-translation-turn-lifecycle.test.ts @@ -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) diff --git a/src/main/codex/codex-structured-notification-retry.ts b/src/main/codex/codex-structured-notification-retry.ts index 15af01da375..2e64fc805a7 100644 --- a/src/main/codex/codex-structured-notification-retry.ts +++ b/src/main/codex/codex-structured-notification-retry.ts @@ -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') diff --git a/src/main/jsonl-rpc/timeline-final-tail.test.ts b/src/main/jsonl-rpc/timeline-final-tail.test.ts index f39f9e78991..ee63969e038 100644 --- a/src/main/jsonl-rpc/timeline-final-tail.test.ts +++ b/src/main/jsonl-rpc/timeline-final-tail.test.ts @@ -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' diff --git a/src/main/jsonl-rpc/timeline-lane.ts b/src/main/jsonl-rpc/timeline-lane.ts index 36c759e42de..ffae11df83b 100644 --- a/src/main/jsonl-rpc/timeline-lane.ts +++ b/src/main/jsonl-rpc/timeline-lane.ts @@ -27,7 +27,9 @@ export type JsonlRpcTimelineLaneDeps = { export class JsonlRpcTimelineLane { private readonly assembler: ProviderTimelineAssembler private readonly identity: ReturnType - 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 { @@ -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): 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() } diff --git a/src/main/native-chat/agent-session-journal/journal-body-ceiling.test.ts b/src/main/native-chat/agent-session-journal/journal-body-ceiling.test.ts deleted file mode 100644 index d19078aa64e..00000000000 --- a/src/main/native-chat/agent-session-journal/journal-body-ceiling.test.ts +++ /dev/null @@ -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) - }) -}) diff --git a/src/main/native-chat/agent-session-journal/journal-body-ceiling.ts b/src/main/native-chat/agent-session-journal/journal-body-ceiling.ts deleted file mode 100644 index 43a3619e9b8..00000000000 --- a/src/main/native-chat/agent-session-journal/journal-body-ceiling.ts +++ /dev/null @@ -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(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 -} diff --git a/src/main/native-chat/agent-session-journal/journal-completed-content.test.ts b/src/main/native-chat/agent-session-journal/journal-completed-content.test.ts new file mode 100644 index 00000000000..7bc41093d7b --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-completed-content.test.ts @@ -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() + 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) }))) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts index e2b9fb4d2ce..e6ee350cefe 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts @@ -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 { - 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() + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts index 48955892902..01e1d246628 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts @@ -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 { 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 { 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 })) } } diff --git a/src/main/native-chat/agent-session-journal/journal-payload-bounds.ts b/src/main/native-chat/agent-session-journal/journal-payload-bounds.ts index e579a429d58..bb18251398a 100644 --- a/src/main/native-chat/agent-session-journal/journal-payload-bounds.ts +++ b/src/main/native-chat/agent-session-journal/journal-payload-bounds.ts @@ -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 diff --git a/src/main/native-chat/agent-session-wire/agent-session-item-wire-form.test.ts b/src/main/native-chat/agent-session-wire/agent-session-item-wire-form.test.ts new file mode 100644 index 00000000000..1f5ceb35692 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/agent-session-item-wire-form.test.ts @@ -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) +}) diff --git a/src/main/native-chat/agent-session-wire/agent-session-item-wire-form.ts b/src/main/native-chat/agent-session-wire/agent-session-item-wire-form.ts index d909bcbbb73..8ba2b25d0f3 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-item-wire-form.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-item-wire-form.ts @@ -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 } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts index 3d5d9847400..2adf4328944 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink-queue.ts @@ -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) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts index e2ed3d46f3b..1f48f068058 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-event-sink.ts @@ -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 = ( + 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) => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-final-tail-reservation.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-final-tail-reservation.ts index 92b8e0f897f..c64b16ce91d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-final-tail-reservation.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-final-tail-reservation.ts @@ -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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-item-append.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-item-append.ts deleted file mode 100644 index d00f080bd64..00000000000 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-item-append.ts +++ /dev/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 - 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 } - ) - } - } -} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.ts index 6859646b6eb..d3640db2a0c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.ts @@ -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 + )) satisfies NonNullable return { tryAppendResolvedItem: identityOnly(false), tryAppendResolvedItemAndPublish: identityOnly(true), diff --git a/src/main/pi/rpc-messages.ts b/src/main/pi/rpc-messages.ts index 0d6d49b0711..c1ebfb5695b 100644 --- a/src/main/pi/rpc-messages.ts +++ b/src/main/pi/rpc-messages.ts @@ -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() private readonly tools = new Map() - /** 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() + // Closing events take these latest full fields; running journal rows retain only previews. + private readonly whole = new BoundedMap({ + 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)), diff --git a/src/main/pi/rpc-session.ts b/src/main/pi/rpc-session.ts index ae693653df4..c7cfea87065 100644 --- a/src/main/pi/rpc-session.ts +++ b/src/main/pi/rpc-session.ts @@ -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 | undefined try { await Promise.race([ diff --git a/src/renderer/src/components/native-chat/NativeChatToolOutput.test.tsx b/src/renderer/src/components/native-chat/NativeChatToolOutput.test.tsx new file mode 100644 index 00000000000..045a7dbcab2 --- /dev/null +++ b/src/renderer/src/components/native-chat/NativeChatToolOutput.test.tsx @@ -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()) + 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() + 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') +}) diff --git a/src/renderer/src/components/native-chat/NativeChatToolOutput.tsx b/src/renderer/src/components/native-chat/NativeChatToolOutput.tsx index b641caf0d78..05398b5d30d 100644 --- a/src/renderer/src/components/native-chat/NativeChatToolOutput.tsx +++ b/src/renderer/src/components/native-chat/NativeChatToolOutput.tsx @@ -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(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 => { 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)} {full?.partial ? (

diff --git a/src/shared/structured-agent-session-projection.ts b/src/shared/structured-agent-session-projection.ts index 6477afefae3..6a2ce2d08da 100644 --- a/src/shared/structured-agent-session-projection.ts +++ b/src/shared/structured-agent-session-projection.ts @@ -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)