mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
This commit is contained in:
@@ -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<string, {
|
||||
input?: unknown; output?: string; inputBytes: number; outputBytes: number
|
||||
}>({
|
||||
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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -27,7 +27,10 @@ export type JsonlRpcTimelineLaneDeps = {
|
||||
export class JsonlRpcTimelineLane {
|
||||
private readonly assembler: ProviderTimelineAssembler
|
||||
private readonly identity: ReturnType<typeof createLegacyProviderTimelineIdentityScheme>
|
||||
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)) {
|
||||
|
||||
@@ -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<void>()
|
||||
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<void>()
|
||||
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) })))
|
||||
})
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
)
|
||||
|
||||
@@ -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,
|
||||
|
||||
+23
-10
@@ -22,9 +22,15 @@ export class PiRpcMessages {
|
||||
private readonly text = new Map<string, { index: number; channel: 'assistant' | 'reasoning' }>()
|
||||
private readonly tools = new Map<string, AgentJournalToolCallItem>()
|
||||
// Closing events take these latest full fields; running journal rows retain only previews.
|
||||
private readonly whole = new BoundedMap<string, {
|
||||
input?: unknown; output?: string; inputBytes: number; outputBytes: number
|
||||
}>({
|
||||
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 (
|
||||
|
||||
@@ -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<typeof setTimeout> | undefined
|
||||
try {
|
||||
|
||||
@@ -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(<NativeChatToolOutput body={block} />))
|
||||
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(<NativeChatToolOutput body={block} />))
|
||||
const click = async () => {
|
||||
await act(async () => container.querySelector('button')?.click())
|
||||
}
|
||||
render(body)
|
||||
mocks.read.mockResolvedValueOnce(response('OLD TAIL'))
|
||||
await click()
|
||||
|
||||
@@ -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<void> => {
|
||||
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 })
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user