From 2bc32b7dbef303691102ab90ed67743e8d17f85c Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Tue, 20 Oct 2026 14:44:28 -0400 Subject: [PATCH] style: format the files #29982 merged unformatted (fix red main) (#31184) --- src/main/acp/acp-tool-timeline.ts | 33 ++- ...journal-translation-turn-lifecycle.test.ts | 8 +- .../jsonl-rpc/timeline-final-tail.test.ts | 33 ++- src/main/jsonl-rpc/timeline-lane.ts | 10 +- .../journal-completed-content.test.ts | 253 +++++++++++++----- .../agent-session-item-wire-form.test.ts | 51 +++- .../agent-session-item-wire-form.ts | 6 +- src/main/pi/rpc-messages.ts | 33 ++- src/main/pi/rpc-session.ts | 3 +- .../native-chat/NativeChatToolOutput.test.tsx | 42 ++- .../native-chat/NativeChatToolOutput.tsx | 36 ++- 11 files changed, 376 insertions(+), 132 deletions(-) diff --git a/src/main/acp/acp-tool-timeline.ts b/src/main/acp/acp-tool-timeline.ts index 7134d4caaee..19ea2b59438 100644 --- a/src/main/acp/acp-tool-timeline.ts +++ b/src/main/acp/acp-tool-timeline.ts @@ -91,9 +91,15 @@ export class AcpToolTimeline { }) // 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({ + 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 }) @@ -126,13 +132,20 @@ export class AcpToolTimeline { 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) ?? '') - })) { + 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 { 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 784e1b148ab..7ec2488a194 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 @@ -483,7 +483,13 @@ describe('codex turn lifecycle rows', () => { }) expect( - retries.handle(SESSION_ID, 'turn/started', { turn: { id: TURN_ID }, output: 'x'.repeat(9 * 1024 * 1024) }, 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/jsonl-rpc/timeline-final-tail.test.ts b/src/main/jsonl-rpc/timeline-final-tail.test.ts index b5688ab41e4..aeee59e581c 100644 --- a/src/main/jsonl-rpc/timeline-final-tail.test.ts +++ b/src/main/jsonl-rpc/timeline-final-tail.test.ts @@ -92,7 +92,13 @@ describe('Pi final tail before host eviction', () => { } ]) 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' } }]) + 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 @@ -114,7 +120,11 @@ 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.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({ @@ -140,10 +150,21 @@ describe('Pi final tail before host eviction', () => { }) cleanups.push(() => deferred.close()) 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( + 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({ diff --git a/src/main/jsonl-rpc/timeline-lane.ts b/src/main/jsonl-rpc/timeline-lane.ts index ffae11df83b..32f6d54b6d3 100644 --- a/src/main/jsonl-rpc/timeline-lane.ts +++ b/src/main/jsonl-rpc/timeline-lane.ts @@ -27,7 +27,10 @@ export type JsonlRpcTimelineLaneDeps = { export class JsonlRpcTimelineLane { private readonly assembler: ProviderTimelineAssembler private readonly identity: ReturnType - private readonly held: { events: { event: ProviderTimelineEvent; bytes: number }[]; bytes: number }[] = [] + private readonly held: { + events: { event: ProviderTimelineEvent; bytes: number }[] + bytes: number + }[] = [] private heldEvents = 0 private sealed = false private heldBytes = 0 @@ -71,7 +74,10 @@ export class JsonlRpcTimelineLane { if (this.ended || this.sealed) { return } - const pending = events.map((event) => ({ event, bytes: Buffer.byteLength(JSON.stringify(event)) })) + 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)) { 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 index 814928051b9..b08cc3e0234 100644 --- 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 @@ -4,86 +4,191 @@ 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 { + 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 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' as const, output: journalPayload(text) } - const cursor = rig.journal.cursor() - const snapshot = rig.journal.snapshot() - const blocked = rig.journal.readInOrder(() => { - expect(rig.eventSink.tryAppendItem?.({ ...identity, recordId: 'before' }, { kind: 'status', tone: 'info', text: 'before' }, options)).toEqual({ accepted: true }) - return gate.promise - }) - expect(rig.journal.cursor()).toEqual(cursor) - expect(rig.journal.snapshot()).toEqual(snapshot) - 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: true }) - let landed = false - const written = rig.eventSink.written?.().then((result) => { landed = true; return result }) - await new Promise((resolve) => setImmediate(resolve)) - expect(rig.journal.cursor()).toEqual(cursor) - expect(rig.journal.snapshot()).toEqual(snapshot) - expect(landed).toBe(false) - gate.resolve() - await blocked - expect(await written).toEqual({ ok: true }) - const reopened = await rig.reopenJournal() - expect(reopened.snapshot().items.map((item) => item.body)).toEqual([ - { kind: 'status', tone: 'info', text: 'before' }, body - ]) +afterEach(async () => { + cleanups.splice(0).forEach((cleanup) => cleanup()) + await closeProviderTimelineRigs() }) -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() - const update = agent === 'acp' ? acp.translate({ toolCallId: id, status: 'in_progress' }, {}, {}) - : pi.tool({ type: 'tool_execution_update', toolCallId: id, toolName: 'bash' }) - const changed = update.find((event) => event.type === 'item.update') - const again = agent === 'acp' ? acp.translate({ toolCallId: id, status: 'in_progress' }, {}, {}) - : pi.tool({ type: 'tool_execution_update', toolCallId: id, toolName: 'bash' }) - const repeated = again.find((event) => event.type === 'item.update') - if (changed?.type !== 'item.update' || repeated?.type !== 'item.update' || - changed.body.kind !== 'tool-call' || repeated.body.kind !== 'tool-call') { - throw new Error('Expected running tool updates') +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 identity = { + provider: 'legacy' as const, + agent: 'grok', + sessionId: SESSION, + recordId: 'full' } - expect(repeated.body.input).toBe(changed.body.input) - expect(repeated.body.output).toBe(changed.body.output) + 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' as const, + output: journalPayload(text) + } + const cursor = rig.journal.cursor() + const snapshot = rig.journal.snapshot() + const blocked = rig.journal.readInOrder(() => { + expect( + rig.eventSink.tryAppendItem?.( + { ...identity, recordId: 'before' }, + { kind: 'status', tone: 'info', text: 'before' }, + options + ) + ).toEqual({ accepted: true }) + return gate.promise + }) + expect(rig.journal.cursor()).toEqual(cursor) + expect(rig.journal.snapshot()).toEqual(snapshot) + 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: true }) + let landed = false + const written = rig.eventSink.written?.().then((result) => { + landed = true + return result + }) + await new Promise((resolve) => setImmediate(resolve)) + expect(rig.journal.cursor()).toEqual(cursor) + expect(rig.journal.snapshot()).toEqual(snapshot) + expect(landed).toBe(false) + gate.resolve() + await blocked + expect(await written).toEqual({ ok: true }) + const reopened = await rig.reopenJournal() + expect(reopened.snapshot().items.map((item) => item.body)).toEqual([ + { kind: 'status', tone: 'info', text: 'before' }, + body + ]) } - 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() +) + +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() + const update = + agent === 'acp' + ? acp.translate({ toolCallId: id, status: 'in_progress' }, {}, {}) + : pi.tool({ type: 'tool_execution_update', toolCallId: id, toolName: 'bash' }) + const changed = update.find((event) => event.type === 'item.update') + const again = + agent === 'acp' + ? acp.translate({ toolCallId: id, status: 'in_progress' }, {}, {}) + : pi.tool({ type: 'tool_execution_update', toolCallId: id, toolName: 'bash' }) + const repeated = again.find((event) => event.type === 'item.update') + if ( + changed?.type !== 'item.update' || + repeated?.type !== 'item.update' || + changed.body.kind !== 'tool-call' || + repeated.body.kind !== 'tool-call' + ) { + throw new Error('Expected running tool updates') + } + expect(repeated.body.input).toBe(changed.body.input) + expect(repeated.body.output).toBe(changed.body.output) + } + 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) + }) + ) + ) } - 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-wire/agent-session-item-wire-form.test.ts b/src/main/native-chat/agent-session-wire/agent-session-item-wire-form.test.ts index 068c28220c2..a1fa4a7350b 100644 --- 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 @@ -5,17 +5,50 @@ import { HISTORY_PAGE_CONTENT_BUDGET_BYTES } from './agent-session-history-page- 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` }] } } + 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` }] + } + } const formed = agentJournalItemForWire(item) expect(formed.item).toBe(item) expect(formed.bytes).toBe(Buffer.byteLength(JSON.stringify(item), 'utf8')) }) -it.each([73_728, HISTORY_PAGE_CONTENT_BUDGET_BYTES, 4 * 1024 * 1024])('keeps tool identity and output when cardinality exceeds %i bytes', (budget) => { - 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) -}) +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 4d99b701c3d..25caed36eed 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 @@ -10,7 +10,11 @@ import type { AgentJournalItemBody, AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' -import { boundToolInput, JOURNAL_RUNNING_PREVIEW_LIMITS, 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 HistoryItemForm, diff --git a/src/main/pi/rpc-messages.ts b/src/main/pi/rpc-messages.ts index bb4acc983d9..4a66b7b72c4 100644 --- a/src/main/pi/rpc-messages.ts +++ b/src/main/pi/rpc-messages.ts @@ -22,9 +22,15 @@ export class PiRpcMessages { private readonly text = new Map() private readonly tools = new Map() // Closing events take these latest full fields; running journal rows retain only previews. - private readonly whole = new BoundedMap({ + 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 }) @@ -161,13 +167,20 @@ export class PiRpcMessages { 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) ?? '') - })) { + 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 ( diff --git a/src/main/pi/rpc-session.ts b/src/main/pi/rpc-session.ts index 2d82202378e..ddab7c61986 100644 --- a/src/main/pi/rpc-session.ts +++ b/src/main/pi/rpc-session.ts @@ -246,7 +246,8 @@ export class PiRpcSession { this.turns.end() // The lane is sealed once finalized, so the end event travels with the final transfer. this.lane.finalize({ - type: 'session.ended', verdict: { state: 'interrupted', completedAt: Date.now() } + type: 'session.ended', + verdict: { state: 'interrupted', completedAt: Date.now() } }) let timer: ReturnType | undefined try { diff --git a/src/renderer/src/components/native-chat/NativeChatToolOutput.test.tsx b/src/renderer/src/components/native-chat/NativeChatToolOutput.test.tsx index 045a7dbcab2..79125b21683 100644 --- a/src/renderer/src/components/native-chat/NativeChatToolOutput.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatToolOutput.test.tsx @@ -6,18 +6,46 @@ import type { AgentJournalRenderItem } from '../../../../shared/agent-session-jo 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' } })) +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 })) +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() }) +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()) } + 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() diff --git a/src/renderer/src/components/native-chat/NativeChatToolOutput.tsx b/src/renderer/src/components/native-chat/NativeChatToolOutput.tsx index 05398b5d30d..1ba778d8e85 100644 --- a/src/renderer/src/components/native-chat/NativeChatToolOutput.tsx +++ b/src/renderer/src/components/native-chat/NativeChatToolOutput.tsx @@ -36,17 +36,24 @@ export function NativeChatToolOutput({ const owner = useNativeChatVisualOwner() const [expanded, setExpanded] = useState(false) const [detail, setDetail] = useState<{ - body: NativeChatToolResultBlock; owner: typeof owner; full?: FullOutput; - loading?: boolean; failed?: boolean + 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]) + 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 && - (body.fullOutputItemId !== undefined || (!expanded && body.output.length > MAX_TOOL_DETAIL_LENGTH)) + (body.fullOutputItemId !== undefined || + (!expanded && body.output.length > MAX_TOOL_DETAIL_LENGTH)) const showFull = async (): Promise => { const itemId = body.fullOutputItemId @@ -56,7 +63,9 @@ export function NativeChatToolOutput({ } 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) + setDetail((held) => + held?.body === body && held.owner === owner ? { body, owner, ...result } : held + ) } try { if (!(await supportsStructuredAgentSessionItemDetail(owner.target))) { @@ -69,12 +78,17 @@ export function NativeChatToolOutput({ 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 - } }) + finish({ + full: { + text: payload.head, + partial: payload.truncated + ? { + shown: new TextEncoder().encode(payload.head).length, + total: payload.byteLength + } + : null + } + }) } catch { finish({ failed: true }) }