diff --git a/config/scripts/vitest-node-runtime-files.mjs b/config/scripts/vitest-node-runtime-files.mjs index 3580506f049..48b32e5010a 100644 --- a/config/scripts/vitest-node-runtime-files.mjs +++ b/config/scripts/vitest-node-runtime-files.mjs @@ -61,6 +61,8 @@ const NODE_RUNTIME_INCLUDE = [ 'src/main/acp/acp-timeline-turn-failures.test.ts', 'src/main/acp/acp-structured-session-adapter.test.ts', 'src/main/acp/acp-structured-session-adapter-resume.test.ts', + // Counts real V8 Inspector coverage alongside delegated operation counters. + 'src/main/acp/acp-history-scaling.test.ts', 'src/main/acp/acp-structured-host-start-aborts.test.ts', 'src/main/acp/acp-conversation-copy.test.ts', 'src/main/pi/rpc-turn-races.test.ts', diff --git a/src/main/acp/acp-history-catch-up.test.ts b/src/main/acp/acp-history-catch-up.test.ts new file mode 100644 index 00000000000..8baa13a1367 --- /dev/null +++ b/src/main/acp/acp-history-catch-up.test.ts @@ -0,0 +1,801 @@ +// A reopened chat catches up on turns its agent replays that the chat never saw, against real +// OpenCode 2.0.14 and Grok 1.0.62 recordings: once, in order, under one divider, and never a turn +// the chat already holds. + +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + agentJournalSubmissionKey, + parseAgentJournalItemKey +} from '../../shared/agent-session-journal-item-key' +import type { AgentJournalRenderItem } from '../../shared/agent-session-journal-types' +import { agentSessionFailureFact } from '../../shared/agent-session-failure' +import { agentSessionFailureWords } from '../../shared/agent-session-failure-words' +import { + AGENT, + closeProviderTimelineRigs, + NAMESPACE, + SESSION, + providerItemId +} from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support' +import type { AcpDialect } from './acp-dialects/acp-dialect' +import { GROK_ACP_DIALECT } from './acp-dialects/grok-dialect' +import { OPENCODE_ACP_DIALECT } from './acp-dialects/opencode-dialect' +import { agentJournalTurnBody, readAgentJournalTurn } from '../../shared/agent-session-turn-record' +import { acpNamedTextKey } from './acp-turn-messages' +import { isHistoricalProviderTimelineItem } from '../native-chat/agent-session-timeline/provider-timeline-history' +import { StructuredAgentSessionCurrentWork } from '../native-chat/agent-session-wire/structured-agent-session-current-work' +import { hasUnfinishedStructuredAgentSessionWork } from '../native-chat/agent-session-wire/structured-agent-session-unfinished-work' +import { planStructuredAgentSessionDeadGeneration } from '../native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement' +import { catchUpAcpHistory } from './acp-history-catch-up' +import { + AcpLoadHistory, + type AcpLoadHistoryFrame, + type AcpHistoryTurnIdentity +} from './acp-history-replay' +import { asReattachHistory } from './acp-structured-session' +import { + openAcpFixtureRig, + readAcpFixture, + type AcpFixtureFrame +} from './acp-timeline-fixture.test-support' + +afterEach(closeProviderTimelineRigs) + +async function recording(name: string): Promise { + // The rig's journal keys rows under its own provider session. + return JSON.parse( + JSON.stringify(await readAcpFixture(name)).replaceAll('"session-1"', JSON.stringify(NAMESPACE)) + ) +} + +/** What the chat's lane kept while process `process` replayed its history on load. */ +function loadReplay(frames: AcpFixtureFrame[], process: string) { + const own = frames.filter((frame) => frame.process === process) + const load = own.findIndex((frame) => frame.message.method === 'session/load') + const answered = own.findIndex((frame, index) => index > load && frame.message.id !== undefined) + return own.slice(load + 1, answered).map((frame) => ({ + method: frame.message.method!, + params: asReattachHistory(frame.message.params) + })) +} + +async function setup(dialect: AcpDialect, identity: AcpHistoryTurnIdentity) { + const fixture = await openAcpFixtureRig(dialect) + const warn = vi.fn() + const catchUp = (frames: ReturnType) => + catchUpAcpHistory({ + frames, + identity, + sink: fixture.rig.eventSink, + sessionId: SESSION, + agent: AGENT, + providerSessionId: NAMESPACE, + dialect, + logger: { warn, error: vi.fn() }, + assertCurrent: () => {} + }) + return { fixture, catchUp, warn } +} + +function userTexts(rows: AgentJournalRenderItem[]): string[] { + return rows.flatMap((row) => + row.body.kind === 'message' && row.body.role === 'user' && row.body.blocks[0]?.type === 'text' + ? [row.body.blocks[0].text] + : [] + ) +} + +function dividers(rows: AgentJournalRenderItem[]): AgentJournalRenderItem[] { + return rows.filter( + (row) => row.body.kind === 'status' && row.body.presentation?.startsWith('continued-outside') + ) +} + +/** The chat's own send, as its row reads: the text a person typed. */ +async function chatSend( + fixture: Awaited>, + clientMessageId: string, + text: string +) { + const identity = parseAgentJournalItemKey(agentJournalSubmissionKey(clientMessageId))! + await fixture.rig.journal.appendItem( + identity, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }, + { fence: 1, turnScope: { kind: 'thread' } } + ) +} + +function replayUpdate( + sessionUpdate: string, + rest: Record = {} +): AcpLoadHistoryFrame { + return { + method: 'session/update', + params: { sessionId: NAMESPACE, update: { sessionUpdate, ...rest } } + } +} +const userReplay = (id: string, text = 'repeat') => + replayUpdate('user_message_chunk', { + messageId: id, + content: { type: 'text', text } + }) +const answerReplay = (text: string, id = 'answer') => + replayUpdate('agent_message_chunk', { + messageId: id, + content: { type: 'text', text } + }) + +describe('ACP history catch-up', () => { + it('adds an OpenCode turn made outside the chat once, completing it on a later open', async () => { + const { fixture, catchUp } = await setup(OPENCODE_ACP_DIALECT, 'user-message-id') + const frames = await recording('opencode-v2-reopen') + const chat = frames.filter((frame) => frame.process === 'A') + const prompts = chat.filter((frame) => frame.message.method === 'session/prompt') + // The chat saw its first two turns; the third ran outside it. + await fixture.feed(chat.slice(0, chat.indexOf(prompts[2]!))) + const before = await fixture.rig.rows() + await fixture.restart() + const replay = loadReplay(frames, 'B') + const outsideTool = replay.findLastIndex((frame) => + JSON.stringify(frame.params).includes('"tool_call"') + ) + + // Opened while the outside turn had only begun: it shows as far as it went. + await catchUp(replay.slice(0, outsideTool)) + const partial = await fixture.rig.rows() + expect(partial.slice(0, before.length).map((row) => row.itemId)).toEqual( + before.map((row) => row.itemId) + ) + const added = partial.slice(before.length) + expect(added[0]?.body).toMatchObject({ kind: 'status', presentation: 'continued-outside' }) + expect(userTexts(added)).toEqual([expect.stringContaining('sleep 40')]) + + await catchUp(replay) + const complete = await fixture.rig.rows() + expect(dividers(complete)).toHaveLength(1) + expect(userTexts(complete)).toEqual([expect.stringContaining('sleep 40')]) + expect(complete.filter((row) => row.body.kind === 'tool-call')).toHaveLength(3) + + await catchUp(replay) + expect(await fixture.rig.rows()).toEqual(complete) + }) + + it('retains unproven visible replay after an OpenCode native failure', async () => { + const { fixture, catchUp } = await setup(OPENCODE_ACP_DIALECT, 'user-message-id') + const frames = await recording('opencode-v2-reopen-outside') + const prompt = frames.find((frame) => frame.message.method === 'session/prompt') + await fixture.feed([prompt!]) + await chatSend(fixture, 'A:3', 'Reply with exactly the word ONE.') + fixture.apply(fixture.lane().promptFailed('A:3', new Error('quota'), 1_100)) + await fixture.restart() + + await catchUp(loadReplay(frames, 'C')) + const rows = await fixture.rig.rows() + expect(userTexts(rows)).toEqual([ + 'Reply with exactly the word ONE.', + 'Reply with exactly the word ONE.', + expect.stringContaining('TWO-TOOL') + ]) + expect(dividers(rows)).toHaveLength(1) + }) + + it('never repeats turns an OpenCode 2.x chat was adopted from', async () => { + const { fixture, catchUp } = await setup(OPENCODE_ACP_DIALECT, 'user-message-id') + const frames = await recording('opencode-v2-reopen-outside') + const replay = loadReplay(frames, 'C') + const adoptedId = JSON.stringify(replay[0]!.params).match(/msg_\w+/)![0] + await fixture.rig.journal.appendItem( + { provider: 'legacy', agent: AGENT, sessionId: NAMESPACE, recordId: `opencode:${adoptedId}` }, + { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'Reply with exactly the word ONE.' }] + }, + { fence: 0, turnScope: { kind: 'thread' } } + ) + + await catchUp(replay) + expect(userTexts(await fixture.rig.rows())).toEqual([ + 'Reply with exactly the word ONE.', + expect.stringContaining('TWO-TOOL') + ]) + }) + + it('adds a Grok turn made outside the chat under its own id, never the turn Orca sent', async () => { + const { fixture, catchUp } = await setup(GROK_ACP_DIALECT, 'prompt-id') + const frames = await recording('grok-v1-reopen-outside') + await fixture.feed(frames.filter((frame) => frame.process === 'A')) + await chatSend(fixture, 'A:3', 'Reply with exactly the word ONE.') + await fixture.restart() + + await catchUp(loadReplay(frames, 'C')) + const rows = await fixture.rig.rows() + expect(userTexts(rows)).toEqual([ + 'Reply with exactly the word ONE.', + expect.stringContaining('TWO-TOOL') + ]) + expect(dividers(rows)).toHaveLength(1) + expect((await fixture.rig.turns()).map((turn) => turn.state)).toEqual([ + 'completed', + 'completed' + ]) + }) + it('refreshes same-key bodies after restart without inventing execution ownership or a new divider', async () => { + const { fixture, catchUp } = await setup(OPENCODE_ACP_DIALECT, 'user-message-id') + const { rig } = fixture + await rig.journal.appendSubmission({ + clientMessageId: 'native', + payloadFingerprint: 'native', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'my native prompt' }] }, + fence: 1 + }) + await rig.journal.resolveDispatch({ + clientMessageId: 'native', + fence: 1, + state: 'accepted', + providerIdentity: null + }) + rig.assembler.apply({ + type: 'input.accepted', + clientMessageId: 'native', + requestedAt: 1000, + join: { turn: 'native-turn' } + }) + rig.assembler.apply({ type: 'turn.open', turn: 'native-turn', at: 1000 }) + rig.assembler.apply({ type: 'turn.end', turn: 'native-turn', state: 'completed', at: 1100 }) + const nativeBefore = await rig.rows() + const source = 'external-'.repeat(40) + const tool = replayUpdate('tool_call', { + toolCallId: 'tool', + name: 'shell', + title: 'shell', + kind: 'execute', + status: 'in_progress', + rawInput: { command: 'sleep 40' } + }) + const partial = [userReplay(source), answerReplay('prefix'), tool] + await catchUp(partial) + const imported = await rig.rows() + const historical = imported.filter((item) => + isHistoricalProviderTimelineItem(item, (key) => rig.journal.item(key)) + ) + expect(historical).toHaveLength(4) + const container = historical.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(container).toMatchObject({ state: 'completed' }) + expect(container).not.toHaveProperty('outcome') + expect(container).not.toHaveProperty('completedAt') + expect(container).not.toHaveProperty('startedAt') + const observedTool = historical.find((item) => item.body.kind === 'tool-call')! + expect(observedTool.body).toMatchObject({ state: 'running' }) + expect( + new StructuredAgentSessionCurrentWork(rig.journal, 1).isCurrentItem(observedTool.itemId) + ).toBe(false) + expect(hasUnfinishedStructuredAgentSessionWork(rig.journal)).toBe(false) + expect( + planStructuredAgentSessionDeadGeneration({ + journal: rig.journal, + sessionId: SESSION, + fence: 1, + settlementId: 'exit', + verdict: { state: 'interrupted' }, + pendingSubmissionReason: 'ended' + }).mutations + ).toEqual([]) + rig.assembler.apply({ type: 'session.ended', verdict: { state: 'interrupted' }, endedAt: 9000 }) + await fixture.restart() + expect((await rig.row(observedTool.itemId))?.body).toEqual(observedTool.body) + const completed = replayUpdate('tool_call_update', { + toolCallId: 'tool', + status: 'completed', + rawOutput: 'done', + content: [{ type: 'content', content: { type: 'text', text: 'done' } }] + }) + const full = [ + userReplay(source), + answerReplay('prefix and full tail'), + tool, + completed, + answerReplay('second', 'answer-2') + ] + await catchUp(full) + const complete = await rig.rows() + expect(complete.slice(0, nativeBefore.length)).toEqual(nativeBefore) + const revised = complete.find((row) => row.itemId === observedTool.itemId)! + expect(revised.body).toMatchObject({ state: 'completed' }) + expect(revised.sequence).toBe(observedTool.sequence) + expect(revised.revision).toBe(observedTool.revision + 1) + expect( + complete + .filter((row) => row.body.kind === 'message' && row.body.role === 'assistant') + .map((row) => row.body) + ).toMatchObject([ + { blocks: [{ text: 'prefix and full tail' }] }, + { blocks: [{ text: 'second' }] } + ]) + expect(dividers(complete)).toHaveLength(1) + await catchUp(full) + expect(await rig.rows()).toEqual(complete) + const reopened = await rig.reopenJournal() + expect( + isHistoricalProviderTimelineItem(reopened.item(observedTool.itemId)!, (key) => + reopened.item(key) + ) + ).toBe(true) + expect(isHistoricalProviderTimelineItem(nativeBefore[0]!, (key) => reopened.item(key))).toBe( + false + ) + }) + + it('keeps same-text outside answers and reasoning while deduplicating exact native output', async () => { + const { fixture, catchUp } = await setup(OPENCODE_ACP_DIALECT, 'user-message-id') + const { rig } = fixture + await chatSend(fixture, 'native', 'repeat') + rig.assembler.apply({ + type: 'input.accepted', + clientMessageId: 'native', + requestedAt: 1000, + join: { turn: 'native-turn' } + }) + rig.assembler.apply({ type: 'turn.open', turn: 'native-turn', at: 1000 }) + rig.assembler.apply({ + type: 'text.delta', + item: { id: acpNamedTextKey('native-answer', 'assistant') }, + channel: 'assistant', + text: 'native answer', + join: { thread: NAMESPACE, turn: 'native-turn' } + }) + rig.assembler.flush() + rig.assembler.apply({ type: 'turn.end', turn: 'native-turn', state: 'completed', at: 1100 }) + const native = await rig.rows() + const reasoning = replayUpdate('agent_thought_chunk', { + messageId: 'thought:reasoning:0', + content: { type: 'text', text: 'outside reasoning' } + }) + await catchUp([ + userReplay('native-provider-id'), + answerReplay('native answer', 'native-answer'), + userReplay('outside'), + answerReplay('outside answer'), + userReplay('outside-thought'), + reasoning + ]) + const rows = await rig.rows() + expect(rows.slice(0, native.length)).toEqual(native) + expect(userTexts(rows)).toEqual(['repeat', 'repeat', 'repeat']) + expect( + rows.filter((row) => row.body.kind === 'message' && row.body.role === 'reasoning') + ).toHaveLength(1) + expect(dividers(rows)).toHaveLength(1) + expect( + rows.filter( + (row) => + row.itemId === + providerItemId('item', acpNamedTextKey('native-answer', 'assistant'), { + thread: NAMESPACE + }) + ) + ).toHaveLength(1) + await chatSend(fixture, 'other', 'other') + rig.assembler.apply({ + type: 'input.accepted', + clientMessageId: 'other', + requestedAt: 2000, + join: { turn: 'other' } + }) + rig.assembler.apply({ type: 'turn.open', turn: 'other', at: 2000 }) + rig.assembler.apply({ + type: 'text.delta', + item: { id: acpNamedTextKey('other-answer', 'assistant') }, + channel: 'assistant', + text: 'other answer', + join: { thread: NAMESPACE, turn: 'other' } + }) + rig.assembler.flush() + rig.assembler.apply({ type: 'turn.end', turn: 'other', state: 'completed', at: 2100 }) + const beforeConflict = await rig.rows() + await catchUp([ + userReplay('conflict'), + answerReplay('native answer', 'native-answer'), + answerReplay('other answer', 'other-answer') + ]) + expect(await rig.rows()).toEqual(beforeConflict) + }) + + it('uses Grok handover proof without claiming a never-handed-over queued send', async () => { + const { fixture, catchUp } = await setup(GROK_ACP_DIALECT, 'prompt-id') + const { journal } = fixture.rig + for (const id of ['handed', 'queued']) { + await journal.appendSubmission({ + clientMessageId: id, + payloadFingerprint: id, + fence: 1, + handoverRecorded: true, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'repeat' }] } + }) + } + await journal.resolveDispatch({ + clientMessageId: 'handed', + fence: 1, + state: 'pending', + turnScope: { kind: 'thread' } + }) + const submissions = journal.submissions() + const replay = ['handed', 'queued'].flatMap((id) => [ + userReplay(id), + { + method: 'session/update', + params: { + sessionId: NAMESPACE, + _meta: { promptId: `prompt:${id}` }, + update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: id } } + } + } + ]) + await catchUp(replay) + expect(journal.submissions()).toEqual(submissions) + expect(userTexts(await fixture.rig.rows())).toEqual(['repeat', 'repeat', 'repeat']) + expect(journal.receiptFor('queued')).toBeNull() + expect(journal.submission('queued')?.dispatchState).toBe('pending') + expect(journal.submission('queued')?.handedOverAt).toBeUndefined() + }) + it.each(['interleaved', 'adjacent', 'hidden-root', 'hidden-later'] as const)( + 'keeps Grok root and later messages in one execution across restart: %s', + async (scenario) => { + const { fixture, catchUp } = await setup(GROK_ACP_DIALECT, 'prompt-id') + const { rig } = fixture + const user = ( + eventId: string, + text: string, + meta: Record = {}, + displayText?: string + ) => ({ + method: 'session/update', + params: { + sessionId: NAMESPACE, + _meta: { eventId, isReplay: true }, + update: { + sessionUpdate: 'user_message_chunk', + _meta: meta, + content: { + type: 'text', + text, + ...(displayText === undefined ? {} : { _meta: { displayText } }) + } + } + } + }) + const output = (text: string) => ({ + method: 'session/update', + params: { + sessionId: NAMESPACE, + _meta: { promptId: 'outside', isReplay: true }, + update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text } } + } + }) + const root = user('root-event', 'root', { + promptIndex: 0, + ...(scenario === 'hidden-root' ? { hideFromScrollback: true } : {}) + }) + const first = user( + 'interjection-event', + 'model envelope', + { + interjection: true, + ...(scenario === 'hidden-later' ? { hideFromScrollback: true } : {}) + }, + 'typed followup' + ) + const image = { + ...first, + params: { + ...first.params, + update: { + ...first.params.update, + content: { type: 'image', mimeType: 'image/png', data: 'aW1hZ2U=' } + } + } + } + const second = user('permission-event', 'visible permission followup') + const rootBlocks = + scenario === 'interleaved' + ? [ + root, + { + ...image, + params: { + ...image.params, + _meta: { eventId: 'root-image-event', isReplay: true }, + update: { ...image.params.update, _meta: { promptIndex: 0 } } + } + } + ] + : [root] + if (scenario === 'interleaved') { + await catchUp([...rootBlocks, output('A')]) + } + const before = await rig.rows() + const submissions = rig.journal.submissions() + await fixture.restart() + const replay = [ + ...rootBlocks, + ...(scenario === 'adjacent' ? [] : [output('A')]), + first, + image, + ...(scenario === 'adjacent' ? [] : [output('B')]), + second, + output('C'), + { + method: '_x.ai/session/update', + params: { + sessionId: NAMESPACE, + update: { + sessionUpdate: 'turn_completed', + prompt_id: 'outside', + stop_reason: 'end_turn' + } + } + } + ] + await catchUp(replay) + const rows = await rig.rows() + const messages = rows.flatMap((row) => (row.body.kind === 'message' ? [row] : [])) + expect(messages.map((row) => row.body)).toMatchObject([ + { + role: 'user', + blocks: + scenario === 'hidden-root' + ? [] + : [ + { text: 'root' }, + ...(scenario === 'interleaved' + ? [{ type: 'image-ref', url: 'data:image/png;base64,aW1hZ2U=' }] + : []) + ] + }, + ...(scenario === 'adjacent' ? [] : [{ role: 'assistant', blocks: [{ text: 'A' }] }]), + { + role: 'user', + blocks: + scenario === 'hidden-later' + ? [] + : [ + { text: 'typed followup' }, + { type: 'image-ref', url: 'data:image/png;base64,aW1hZ2U=' } + ] + }, + ...(scenario === 'adjacent' ? [] : [{ role: 'assistant', blocks: [{ text: 'B' }] }]), + { role: 'user', blocks: [{ text: 'visible permission followup' }] }, + { role: 'assistant', blocks: [{ text: 'C' }] } + ]) + const lifecycle = rows.filter((row) => readAgentJournalTurn(row.body)) + expect(lifecycle).toHaveLength(1) + const opener = messages[0]! + expect(opener.itemId).toBe(providerItemId('item', 'user:outside', { thread: NAMESPACE })) + const laterUsers = messages + .filter((row) => row.body.kind === 'message' && row.body.role === 'user') + .slice(1) + expect(laterUsers.map((row) => row.itemId)).toEqual( + ['interjection-event', 'permission-event'].map((id) => + providerItemId('item', `user-message:${JSON.stringify(['outside', id])}`, { + thread: NAMESPACE + }) + ) + ) + expect(readAgentJournalTurn(lifecycle[0]!.body)?.userItemId).toBe(opener.itemId) + expect(opener.turnScope).toEqual({ kind: 'turn', turnItemId: lifecycle[0]!.itemId }) + expect( + messages.every((row) => + isHistoricalProviderTimelineItem(row, (key) => rig.journal.item(key)) + ) + ).toBe(true) + for (const held of before) { + const current = rows.find((row) => row.itemId === held.itemId)! + expect(current.sequence).toBe(held.sequence) + expect(current.turnScope).toEqual(held.turnScope) + if (readAgentJournalTurn(held.body)) { + expect(current.body).toMatchObject({ state: 'completed', outcome: 'success' }) + expect(current.revision).toBe(held.revision + 1) + } else { + expect(current).toEqual(held) + } + } + expect(dividers(rows)).toHaveLength(0) + expect(rig.journal.submissions()).toEqual(submissions) + await catchUp(replay) + expect(await rig.rows()).toEqual(rows) + } + ) + it('skips unsupported or contradictory Grok source atomically and permits the next load', async () => { + const { fixture, catchUp, warn } = await setup(GROK_ACP_DIALECT, 'prompt-id') + const root = replayUpdate('user_message_chunk', { + content: { type: 'text', text: 'root' }, + _meta: { promptIndex: 0 } + }) + const output = { + method: 'session/update', + params: { + sessionId: NAMESPACE, + _meta: { promptId: 'outside' }, + update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'A' } } + } + } + const end = replayUpdate('turn_completed', { prompt_id: 'outside', stop_reason: 'end_turn' }) + const later = userReplay('later', 'followup') + const snapshots = [ + [root, output, later], + [ + root, + output, + replayUpdate('user_message_chunk', { + content: { type: 'text', text: '/help' }, + _meta: { hostTurn: true } + }) + ], + [ + root, + output, + replayUpdate('turn_completed', { prompt_id: 'conflict', stop_reason: 'end_turn' }) + ], + [root, output, end, later] + ] + for (const frames of snapshots) { + await catchUp(frames) + expect(await fixture.rig.rows()).toEqual([]) + } + expect(warn).toHaveBeenCalledTimes(snapshots.length) + expect(warn.mock.calls.every(([, fields]) => fields.reason === 'import-skipped')).toBe(true) + await catchUp([root, output]) + expect(userTexts(await fixture.rig.rows())).toEqual(['root']) + }) + + it('coalesces OpenCode root text and attachments under the provider message id', async () => { + const { fixture, catchUp } = await setup(OPENCODE_ACP_DIALECT, 'user-message-id') + const frames = [ + userReplay('root', 'read this'), + replayUpdate('user_message_chunk', { + messageId: 'root', + content: { type: 'image', mimeType: 'image/png', data: 'aW1hZ2U=' } + }), + answerReplay('answer') + ] + await catchUp(frames) + const rows = await fixture.rig.rows() + expect( + rows + .filter((row) => row.body.kind === 'message' && row.body.role === 'user') + .map((row) => row.body) + ).toEqual([ + { + kind: 'message', + role: 'user', + blocks: [ + { type: 'text', text: 'read this' }, + { type: 'image-ref', url: 'data:image/png;base64,aW1hZ2U=' } + ] + } + ]) + await catchUp(frames) + expect(await fixture.rig.rows()).toEqual(rows) + }) + it.each(['refused', 'direct', 'queued', 'card', 'empty'] as const)( + 'places the divider from visible retained content with a %s submission', + async (scenario) => { + const { fixture, catchUp } = await setup(GROK_ACP_DIALECT, 'prompt-id') + const { journal } = fixture.rig + await journal.appendSubmission({ + clientMessageId: 'retained', + payloadFingerprint: 'retained', + fence: 1, + ...(scenario === 'direct' ? {} : { handoverRecorded: true }), + body: { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: scenario === 'empty' ? '' : 'keep my message' }] + } + }) + if (scenario !== 'direct' && scenario !== 'queued') { + await journal.resolveDispatch({ + clientMessageId: 'retained', + fence: 1, + state: 'pending', + turnScope: { kind: 'thread' } + }) + } + if (scenario === 'refused' || scenario === 'card') { + await journal.resolveDispatch({ + clientMessageId: 'retained', + fence: 1, + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('providerRejected'), { + surface: 'rejection' + }), + ...(scenario === 'card' ? { keptAsQueuedMessageId: 'card' } : {}) + }) + } + const before = await fixture.rig.rows() + const submissions = journal.submissions() + const replay = [ + userReplay('outside'), + { + method: 'session/update', + params: { + sessionId: NAMESPACE, + _meta: { promptId: 'outside' }, + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'answer' } + } + } + } + ] + await catchUp(replay) + const rows = await fixture.rig.rows() + expect(dividers(rows)).toHaveLength(scenario === 'refused' || scenario === 'direct' ? 1 : 0) + expect(rows.slice(0, before.length)).toEqual(before) + expect(journal.submissions()).toEqual(submissions) + await catchUp(replay) + expect(await fixture.rig.rows()).toEqual(rows) + } + ) + it.each(['queued-only', 'ownership-revision'] as const)( + 'captures immutable coverage before concurrent %s', + async (scenario) => { + const { fixture, catchUp } = await setup(OPENCODE_ACP_DIALECT, 'user-message-id') + const { rig } = fixture + const replay = [userReplay('outside'), answerReplay('prefix')] + await catchUp(replay) + const turn = (await rig.rows()).find((item) => item.body.kind === 'turn')! + const append = rig.eventSink.tryAppendTransition! + let intervened = false + rig.eventSink.tryAppendTransition = (transition) => { + if (!intervened) { + intervened = true + if (scenario === 'queued-only') { + void rig.journal.appendSubmission({ + clientMessageId: 'queued', + payloadFingerprint: 'queued', + fence: 1, + handoverRecorded: true, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'next' }] } + }) + } else { + void rig.journal.appendItem( + parseAgentJournalItemKey(turn.itemId)!, + agentJournalTurnBody({ + ...readAgentJournalTurn(turn.body)!, + userItemId: 'orca:concurrent-owner' + }), + { fence: 1, turnScope: { kind: 'thread' } } + ) + } + } + return append(transition) + } + await catchUp([userReplay('outside'), answerReplay('full')]) + expect(intervened).toBe(true) + const answer = (await rig.rows()).find( + (item) => item.body.kind === 'message' && item.body.role === 'assistant' + ) + expect(answer?.body).toMatchObject({ + blocks: [{ text: scenario === 'queued-only' ? 'full' : 'prefix' }] + }) + if (scenario === 'queued-only') { + expect(rig.journal.submission('queued')?.handedOverAt).toBeUndefined() + } + } + ) + it('bounds replay by UTF-8 bytes before staging', () => { + const onOverflow = vi.fn() + const history = new AcpLoadHistory(NAMESPACE, onOverflow) + history.keep('session/update', { + sessionId: NAMESPACE, + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: '🙂'.repeat(4 * 1024 * 1024) } + } + }) + expect(history.frames()).toBeNull() + history.keep('session/update', { sessionId: NAMESPACE }) + expect(onOverflow).toHaveBeenCalledOnce() + }) +}) diff --git a/src/main/acp/acp-history-catch-up.ts b/src/main/acp/acp-history-catch-up.ts new file mode 100644 index 00000000000..12553edfd1d --- /dev/null +++ b/src/main/acp/acp-history-catch-up.ts @@ -0,0 +1,240 @@ +import { + agentJournalItemKey, + agentJournalSubmissionKey, + parseAgentJournalItemKey +} from '../../shared/agent-session-journal-item-key' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalRenderItem +} from '../../shared/agent-session-journal-types' +import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' +import { nativeChatRowRendersContent } from '../../shared/native-chat-row-content' +import { projectNativeChatTranscriptMessages } from '../../shared/native-chat-transcript-projection' +import { projectStructuredAgentSessionMessages } from '../../shared/structured-agent-session-message-projection' +import { usableJournalHistoryOrigin } from '../native-chat/agent-session-journal/journal-history-origin' +import { journalLifecycleItemMutation } from '../native-chat/agent-session-journal/journal-lifecycle-mutation' +import { isHistoricalProviderTimelineItem } from '../native-chat/agent-session-timeline/provider-timeline-history' +import { isProviderTimelineTurnInNamespace } from '../native-chat/agent-session-timeline/provider-timeline-identity' +import type { + StructuredAgentSessionEventSink, + StructuredAgentSessionHistoryJournal +} from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { + continuedOutsideNotice, + continuedOutsideRecordId, + createHistoryCatchUpBatch, + runHistoryCatchUp +} from '../native-chat/agent-session-wire/structured-agent-session-history-catch-up' +import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' +import { + acpHistoryTurns, + type AcpIdentifiedHistoryTurn, + type AcpLoadHistoryFrame +} from './acp-history-replay' +import { translateHistoryTurn, type AcpHistoryScope } from './acp-history-translation' +import { matchAcpHistoryTurns, type AcpHistoryCandidate } from './acp-history-turn-matching' +import { acpAgentName } from './acp-structured-agent-definitions' + +export type AcpHistoryCatchUpInput = AcpHistoryScope & { + frames: readonly AcpLoadHistoryFrame[] + sink: StructuredAgentSessionEventSink + assertCurrent: () => void + timeoutMs?: number + logger?: StructuredAgentSessionLogger +} + +/** Optional history work owns no sends; one checked runner settlement lands all changed bodies. */ +export async function catchUpAcpHistory(input: AcpHistoryCatchUpInput): Promise { + let candidates: AcpHistoryCandidate[] = [] + await runHistoryCatchUp({ + agentName: acpAgentName(input.agent), + scope: 'acp-history-catch-up', + sessionId: input.sessionId, + sink: input.sink, + assertCurrent: input.assertCurrent, + timeoutMs: input.timeoutMs ?? 30_000, + ...(input.logger ? { logger: input.logger } : {}), + readView: (history) => journalView(input, history, candidates), + sameView: (a, b) => a.token === b.token, + prepare: async ({ journal, view, check, yieldTurn }) => { + // Derive source keys once without saved facts, then capture before any saved read or yield. + candidates = acpHistoryTurns(input.frames, input.identity, check) + .filter((turn): turn is AcpIdentifiedHistoryTurn => turn.turn !== null) + .map((source) => ({ + source, + items: translateHistoryTurn(input, source, check) + })) + const baseline = view() + const { natives, kind, matched } = matchAcpHistoryTurns( + input, + baseline, + candidates, + baseline.importedAt, + baseline.requests, + check + ) + const batch = createHistoryCatchUpBatch() + // The rows written name the import, so a later open's additions settle under their own id. + const written: string[] = [] + const read = (key: string) => baseline.items.get(key) ?? null + let divided = false + for (const { source, items } of candidates) { + check() + const coverage = kind.get(source.turn) + if (coverage === 'native' || coverage === 'adopted') { + continue + } + for (const mutation of items) { + check() + const key = agentJournalItemKey(mutation.identity) + const canonical = baseline.canonical.get(key) ?? key + const held = read(canonical) + if (canonical !== key || (held && !isHistoricalProviderTimelineItem(held, read))) { + throw new Error('unowned-history-target') + } + if (held && JSON.stringify(held.body) === JSON.stringify(mutation.body)) { + continue + } + if (!held && coverage !== 'imported' && !divided && baseline.retained) { + const divider = journalLifecycleItemMutation( + { turnScope: AGENT_JOURNAL_THREAD_SCOPE }, + { + provider: 'legacy', + agent: input.agent, + sessionId: input.sessionId, + recordId: continuedOutsideRecordId(source.turn) + }, + continuedOutsideNotice( + natives.length > 0 && ![...matched.values()].includes(natives.length - 1) + ) + ) + const dividerKey = agentJournalItemKey(divider.identity) + if (!read(dividerKey)) { + batch.push(divider) + written.push(dividerKey) + } + divided = true + } + if (!held) { + batch.push(mutation) + written.push(key) + } else if (mutation.kind === 'item') { + // Ordinary revisions retain the creating scope/position and an unstated producer. + batch.revise( + { ...mutation, linkage: undefined }, + { canonicalItemId: canonical, revision: held.revision } + ) + } else { + throw new Error('unowned-history-target') + } + } + await yieldTurn() + } + return batch.size + ? batch.finish('acp-history-import', [journal.epoch, input.providerSessionId, written]) + : null + } + }) +} + +/** Only dependencies used for coverage, ownership, bodies and boundary policy enter the view. */ +function journalView( + scope: AcpHistoryCatchUpInput, + history: StructuredAgentSessionHistoryJournal, + candidates: readonly AcpHistoryCandidate[] +) { + const journal = scope.sink.journalItems?.() + if (!journal) { + throw new Error('no-history-journal') + } + const all: AgentJournalRenderItem[] = [] + history.visitItems((key) => { + const item = history.item(key) + if (item) { + all.push(item) + } + }) + const submissions = history.submissions() + const eligible = submissions.filter( + (send) => send.handedOverAt !== undefined || send.handoverRecorded !== true + ) + const eligibleUsers = new Set( + eligible + .filter((send) => send.keptAsQueuedMessageId === undefined) + .map((send) => agentJournalSubmissionKey(send.clientMessageId)) + ) + const retained = projectNativeChatTranscriptMessages( + projectStructuredAgentSessionMessages(all, [], submissions) + ).some( + (message) => + nativeChatRowRendersContent(message.blocks) && + (parseAgentJournalItemKey(message.id)?.provider !== 'orca' || + (message.role === 'user' && eligibleUsers.has(message.id))) + ) + const turnKeys = new Set( + all.flatMap((item) => { + const turn = readAgentJournalTurn(item.body) + const id = parseAgentJournalItemKey(item.itemId) + return turn && + id?.provider === 'legacy' && + id.agent === scope.agent && + id.sessionId === scope.sessionId && + isProviderTimelineTurnInNamespace(turn.turnId, scope.providerSessionId) + ? [item.itemId] + : [] + }) + ) + const userKeys = new Set( + all.flatMap((item) => { + const turn = readAgentJournalTurn(item.body) + return turnKeys.has(item.itemId) && turn?.userItemId ? [turn.userItemId] : [] + }) + ) + const items = new Map( + all + .filter((item) => { + const id = parseAgentJournalItemKey(item.itemId) + return ( + turnKeys.has(item.itemId) || + userKeys.has(item.itemId) || + (item.turnScope?.kind === 'turn' && turnKeys.has(item.turnScope.turnItemId)) || + (id?.provider === 'legacy' && + id.agent === scope.agent && + id.sessionId === scope.providerSessionId) + ) + }) + .map((item) => [item.itemId, item]) + ) + const keys = candidates.flatMap(({ source, items }) => [ + ...items.map((item) => agentJournalItemKey(item.identity)), + agentJournalItemKey({ + provider: 'legacy', + agent: scope.agent, + sessionId: scope.sessionId, + recordId: continuedOutsideRecordId(source.turn) + }) + ]) + const canonical = new Map( + [...keys, ...items.keys()].map((key) => [key, journal.canonicalItemId(key)]) + ) + // Aliases are rejected; their target bodies cannot establish exact provider coverage. + const requests = new Set(eligible.map((send) => `prompt:${send.clientMessageId}`)) + const origin = usableJournalHistoryOrigin(history.historyOrigin()) + const handle = origin?.providerHandle + const importedAt = + origin && + handle?.kind === 'opaque' && + handle.agent === scope.agent && + handle.value === scope.providerSessionId + ? origin.importedAt + : null + const token = JSON.stringify([ + history.epoch, + [...items.values()].map((item) => [item.itemId, item.revision]), + [...canonical], + [...requests], + importedAt, + retained + ]) + return { items, canonical, requests, importedAt, retained, token } +} diff --git a/src/main/acp/acp-history-journal-overlay.ts b/src/main/acp/acp-history-journal-overlay.ts new file mode 100644 index 00000000000..51d644f6733 --- /dev/null +++ b/src/main/acp/acp-history-journal-overlay.ts @@ -0,0 +1,96 @@ +import type { AgentJournalRenderItem } from '../../shared/agent-session-journal-types' +import { parseAgentJournalItemKey } from '../../shared/agent-session-journal-item-key' +import { + journalLifecycleItemMutation, + journalLifecycleMutationItemId, + type JournalLifecycleIdentityMutationInput, + type JournalLifecycleMutationInput +} from '../native-chat/agent-session-journal/journal-lifecycle-mutation' +import type { ProviderTimelineSink } from '../native-chat/agent-session-timeline/provider-timeline-plan' +import type { StructuredAgentSessionTransitionJournal } from '../native-chat/agent-session-wire/structured-agent-session-transition' + +/** Pure per-turn scratch: held durable rows never affect replay translation. */ +export type AcpHistoryItem = Extract + +export class AcpHistoryJournalOverlay { + private readonly staged = new Map< + string, + { + item: AgentJournalRenderItem + mutation: AcpHistoryItem + } + >() + readonly journal: StructuredAgentSessionTransitionJournal + readonly sink: ProviderTimelineSink + + constructor(epoch: string) { + const item = (key: string) => this.staged.get(key)?.item ?? null + const visitItemsWithLinkage: StructuredAgentSessionTransitionJournal['visitItemsWithLinkage'] = + (visit) => + this.staged.forEach(({ item }) => visit(item.itemId, item.sequence, item.body, item)) + this.journal = { + epoch, + item, + canonicalItemId: (key) => key, + itemBody: (key) => item(key)?.body ?? null, + visitItems: visitItemsWithLinkage, + visitItemsWithLinkage + } + this.sink = { + journalItems: () => this.journal, + tryAppendTransition: (transition) => { + for (const step of transition.steps) { + if (step.kind === 'settlement') { + step.resolve(this.journal).forEach((mutation) => this.write(mutation)) + } else { + const resolved = step.resolve(this.journal) + if (resolved) { + this.write( + journalLifecycleItemMutation(step.options, resolved.identity, resolved.body) + ) + } + } + } + return { accepted: true } + } + } + } + + created(): AcpHistoryItem[] { + return [...this.staged.values()].map(({ mutation }) => mutation) + } + + private write(mutation: JournalLifecycleMutationInput): void { + if (mutation.kind !== 'item') { + throw new Error('unexpected-history-tombstone') + } + const itemId = journalLifecycleMutationItemId(mutation) + const identity = mutation.identity ?? parseAgentJournalItemKey(itemId) + if (!identity) { + throw new Error('unkeyed-history-row') + } + const previous = this.staged.get(itemId) + const turnScope = previous?.item.turnScope ?? mutation.turnScope + this.staged.set(itemId, { + item: { + ...mutation.linkage, + itemId, + body: mutation.body, + turnScope, + revision: (previous?.item.revision ?? 0) + 1, + sequence: previous?.item.sequence ?? this.staged.size + 1, + observedAt: 0 + }, + mutation: { + ...(previous?.mutation ?? mutation), + identity, + itemId: undefined, + body: mutation.body, + turnScope, + ...(mutation.body.kind === 'message' && mutation.body.role === 'user' + ? { echoOf: null } + : {}) + } + }) + } +} diff --git a/src/main/acp/acp-history-ordered-availability.ts b/src/main/acp/acp-history-ordered-availability.ts new file mode 100644 index 00000000000..a3f191efc20 --- /dev/null +++ b/src/main/acp/acp-history-ordered-availability.ts @@ -0,0 +1,57 @@ +/** Invocation-local successor queries over sorted coordinates, with point activation. */ +export class OrderedAvailability { + private readonly capacity: number + private readonly maximum: number[] + + constructor( + private readonly coordinates: readonly number[], + private readonly check: () => void, + enabled = true + ) { + let capacity = 1 + while (capacity < coordinates.length) { + check() + capacity *= 2 + } + this.capacity = capacity + this.maximum = Array.from({ length: capacity * 2 }, () => -Infinity) + for (let index = 0; index < coordinates.length; index++) { + check() + if (enabled) { + this.maximum[capacity + index] = coordinates[index] + } + } + for (let index = capacity - 1; index > 0; index--) { + check() + this.maximum[index] = Math.max(this.maximum[index * 2], this.maximum[index * 2 + 1]) + } + } + + after(floor: number): { index: number; value: number } | undefined { + this.check() + if (this.maximum[1] <= floor) { + return undefined + } + let node = 1 + while (node < this.capacity) { + this.check() + node *= 2 + if (this.maximum[node] <= floor) { + node += 1 + } + } + const index = node - this.capacity + return { index, value: this.coordinates[index] } + } + + set(index: number, enabled: boolean): void { + this.check() + let node = this.capacity + index + this.maximum[node] = enabled ? this.coordinates[index] : -Infinity + while (node > 1) { + this.check() + node = Math.floor(node / 2) + this.maximum[node] = Math.max(this.maximum[node * 2], this.maximum[node * 2 + 1]) + } + } +} diff --git a/src/main/acp/acp-history-replay.ts b/src/main/acp/acp-history-replay.ts new file mode 100644 index 00000000000..6628495242e --- /dev/null +++ b/src/main/acp/acp-history-replay.ts @@ -0,0 +1,249 @@ +// Replay grouping uses provider message/turn IDs and preserves source arrival order. + +import { z } from 'zod' +import type { NativeChatBlock } from '../../shared/native-chat-types' + +/** What one load may keep for catch-up; a larger history is not caught up at all. */ +const LOAD_HISTORY_MAX_FRAMES = 20_000 +const LOAD_HISTORY_MAX_BYTES = 16 * 1024 * 1024 + +/** One notification the agent sent while it replayed its history, as received. */ +export type AcpLoadHistoryFrame = { method: string; params: unknown } + +/** A bounded copy of what one session's load replayed, in arrival order, its own frames only. */ +export class AcpLoadHistory { + private kept: AcpLoadHistoryFrame[] | null = [] + private bytes = 0 + + constructor( + private readonly providerSessionId: string, + private readonly onOverflow?: () => void + ) {} + + keep(method: string, params: unknown): void { + if (!this.kept || sessionSchema.safeParse(params).data?.sessionId !== this.providerSessionId) { + return + } + this.bytes += Buffer.byteLength(JSON.stringify(params), 'utf8') + this.kept.push({ method, params }) + if (this.kept.length > LOAD_HISTORY_MAX_FRAMES || this.bytes > LOAD_HISTORY_MAX_BYTES) { + this.kept = null + this.onOverflow?.() + } + } + + /** Null once the history outgrew what a load keeps. */ + frames(): AcpLoadHistoryFrame[] | null { + return this.kept + } +} + +/** Where a replayed turn's id comes from. */ +export type AcpHistoryTurnIdentity = 'prompt-id' | 'user-message-id' + +type AcpHistoryUser = { + blocks: NativeChatBlock[] + boundary: AcpLoadHistoryFrame + eventId?: string +} + +export type AcpHistoryTurn = { + /** The agent's id for the turn; null when its replay names none. */ + turn: string | null + /** The execution's root message, in the order its chunks came. */ + user: AcpHistoryUser + textOnly: boolean + /** Later messages and output in source order, no longer marked as replay. */ + observations: (AcpHistoryUser | AcpLoadHistoryFrame)[] + /** Every message id the replay gave this turn (OpenCode): the person's and the agent's. */ + messageIds: string[] + /** When the agent recorded the person's message (Grok), on the agent host's clock. */ + startedAt: number | null +} + +/** A replayed turn under the agent's own id. */ +export type AcpIdentifiedHistoryTurn = AcpHistoryTurn & { turn: string } + +const sessionSchema = z.object({ sessionId: z.string() }) +const looseRecord = z.record(z.string(), z.unknown()) +const frameSchema = z.looseObject({ + update: z.looseObject({ + sessionUpdate: z.string(), + messageId: z.string().optional(), + prompt_id: z.string().optional(), + content: z.unknown().optional(), + _meta: z.looseObject({ promptIndex: z.number().int().nonnegative().optional() }).optional() + }), + _meta: z + .looseObject({ + promptId: z.string().optional(), + eventId: z.string().min(1).optional(), + agentTimestampMs: z.number().optional() + }) + .optional() +}) +const contentSchema = z.discriminatedUnion('type', [ + z.looseObject({ + type: z.literal('text'), + text: z.string(), + _meta: z.looseObject({ displayText: z.unknown().optional() }).optional() + }), + z.looseObject({ type: z.literal('image'), mimeType: z.string(), data: z.string() }), + z.looseObject({ type: z.literal('resource_link'), uri: z.string(), name: z.string().optional() }), + z.looseObject({ + type: z.literal('resource'), + resource: z.looseObject({ uri: z.string(), text: z.string().optional() }) + }) +]) + +/** OpenCode names reasoning `:reasoning:`; the message is what its store keys. */ +const REASONING_SUFFIX = /:reasoning:\d+$/ + +function userBlock(block: z.infer): NativeChatBlock { + switch (block.type) { + case 'text': + return { + type: 'text', + text: typeof block._meta?.displayText === 'string' ? block._meta.displayText : block.text + } + case 'image': + return { type: 'image-ref', url: `data:${block.mimeType};base64,${block.data}` } + case 'resource_link': + return { type: 'text', text: block.name ?? block.uri } + case 'resource': + return { type: 'text', text: block.resource.text ?? block.resource.uri } + } +} + +/** The frame with the agent's replay marking removed, so it reads as the turn it records. */ +function unmarked(frame: AcpLoadHistoryFrame): AcpLoadHistoryFrame { + const params = looseRecord.safeParse(frame.params) + const meta = params.success ? looseRecord.safeParse(params.data._meta) : null + if (!params.success || !meta?.success) { + return frame + } + const { isReplay: _isReplay, ...rest } = meta.data + return { method: frame.method, params: { ...params.data, _meta: rest } } +} + +/** The replay's turns in the agent's order. Throws when one turn names two prompt ids. */ +export function acpHistoryTurns( + frames: readonly AcpLoadHistoryFrame[], + identity: AcpHistoryTurnIdentity, + check: () => void = () => {} +): AcpHistoryTurn[] { + const turns: AcpHistoryTurn[] = [] + const messageIds = new Set() + let current: AcpHistoryTurn | undefined + let lastUser: AcpHistoryUser | undefined + let userKey: string | number | undefined + let inUser = false + let marked = false + let ended = false + const nameTurn = (turn: AcpHistoryTurn, id: string | undefined) => { + if (id === undefined || id === turn.turn) { + return + } + if (turn.turn !== null) { + throw new Error('conflicting-history-turn') + } + turn.turn = id + } + for (const frame of frames) { + check() + const parsed = frameSchema.safeParse(frame.params) + if (!parsed.success) { + throw new Error('unrecognised-history-frame') + } + const { update, _meta: meta } = parsed.data + if (update.sessionUpdate === 'user_message_chunk') { + const key = identity === 'user-message-id' ? update.messageId : update._meta?.promptIndex + marked ||= update._meta?.promptIndex !== undefined + // After marked roots appear, unmarked users stay inside their execution. + const root = + identity === 'user-message-id' + ? !inUser || userKey !== key + : update._meta?.interjection !== true && + (!marked || key !== undefined) && + (!inUser || userKey !== key) + if (update._meta?.hostTurn === true || (identity === 'user-message-id' && !key)) { + throw new Error('unattributed-history-user') + } + if (root) { + messageIds.clear() + current = { + turn: null, + user: { blocks: [], boundary: unmarked(frame) }, + textOnly: true, + observations: [], + messageIds: [], + startedAt: meta?.agentTimestampMs ?? null + } + turns.push(current) + lastUser = current.user + ended = false + if (identity === 'user-message-id') { + nameTurn(current, update.messageId) + } + } else if (!current || ended) { + throw new Error('unattributed-history-user') + } else if ( + lastUser !== current.user || + !inUser || + userKey !== key || + update._meta?.interjection === true + ) { + if (!meta?.eventId) { + throw new Error('unkeyed-history-user') + } + if (!inUser || lastUser?.eventId !== meta.eventId) { + lastUser = { blocks: [], boundary: unmarked(frame), eventId: meta.eventId } + current.observations.push(lastUser) + } + } + if (!current || !lastUser) { + throw new Error('unattributed-history-user') + } + inUser = true + userKey = key + const raw = contentSchema.safeParse(update.content) + if (!raw.success) { + throw new Error('unrecognised-history-user-content') + } + const hidden = update._meta?.hideFromScrollback === true + if (lastUser === current.user ? raw.data.type !== 'text' || hidden : !hidden) { + current.textOnly = false + } + if (!hidden) { + lastUser.blocks.push(userBlock(raw.data)) + } + } else if (current) { + inUser = false + current.observations.push(unmarked(frame)) + if (identity === 'prompt-id') { + nameTurn(current, meta?.promptId) + if (update.sessionUpdate === 'turn_completed') { + nameTurn(current, update.prompt_id) + ended = true + } + } + } else { + continue + } + if (identity === 'user-message-id' && update.messageId) { + const id = update.messageId.replace(REASONING_SUFFIX, '') + if (!messageIds.has(id)) { + messageIds.add(id) + current.messageIds.push(id) + } + } + } + if (turns.some((turn) => turn.turn === null)) { + throw new Error('unattributed-history-turn') + } + return turns +} + +export function acpHistoryFrameTime(frame: AcpLoadHistoryFrame): number | null { + return frameSchema.safeParse(frame.params).data?._meta?.agentTimestampMs ?? null +} diff --git a/src/main/acp/acp-history-scaling-counts.test-fixture.ts b/src/main/acp/acp-history-scaling-counts.test-fixture.ts new file mode 100644 index 00000000000..55743a0753e --- /dev/null +++ b/src/main/acp/acp-history-scaling-counts.test-fixture.ts @@ -0,0 +1,73 @@ +export function hasArrayKey(array: unknown[], key: PropertyKey): key is keyof unknown[] { + return key in array +} + +// Raw callsites avoid source-map formatting on every delegated operation. +export function operationStack(): string { + const prepare = Error.prepareStackTrace + try { + Error.prepareStackTrace = (_error, sites) => { + let result = 'operation attribution' + for (const site of sites) { + result += `\n at ${site.getFunctionName() ?? site.getMethodName() ?? ''} (${site.getFileName() ?? ''})` + } + return result + } + const error = new Error('operation attribution') + Error.captureStackTrace(error, operationStack) + return error.stack ?? '' + } finally { + Error.prepareStackTrace = prepare + } +} + +export type ViewWork = { + visitedRows: number + itemReads: number + submissionReturns: number + submissionCopies: number + canonicalEntries: number + canonicalReads: number +} +type ExecutionPass = { + phase: string + construction: number + predicates: number + reciprocalReads: number + executionItemIds: string[] +} +export function newCounts() { + const views: ViewWork[] = [] + const execution: ExecutionPass[] = [] + const snapshots: { phase: string; rows: number }[] = [] + return { + native: 0, + suffix: 0, + membership: 0, + tree: 0, + matching: 0, + copies: 0, + adopted: 0, + views, + execution, + snapshots, + frames: 0, + inputUtf8Bytes: 0, + frameVisits: 0, + candidates: 0, + translationObservations: 0, + translatedItems: 0, + candidateKeyUtf8Bytes: 0, + retainedRows: 0, + submissions: 0, + canonicalKeyUtf8Bytes: 0, + longestCanonicalKey: 0, + resolverCanonicalReads: 0, + otherCanonicalReads: 0, + otherItemReads: 0, + otherSubmissionReturns: 0, + otherSubmissionCopies: 0, + mutations: 0, + commits: 0 + } +} diff --git a/src/main/acp/acp-history-scaling-operations.test-fixture.ts b/src/main/acp/acp-history-scaling-operations.test-fixture.ts new file mode 100644 index 00000000000..db3b496ac35 --- /dev/null +++ b/src/main/acp/acp-history-scaling-operations.test-fixture.ts @@ -0,0 +1,269 @@ +import { Session } from 'node:inspector/promises' +import { hasArrayKey, newCounts, operationStack } from './acp-history-scaling-counts.test-fixture' + +type CountedArrayFactory = ( + this: ArrayConstructor, + input: Iterable | ArrayLike, + mapper?: (value: unknown, index: number) => unknown, + thisArg?: unknown +) => unknown[] + +// Count executed production blocks and built-in indexed reads, without replacing any result. +export async function work(run: (counts: ReturnType) => Promise) { + const inspector = new Session() + inspector.connect() + const counts = newCounts() + const sorted = Object.getOwnPropertyDescriptor(Array.prototype, 'toSorted')! + const slice = Object.getOwnPropertyDescriptor(Array.prototype, 'slice')! + const includes = Object.getOwnPropertyDescriptor(Array.prototype, 'includes')! + const some = Object.getOwnPropertyDescriptor(Array.prototype, 'some')! + const values = Object.getOwnPropertyDescriptor(Map.prototype, 'values')! + const from = Object.getOwnPropertyDescriptor(Array, 'from')! + const get = Object.getOwnPropertyDescriptor(Map.prototype, 'get')! + const set = Object.getOwnPropertyDescriptor(Map.prototype, 'set')! + const has = Object.getOwnPropertyDescriptor(Set.prototype, 'has')! + const map = Object.getOwnPropertyDescriptor(Array.prototype, 'map')! + const filter = Object.getOwnPropertyDescriptor(Array.prototype, 'filter')! + const sortArray = Array.prototype.toSorted + const sliceArray = Array.prototype.slice + const arrayIncludes = Array.prototype.includes + const arraySome = Array.prototype.some + const setHas = Set.prototype.has + const mapValues = Map.prototype.values + const arrayFrom: CountedArrayFactory = Array.from + const mapGet = Map.prototype.get + const mapSet = Map.prototype.set + const arrayMap = Array.prototype.map + const arrayFilter = Array.prototype.filter + const countedArray = ( + array: unknown[], + counter: 'native' | 'suffix' | 'membership' | 'copies' | 'tree' + ) => + new Proxy(array, { + get(target, key) { + if (typeof key === 'string' && /^\d+$/.test(key)) { + counts[counter]++ + } + return hasArrayKey(target, key) ? target[key] : undefined + }, + set(target, key, value, receiver) { + if (counter === 'tree' && typeof key === 'string' && /^\d+$/.test(key)) { + counts.tree++ + } + return Reflect.set(target, key, value, receiver) + } + }) + await inspector.post('Profiler.enable') + await inspector.post('Debugger.enable') + await inspector.post('Profiler.startPreciseCoverage', { callCount: true, detailed: true }) + Object.defineProperty(Array.prototype, 'toSorted', { + ...sorted, + value: function (this: unknown[], compare?: (a: unknown, b: unknown) => number) { + const result: unknown[] = sortArray.call(this, compare) + return operationStack()?.split('\n')[2]?.includes('acp-history-turn-matching.ts') + ? countedArray(result, 'native') + : result + } + }) + Object.defineProperty(Array.prototype, 'slice', { + ...slice, + value: function (this: unknown[], start?: number, end?: number) { + const result: unknown[] = sliceArray.call(this, start, end) + return operationStack()?.split('\n')[2]?.includes('acp-history-turn-matching.ts') + ? countedArray(result, 'suffix') + : result + } + }) + Object.defineProperty(Array.prototype, 'includes', { + ...includes, + value: function (this: unknown[], value: unknown, start?: number) { + const array = operationStack()?.split('\n')[2]?.includes('acp-history-replay.ts') + ? countedArray(this, 'membership') + : this + return arrayIncludes.call(array, value, start) + } + }) + Object.defineProperty(Array.prototype, 'some', { + ...some, + value: function ( + this: unknown[], + predicate: (value: unknown, index: number, array: unknown[]) => unknown, + thisArg?: unknown + ) { + const array = operationStack() + ?.split('\n')[2] + ?.includes('structuredAgentSessionRejectedShownInPlace') + ? countedArray(this, 'copies') + : this + return arraySome.call(array, predicate, thisArg) + } + }) + Object.defineProperty(Array.prototype, 'map', { + ...map, + value: function ( + this: unknown[], + callback: (value: unknown, index: number, array: unknown[]) => unknown, + thisArg?: unknown + ) { + const stack = operationStack() + const caller = stack.split('\n')[2] ?? '' + const provenance = caller.includes('providerTimelineExecutionItems') + if (provenance) { + counts.execution.push({ + phase: stack.split('\n')[3]?.trim() ?? '', + construction: 0, + predicates: 0, + reciprocalReads: 0, + executionItemIds: [] + }) + } + if (caller.includes('journalView') && this.every((key) => typeof key === 'string')) { + counts.views.at(-1)!.canonicalEntries += this.length + } + const result: unknown[] = arrayMap.call(this, (value, index, array) => { + if (provenance) { + counts.execution.at(-1)!.construction++ + } + return callback.call(thisArg, value, index, array) + }) + if (caller.includes('acp-history-translation.ts')) { + counts.translatedItems += result.length + } + return result + } + }) + Object.defineProperty(Array.prototype, 'filter', { + ...filter, + value: function ( + this: unknown[], + predicate: (value: unknown, index: number, array: unknown[]) => unknown, + thisArg?: unknown + ) { + const provenance = operationStack().split('\n')[2]?.includes('providerTimelineExecutionItems') + const result: unknown[] = arrayFilter.call(this, (value, index, array) => { + if (provenance) { + counts.execution.at(-1)!.predicates++ + } + return predicate.call(thisArg, value, index, array) + }) + if (provenance) { + const ids = counts.execution.at(-1)!.executionItemIds + for (const item of result) { + if ( + typeof item !== 'object' || + item === null || + !('itemId' in item) || + typeof item.itemId !== 'string' + ) { + throw new Error('expected execution journal item') + } + ids.push(item.itemId) + } + } + return result + } + }) + Object.defineProperty(Set.prototype, 'has', { + ...has, + value: function (this: Set, value: unknown) { + if (operationStack()?.split('\n')[2]?.includes('acp-history-replay.ts')) { + counts.membership++ + } + return setHas.call(this, value) + } + }) + Object.defineProperty(Map.prototype, 'values', { + ...values, + value: function (this: Map) { + const iterator: IterableIterator = mapValues.call(this) + if (!operationStack().includes('acp-history-turn-matching.ts')) { + return iterator + } + return (function* () { + for (const value of iterator) { + if (typeof value === 'number') { + counts.adopted++ + } + yield value + } + })() + } + }) + Object.defineProperty(Array, 'from', { + ...from, + value: function ( + this: ArrayConstructor, + input: Iterable | ArrayLike, + mapper?: (value: unknown, index: number) => unknown, + thisArg?: unknown + ) { + const result = arrayFrom.call(this, input, mapper, thisArg) + return operationStack().includes('acp-history-ordered-availability.ts') + ? countedArray(result, 'tree') + : result + } + }) + Object.defineProperty(Map.prototype, 'get', { + ...get, + value: function (this: Map, key: unknown) { + const stack = operationStack() + if (stack.includes('structuredAgentSessionRejectedShownInPlace')) { + counts.copies++ + } + if ( + stack.includes('isHistoricalProviderTimelineItem') && + stack.includes('providerTimelineExecutionItems') + ) { + counts.execution.at(-1)!.reciprocalReads++ + } + return mapGet.call(this, key) + } + }) + Object.defineProperty(Map.prototype, 'set', { + ...set, + value: function (this: Map, key: unknown, value: unknown) { + const stack = operationStack() + if (stack.includes('structuredAgentSessionRejectedShownInPlace')) { + counts.copies++ + } + return mapSet.call(this, key, value) + } + }) + try { + await run(counts) + const coverage = await inspector.post('Profiler.takePreciseCoverage') + for (const script of coverage.result) { + const path = script.url.split('?')[0] + if ( + !/acp-history-(ordered-availability|turn-matching)\.ts$|structured-agent-session-message-projection\.ts$/.test( + path + ) + ) { + continue + } + for (const fn of script.functions) { + const total = fn.ranges.reduce((sum, block) => sum + block.count, 0) + if (path.endsWith('acp-history-turn-matching.ts')) { + counts.matching += total + } + } + } + console.info('ACP history underlying work', JSON.stringify(counts)) + return counts + } finally { + Object.defineProperty(Array.prototype, 'toSorted', sorted) + Object.defineProperty(Array.prototype, 'slice', slice) + Object.defineProperty(Array.prototype, 'includes', includes) + Object.defineProperty(Array.prototype, 'some', some) + Object.defineProperty(Map.prototype, 'values', values) + Object.defineProperty(Array, 'from', from) + Object.defineProperty(Map.prototype, 'get', get) + Object.defineProperty(Map.prototype, 'set', set) + Object.defineProperty(Set.prototype, 'has', has) + Object.defineProperty(Array.prototype, 'map', map) + Object.defineProperty(Array.prototype, 'filter', filter) + await inspector.post('Profiler.disable') + await inspector.post('Debugger.disable') + inspector.disconnect() + } +} diff --git a/src/main/acp/acp-history-scaling.test.ts b/src/main/acp/acp-history-scaling.test.ts new file mode 100644 index 00000000000..cc9ecdfc2c0 --- /dev/null +++ b/src/main/acp/acp-history-scaling.test.ts @@ -0,0 +1,656 @@ +import { afterEach, expect, it, vi } from 'vitest' +import type { AgentJournalRenderItem } from '../../shared/agent-session-journal-types' +import { + agentJournalItemKey, + agentJournalSubmissionKey +} from '../../shared/agent-session-journal-item-key' +import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' +import { agentSessionFailureFact } from '../../shared/agent-session-failure' +import { agentSessionFailureWords } from '../../shared/agent-session-failure-words' +import { + AGENT, + SESSION, + NAMESPACE, + closeProviderTimelineRigs, + openProviderTimelineRig, + providerTurnItemId +} from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support' +import { DISPATCH_DOUBT_PROVIDER_EXITED } from '../native-chat/agent-session-journal/journal-dispatch-doubt-reasons' +import { isHistoricalProviderTimelineItem } from '../native-chat/agent-session-timeline/provider-timeline-history' +import { settleObservedExitForTest } from '../native-chat/agent-session-wire/structured-agent-session-observed-exit.test-fixture' +import { captureUnfinishedStructuredAgentSessionWork } from '../native-chat/agent-session-wire/structured-agent-session-unfinished-work' +import { createDeferredStructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { + recordingStructuredAgentSessionLogger, + testEventSinkLogging +} from '../native-chat/agent-session-wire/structured-agent-session-logger-test-support' +import { + hasArrayKey, + operationStack, + type ViewWork +} from './acp-history-scaling-counts.test-fixture' +import { work } from './acp-history-scaling-operations.test-fixture' +import { catchUpAcpHistory } from './acp-history-catch-up' +import { + acpHistoryTurns, + type AcpIdentifiedHistoryTurn, + type AcpLoadHistoryFrame +} from './acp-history-replay' +import { translateHistoryTurn, type AcpHistoryScope } from './acp-history-translation' +import { matchAcpHistoryTurns } from './acp-history-turn-matching' +import { acpNamedTextKey } from './acp-turn-messages' +import { OPENCODE_ACP_DIALECT } from './acp-dialects/opencode-dialect' +import { GROK_ACP_DIALECT } from './acp-dialects/grok-dialect' + +afterEach(closeProviderTimelineRigs) +const scope: AcpHistoryScope = { + agent: AGENT, + sessionId: SESSION, + providerSessionId: NAMESPACE, + identity: 'user-message-id', + dialect: OPENCODE_ACP_DIALECT +} +const frame = ( + update: Record, + meta?: Record +): AcpLoadHistoryFrame => ({ + method: 'session/update', + params: { sessionId: NAMESPACE, update, ...(meta ? { _meta: meta } : {}) } +}) +const user = (id: string, text: string) => + frame({ sessionUpdate: 'user_message_chunk', messageId: id, content: { type: 'text', text } }) +const answer = (id: string) => + frame({ + sessionUpdate: 'agent_message_chunk', + messageId: id, + content: { type: 'text', text: 'answer' } + }) +const link = (id: string) => + frame({ + sessionUpdate: 'agent_message_chunk', + messageId: id, + content: { type: 'resource_link', uri: `file:///outside/${id}`, name: id } + }) +type Rig = Awaited> + +async function native(rig: Rig, id: string, text: string, proof?: string) { + await rig.journal.appendSubmission({ + clientMessageId: id, + payloadFingerprint: id, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] }, + fence: 1 + }) + await rig.journal.resolveDispatch({ + clientMessageId: id, + fence: 1, + state: 'accepted', + providerIdentity: null + }) + rig.assembler.apply({ + type: 'input.accepted', + clientMessageId: id, + requestedAt: 1000, + join: { turn: id } + }) + rig.assembler.apply({ type: 'turn.open', turn: id, at: 1000 }) + if (proof) { + rig.assembler.apply({ + type: 'text.delta', + item: { id: acpNamedTextKey(proof, 'assistant') }, + channel: 'assistant', + text: 'answer', + join: { thread: NAMESPACE, turn: id } + }) + rig.assembler.flush() + } + rig.assembler.apply({ type: 'turn.end', turn: id, state: 'completed', at: 1100 }) + await rig.rows() +} + +async function catchUp(rig: Pick, frames: AcpLoadHistoryFrame[]) { + const { logger, entries } = recordingStructuredAgentSessionLogger() + await catchUpAcpHistory({ + ...scope, + frames, + sink: rig.eventSink, + assertCurrent: () => {}, + logger + }) + expect(entries).toEqual([]) +} + +async function repeat(rig: Rig, frames: AcpLoadHistoryFrame[]) { + const before = await rig.rows() + const reopened = await rig.reopenJournal() + const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging()) + deferred.bind({ journal: reopened, fence: 1, publish: () => {} }) + try { + const counts = await measureCatchUp({ eventSink: deferred.sink }, frames, false) + expect(reopened.snapshot().items).toEqual(before) + return counts + } finally { + deferred.close() + await reopened.close() + } +} + +async function measureCatchUp( + rig: Pick, + frames: AcpLoadHistoryFrame[], + commit: boolean +) { + const history = rig.eventSink.journalHistory?.() + const journal = rig.eventSink.journalItems?.() + const append = rig.eventSink.tryAppendTransition + if (!history || !journal || !append) { + throw new Error('expected bound real journal and transition sink') + } + // Both readers and the resolver receive the same journal; spying preserves identity and epoch. + expect(history).toBe(journal) + let retainedRows = 0 + history.visitItems(() => retainedRows++) + const submissions = history.submissions().length + const sources = acpHistoryTurns(frames, scope.identity) + const counts = await work(async (count) => { + count.retainedRows = retainedRows + count.submissions = submissions + count.frames = frames.length + count.inputUtf8Bytes = Buffer.byteLength(JSON.stringify(frames), 'utf8') + count.candidates = sources.filter((source) => source.turn !== null).length + count.translationObservations = sources.reduce( + (sum, source) => sum + 1 + source.observations.length, + 0 + ) + count.candidateKeyUtf8Bytes = sources.reduce( + (sum, source) => sum + Buffer.byteLength(source.turn ?? '', 'utf8'), + 0 + ) + const visit = history.visitItems + const item = history.item + const sends = history.submissions + const canonical = journal.canonicalItemId + const spies = [ + vi.spyOn(history, 'visitItems').mockImplementation((callback) => { + const view: ViewWork = { + visitedRows: 0, + itemReads: 0, + submissionReturns: 0, + submissionCopies: 0, + canonicalEntries: 0, + canonicalReads: 0 + } + count.views.push(view) + visit.call(history, (...args) => { + view.visitedRows++ + callback(...args) + }) + }), + vi.spyOn(history, 'item').mockImplementation((key) => { + if (operationStack().includes('journalView')) { + count.views.at(-1)!.itemReads++ + } else { + count.otherItemReads++ + } + return item.call(history, key) + }), + vi.spyOn(history, 'submissions').mockImplementation(() => { + const result = sends.call(history) + if (operationStack().includes('journalView')) { + const view = count.views.at(-1)! + view.submissionReturns++ + view.submissionCopies += result.length + } else { + count.otherSubmissionReturns++ + count.otherSubmissionCopies += result.length + } + return result + }), + vi.spyOn(journal, 'canonicalItemId').mockImplementation((key) => { + const stack = operationStack() + count.canonicalKeyUtf8Bytes += Buffer.byteLength(key, 'utf8') + count.longestCanonicalKey = Math.max(count.longestCanonicalKey, key.length) + if (stack.includes('journalView')) { + count.views.at(-1)!.canonicalReads++ + } else if (stack.includes('structured-agent-session-history-catch-up.ts')) { + count.resolverCanonicalReads++ + } else { + count.otherCanonicalReads++ + } + return canonical.call(journal, key) + }) + ] + const countedFrames = new Proxy(frames, { + get(target, key) { + if (key === Symbol.iterator) { + return function* () { + for (const frame of target) { + count.frameVisits++ + yield frame + } + } + } + return hasArrayKey(target, key) ? target[key] : undefined + } + }) + try { + await catchUp( + { + eventSink: { + ...rig.eventSink, + tryAppendTransition: (transition) => + append.call(rig.eventSink, { + ...transition, + steps: transition.steps.map((step) => + step.kind !== 'settlement' + ? step + : { + ...step, + resolve: (current) => { + const mutations = step.resolve(current) + count.mutations += mutations.length + return mutations + } + } + ) + }) + } + }, + countedFrames + ) + } finally { + for (const spy of spies) { + spy.mockRestore() + } + } + }) + expect(counts.frameVisits).toBe(frames.length) + expect(counts.views).toHaveLength(commit ? 2 : 1) + for (const view of counts.views) { + expect(view.visitedRows).toBe(retainedRows) + expect(view.itemReads).toBe(retainedRows) + expect(view.submissionReturns).toBe(1) + expect(view.submissionCopies).toBe(submissions) + expect(view.canonicalReads).toBe(view.canonicalEntries) + } + // These fixtures create new keys or no plan, so there are no expected-revision entries. + expect(counts.resolverCanonicalReads).toBe(counts.mutations) + expect(counts.mutations > 0).toBe(commit) + return counts +} + +// Real journal I/O across the 20/200 population matrix needs the extended timeout on CI. +it('bounds actual first/reopened search and repeated source activation at 20 and 200', async () => { + const populations: { + n: number + k: number + first: Awaited> + unchanged: Awaited> + }[] = [] + for (const n of [20, 200]) { + for (const k of [0, 2 * n]) { + const rig = await openProviderTimelineRig() + for (let i = 0; i < k; i++) { + await native(rig, `native-${i}`, `native-${i}`) + } + const before = structuredClone(await rig.rows()) + const frames = Array.from({ length: n }, (_, i) => user(`outside-${i}`, `outside-${i}`)) + const first = await measureCatchUp(rig, frames, true) + const after = await rig.rows() + expect(after.slice(0, before.length)).toEqual(before) + expect( + after.filter((row) => row.body.kind === 'message' && row.body.role === 'user') + ).toHaveLength(n + k) + expect( + after + .slice(before.length) + .flatMap((row) => + row.body.kind === 'message' && row.body.role === 'user' ? [row.body.blocks] : [] + ) + ).toEqual(Array.from({ length: n }, (_, i) => [{ type: 'text', text: `outside-${i}` }])) + expect( + after.filter( + (row) => + row.body.kind === 'status' && row.body.presentation?.startsWith('continued-outside') + ) + ).toHaveLength(k === 0 ? 0 : 1) + const unchanged = await repeat(rig, frames) + for (const count of [first, unchanged]) { + expect(count.suffix).toBeLessThanOrEqual(n * Math.ceil(Math.log2(n + 1))) + expect(count.native).toBeLessThanOrEqual(2 * k + n * Math.ceil(Math.log2(k + 1))) + expect(count.tree).toBeGreaterThan(0) + expect(count.tree).toBeLessThan(32 * (n + k) * Math.ceil(Math.log2(n + k + 1))) + } + populations.push({ n, k, first, unchanged }) + } + const rig = await openProviderTimelineRig() + for (let i = 0; i < n; i++) { + await native(rig, `native-${i}`, `a-${i}`) + } + await native(rig, 'end', 'end', 'end-proof') + const before = structuredClone(await rig.rows()) + const frames = Array.from({ length: n }, (_, i) => [ + user(`A-${i}`, `a-${i}`), + user(`Z-${i}`, `outside-${i}`), + user(`A-${i}`, `a-${i}`) + ]).flat() + frames.push(user('E', 'end'), answer('end-proof')) + const count = await measureCatchUp(rig, frames, true) + const after = await rig.rows() + expect(after.slice(0, before.length)).toEqual(before) + expect( + after + .slice(before.length) + .flatMap((row) => (row.body.kind === 'message' ? [row.body.blocks] : [])) + ).toEqual(Array.from({ length: n }, (_, i) => [{ type: 'text', text: `outside-${i}` }])) + expect(count.tree).toBeLessThan(64 * n * Math.ceil(Math.log2(n + 1))) + populations.push({ n, k: -1, first: count, unchanged: count }) + } + for (const small of populations.filter((entry) => entry.n === 20)) { + const large = populations.find( + (entry) => entry.n === 200 && (small.k === -1 ? entry.k === -1 : entry.k === small.k * 10) + )! + expect(large.first.matching).toBeLessThan(small.first.matching * 20) + expect(large.unchanged.matching).toBeLessThan(small.unchanged.matching * 20) + } +}, 120000) + +// Real journal writes and both projected views at 20/200 entries need the extended CI timeout. +it('bounds grouped IDs and rejected copies through real empty and committed views', async () => { + for (const n of [20, 200]) { + const rig = await openProviderTimelineRig() + await native(rig, 'root', 'root', 'proof') + const before = structuredClone(await rig.rows()) + // Link observations retain distinct source IDs without keeping 200 text streams open. + const frames = [ + user('provider-root', 'root'), + answer('proof'), + ...Array.from({ length: n }, (_, i) => link(`part-${i}`)) + ] + const grouped = await measureCatchUp(rig, frames, false) + expect(grouped.membership).toBeGreaterThanOrEqual(n) + expect(grouped.membership).toBeLessThanOrEqual(n + 2) + expect(await rig.rows()).toEqual(before) + const parsed = acpHistoryTurns(frames, 'user-message-id') + expect(parsed[0].messageIds).toEqual([ + 'provider-root', + 'proof', + ...Array.from({ length: n }, (_, i) => `part-${i}`) + ]) + const rejected = await openProviderTimelineRig() + for (let i = 0; i < n; i++) { + await rejected.journal.appendSubmission({ + clientMessageId: `r-${i}`, + payloadFingerprint: 'same', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'repeat' }] }, + fence: 1 + }) + await rejected.journal.resolveDispatch({ + clientMessageId: `r-${i}`, + fence: 1, + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('providerRejected'), { + surface: 'rejection' + }) + }) + } + const seeded = structuredClone(await rejected.rows()) + const submissions = structuredClone(rejected.journal.submissions()) + const empty = await measureCatchUp(rejected, [], false) + expect(await rejected.rows()).toEqual(seeded) + const committed = await measureCatchUp(rejected, [user('outside', 'new')], true) + const after = await rejected.rows() + expect(after.slice(0, seeded.length)).toEqual(seeded) + expect( + after + .slice(seeded.length) + .flatMap((row) => (row.body.kind === 'message' ? [row.body.blocks] : [])) + ).toEqual([[{ type: 'text', text: 'new' }]]) + expect(rejected.journal.submissions()).toEqual(submissions) + expect(empty.copies).toBeGreaterThan(2 * n) + expect(empty.copies).toBeLessThan(16 * n) + expect(committed.copies).toBe(empty.copies * 2) + } +}, 120000) + +it('preserves regressing floors, future exact reservations, ties and non-consuming misses', async () => { + const rig = await openProviderTimelineRig() + for (const [id, text, proof] of [ + ['a', 'a'], + ['b', 'b'], + ['B', 'high'], + ['tie', 'tie'], + ['empty', ''], + ['end', 'end'] + ]) { + await native(rig, id, text, proof) + } + const items = new Map((await rig.rows()).map((row) => [row.itemId, row])) + const tie = items.get(providerTurnItemId('tie'))! + const adopted = { + itemId: agentJournalItemKey({ + provider: 'legacy', + agent: AGENT, + sessionId: NAMESPACE, + recordId: 'Q' + }), + sequence: tie.sequence, + revision: 1, + observedAt: 0, + body: { + kind: 'message' as const, + role: 'assistant' as const, + blocks: [{ type: 'text' as const, text: 'retained' }] + } + } + items.set(adopted.itemId, adopted) + const before = structuredClone([...items.values()]) + const frames = [ + user('A', 'a'), + user('B', 'high'), + user('A', 'a'), + user('C', 'b'), + user('O', 'tie'), + user('Q', 'adopted'), + user('P', 'tie'), + user('U', ''), + user('end', 'end'), + user('B', 'high'), + user('Z', 'end') + ] + const candidates = acpHistoryTurns(frames, scope.identity) + .filter((turn): turn is AcpIdentifiedHistoryTurn => turn.turn !== null) + .map((source) => ({ source, items: translateHistoryTurn(scope, source) })) + const result = matchAcpHistoryTurns(scope, { items }, candidates, null, new Set(), () => {}) + expect([...result.matched]).toEqual([ + ['B', 2], + ['end', 5], + ['A', 0], + ['C', 1], + ['P', 3], + ['U', 4] + ]) + expect(result.kind.get('Q')).toBe('adopted') + expect(result.kind.has('O')).toBe(false) + expect(result.kind.has('Z')).toBe(false) + expect([...items.values()]).toEqual(before) +}) + +it('keeps Grok timestamp adoption and never enumerates its unread adopted maximum', async () => { + const grok: AcpHistoryScope = { ...scope, identity: 'prompt-id', dialect: GROK_ACP_DIALECT } + for (const n of [20, 200]) { + const items = new Map( + Array.from({ length: n }, (_, i): [string, AgentJournalRenderItem] => { + const itemId = agentJournalItemKey({ + provider: 'legacy', + agent: AGENT, + sessionId: NAMESPACE, + recordId: `old-${i}` + }) + return [ + itemId, + { + itemId, + sequence: i + 1, + revision: 1, + observedAt: 0, + body: { kind: 'message', role: 'assistant', blocks: [] } + } + ] + }) + ) + const frames = Array.from({ length: n }, (_, i) => [ + frame( + { + sessionUpdate: 'user_message_chunk', + _meta: { promptIndex: i }, + content: { type: 'text', text: 'old' } + }, + { agentTimestampMs: 500 } + ), + frame( + { + sessionUpdate: 'agent_message_chunk', + content: { type: 'resource_link', uri: 'file:///outside/old' } + }, + { promptId: `prompt-${i}` } + ) + ]).flat() + const candidates = acpHistoryTurns(frames, grok.identity) + .filter((turn): turn is AcpIdentifiedHistoryTurn => turn.turn !== null) + .map((source) => ({ source, items: translateHistoryTurn(grok, source) })) + const count = await work(async () => { + const result = matchAcpHistoryTurns(grok, { items }, candidates, 500, new Set(), () => {}) + expect([...result.kind.values()]).toEqual(Array.from({ length: n }, () => 'adopted')) + }) + expect(count.adopted).toBe(0) + expect( + matchAcpHistoryTurns(grok, { items }, candidates, 499, new Set(), () => {}).kind.size + ).toBe(0) + expect(() => + matchAcpHistoryTurns(grok, { items }, candidates, null, new Set(), () => {}) + ).toThrow('no-legacy-origin') + } +}) + +// Importing and settling real 20/200-entry journals needs the extended timeout on CI. +it('counts observed-exit provenance passes while preserving imported history at 20 and 200', async () => { + for (const n of [20, 200]) { + const rig = await openProviderTimelineRig() + await catchUp( + rig, + Array.from({ length: n }, (_, i) => user(`archived-${i}`, 'past')) + ) + const archival = structuredClone(await rig.rows()) + expect(archival).toHaveLength(2 * n) + const archivedById = new Map(archival.map((row) => [row.itemId, row])) + expect( + archival.every((row) => + isHistoricalProviderTimelineItem(row, (key) => archivedById.get(key) ?? null) + ) + ).toBe(true) + await rig.journal.appendSubmission({ + clientMessageId: 'live', + payloadFingerprint: 'live', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'working' }] }, + fence: 1 + }) + rig.assembler.apply({ + type: 'input.accepted', + clientMessageId: 'live', + requestedAt: 1000, + join: { turn: 'live' } + }) + rig.assembler.apply({ type: 'turn.open', turn: 'live', at: 1000 }) + const before = await rig.rows() + const unfinished = captureUnfinishedStructuredAgentSessionWork(rig.journal) + expect(unfinished.items.map((item) => item.itemId)).toEqual([providerTurnItemId('live')]) + expect(unfinished.hadUnsettledSubmissions).toBe(true) + expect( + before.find((row) => row.itemId === agentJournalSubmissionKey('live'))?.turnScope + ).toEqual({ + kind: 'turn', + turnItemId: archival.findLast((row) => readAgentJournalTurn(row.body))!.itemId + }) + const snapshot = rig.journal.snapshot + const batch = rig.journal.appendPlannedLifecycleBatch + let batchCalls = 0 + let settlement: Awaited> | undefined + const submissionCount = rig.journal.submissions().length + const counts = await work(async (count) => { + count.retainedRows = before.length + count.submissions = submissionCount + rig.journal.observeCommits(() => count.commits++) + const spies = [ + vi.spyOn(rig.journal, 'snapshot').mockImplementation(() => { + const result = snapshot.call(rig.journal) + const phase = + operationStack() + .split('\n') + .find((line) => + /captureUnfinishedStructuredAgentSessionWork|planStructuredAgentSessionDeadGeneration|planStaleStructuredAgentSessionState/.test( + line + ) + ) ?? 'other' + count.snapshots.push({ phase: phase.trim(), rows: result.items.length }) + return result + }), + vi.spyOn(rig.journal, 'appendPlannedLifecycleBatch').mockImplementation((input) => { + batchCalls++ + return batch.call(rig.journal, input) + }) + ] + try { + settlement = await settleObservedExitForTest({ + journal: rig.journal, + sessionId: SESSION, + fence: 2, + ownerFence: 1, + settlementId: `observed-exit-${n}`, + pendingSubmissionReason: DISPATCH_DOUBT_PROVIDER_EXITED, + verdict: { state: 'interrupted', completedAt: 1200 }, + exit: { ownerFence: 1, observedAt: 1200 } + }) + } finally { + for (const spy of spies) { + spy.mockRestore() + } + rig.journal.observeCommits(() => {}) + } + }) + expect(settlement).toMatchObject({ ok: true }) + expect(batchCalls).toBe(1) + expect(counts.commits).toBe(1) + expect(counts.execution.map((pass) => pass.phase.replace(/.*at (\w+).*/, '$1'))).toEqual([ + 'captureUnfinishedStructuredAgentSessionWork', + 'planStructuredAgentSessionDeadGeneration', + 'planStaleStructuredAgentSessionState' + ]) + expect(counts.snapshots).toHaveLength(3) + for (const pass of counts.execution) { + expect(pass.construction).toBe(before.length) + expect(pass.predicates).toBe(before.length) + expect(pass.executionItemIds).toEqual([providerTurnItemId('live')]) + expect(pass.reciprocalReads).toBeLessThanOrEqual(2 * before.length) + } + expect(counts.snapshots.map((entry) => entry.rows)).toEqual([ + before.length, + before.length, + before.length + ]) + const after = await rig.rows() + expect(after.filter((row) => archivedById.has(row.itemId))).toEqual(archival) + expect( + readAgentJournalTurn(rig.journal.itemBody(providerTurnItemId('live')) ?? undefined) + ).toMatchObject({ state: 'interrupted', completedAt: 1200 }) + expect(rig.journal.submissions()).toMatchObject([ + { + clientMessageId: 'live', + dispatchState: 'unknown', + reason: DISPATCH_DOUBT_PROVIDER_EXITED + } + ]) + expect( + after.filter((row) => row.body.kind === 'status' && row.body.tone === 'error') + ).toHaveLength(1) + } +}, 120000) diff --git a/src/main/acp/acp-history-translation.ts b/src/main/acp/acp-history-translation.ts new file mode 100644 index 00000000000..c8698fa43e5 --- /dev/null +++ b/src/main/acp/acp-history-translation.ts @@ -0,0 +1,112 @@ +// A replayed turn written as the chat's lane writes a live one: the same translator and assembler, +// so its rows carry exactly the keys and bodies a live turn would have, into a staging overlay. + +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' +import type { AgentType } from '../../shared/agent-status-types' +import { createProviderTimelineAssembler } from '../native-chat/agent-session-timeline/provider-timeline-assembler' +import type { ProviderTimelineEvent } from '../native-chat/agent-session-timeline/provider-timeline-event' +import { agentJournalTurnBody, readAgentJournalTurn } from '../../shared/agent-session-turn-record' +import type { AcpDialect } from './acp-dialects/acp-dialect' +import { AcpHistoryJournalOverlay, type AcpHistoryItem } from './acp-history-journal-overlay' +import { + acpHistoryFrameTime, + type AcpHistoryTurnIdentity, + type AcpIdentifiedHistoryTurn +} from './acp-history-replay' +import { AcpTimelineTranslator } from './acp-timeline-translator' + +/** The chat a replay belongs to. */ +export type AcpHistoryScope = { + /** Orca's session id. */ + sessionId: string + agent: AgentType + providerSessionId: string + dialect: AcpDialect + /** How the agent names each turn it replays. */ + identity: AcpHistoryTurnIdentity +} + +class AcpHistoryTranslator extends AcpTimelineTranslator { + openHistoryTurn(turn: string, at: number): ProviderTimelineEvent[] { + return this.start(turn, at) + } +} + +/** Each turn gets its own translator and assembler, so its rows are keyed the same on every open. */ +export function translateHistoryTurn( + scope: AcpHistoryScope, + turn: AcpIdentifiedHistoryTurn, + check: () => void = () => {} +): AcpHistoryItem[] { + const overlay = new AcpHistoryJournalOverlay('history-snapshot') + const translator = new AcpHistoryTranslator({ + sessionId: scope.providerSessionId, + dialect: scope.dialect, + onTextDropped: () => { + throw new Error('history-text-dropped') + } + }) + const assembler = createProviderTimelineAssembler({ + sink: overlay.sink, + sessionId: scope.sessionId, + agent: scope.agent, + generation: `history:${turn.turn}`, + namespace: scope.providerSessionId, + schedule: () => () => {} + }) + const at = turn.startedAt ?? 0 + const apply = (events: readonly ProviderTimelineEvent[]) => { + for (const event of events) { + const result = assembler.apply(event) + if (!result.admission.accepted || (result.dropped && result.dropped !== 'turn-duplicate')) { + throw new Error('history-not-admitted') + } + } + } + try { + apply(translator.openHistoryTurn(turn.turn, at)) + for (const observation of [turn.user, ...turn.observations]) { + check() + const frame = 'blocks' in observation ? observation.boundary : observation + apply(translator.notification(frame.method, frame.params, acpHistoryFrameTime(frame) ?? at)) + if ('blocks' in observation) { + apply([ + { + type: 'item.update', + item: + observation === turn.user + ? `user:${turn.turn}` + : `user-message:${JSON.stringify([turn.turn, observation.eventId])}`, + body: { kind: 'message', role: 'user', blocks: observation.blocks }, + join: { thread: scope.providerSessionId, turn: turn.turn } + } + ]) + } + } + assembler.flush() + const created = overlay.created() + const opener = created.find((item) => item.body.kind === 'message' && item.body.role === 'user') + if (!opener) { + throw new Error('no-history-opener') + } + return created.map((mutation) => { + const lifecycle = readAgentJournalTurn(mutation.body) + if (!lifecycle) { + return mutation + } + const { startedAt, ...body } = lifecycle + return { + ...mutation, + body: agentJournalTurnBody({ + ...body, + userItemId: agentJournalItemKey(opener.identity), + ...(turn.startedAt === null ? {} : { startedAt }), + ...(lifecycle.state === 'running' ? { state: 'completed' as const } : {}) + }) + } + }) + } finally { + assembler.dispose() + translator.dispose() + } +} diff --git a/src/main/acp/acp-history-turn-matching.ts b/src/main/acp/acp-history-turn-matching.ts new file mode 100644 index 00000000000..55ee94b0400 --- /dev/null +++ b/src/main/acp/acp-history-turn-matching.ts @@ -0,0 +1,260 @@ +import { OrderedAvailability } from './acp-history-ordered-availability' +import { + agentJournalItemKey, + parseAgentJournalItemKey +} from '../../shared/agent-session-journal-item-key' +import type { + AgentJournalItemBody, + AgentJournalRenderItem +} from '../../shared/agent-session-journal-types' +import type { NativeChatBlock } from '../../shared/native-chat-types' +import { readAgentJournalTurn } from '../../shared/agent-session-turn-record' +import { isProviderTimelineTurnInNamespace } from '../native-chat/agent-session-timeline/provider-timeline-identity' +import { isHistoricalProviderTimelineItem } from '../native-chat/agent-session-timeline/provider-timeline-history' +import type { AcpHistoryItem } from './acp-history-journal-overlay' +import type { AcpIdentifiedHistoryTurn } from './acp-history-replay' +import type { AcpHistoryScope } from './acp-history-translation' + +export type AcpHistoryCandidate = { source: AcpIdentifiedHistoryTurn; items: AcpHistoryItem[] } +export type AcpHistoryReadJournal = { + items: ReadonlyMap +} +type NativeTurn = { itemId: string; sequence: number; text: string | null; output: boolean } + +export function acpHistoryHasOutput(body: AgentJournalItemBody): boolean { + return ( + body.kind !== 'turn' && + !(body.kind === 'message' && body.role === 'user') && + (body.kind !== 'message' || + body.blocks.some((block) => block.type !== 'text' || block.text.length > 0)) + ) +} + +export function acpNativeTurns( + scope: AcpHistoryScope, + journal: AcpHistoryReadJournal +): NativeTurn[] { + const turns: NativeTurn[] = [] + for (const item of journal.items.values()) { + const turn = readAgentJournalTurn(item.body) + const identity = parseAgentJournalItemKey(item.itemId) + if ( + turn?.userItemId && + parseAgentJournalItemKey(turn.userItemId)?.provider === 'orca' && + identity?.provider === 'legacy' && + identity.agent === scope.agent && + identity.sessionId === scope.sessionId && + identity.recordId === `turn-lifecycle:${turn.turnId}` && + isProviderTimelineTurnInNamespace(turn.turnId, scope.providerSessionId) + ) { + const user = journal.items.get(turn.userItemId)?.body + turns.push({ + itemId: item.itemId, + sequence: item.sequence, + output: false, + text: user?.kind === 'message' && !user.command ? promptText(user.blocks) : null + }) + } + } + const byId = new Map(turns.map((turn) => [turn.itemId, turn])) + for (const item of journal.items.values()) { + const owner = item.turnScope?.kind === 'turn' ? byId.get(item.turnScope.turnItemId) : undefined + if (owner && acpHistoryHasOutput(item.body)) { + owner.output = true + } + } + return turns.toSorted((a, b) => a.sequence - b.sequence) +} + +/** One source index supplies both exact coverage proof and the candidate bodies to reconcile. */ +export function matchAcpHistoryTurns( + scope: AcpHistoryScope, + journal: AcpHistoryReadJournal, + candidates: readonly AcpHistoryCandidate[], + importedAt: number | null, + requests: ReadonlySet, + check: () => void +) { + const natives = acpNativeTurns(scope, journal) + const nativeIndex = new Map(natives.map((native, index) => [native.itemId, index])) + const adopted = new Map() + for (const item of journal.items.values()) { + const identity = parseAgentJournalItemKey(item.itemId) + if ( + identity?.provider === 'legacy' && + identity.agent === scope.agent && + identity.sessionId === scope.providerSessionId + ) { + adopted.set(identity.recordId, item.sequence) + } + } + if (scope.identity === 'prompt-id' && adopted.size > 0 && importedAt === null) { + throw new Error('no-legacy-origin') + } + const kind = new Map() + const matched = new Map() + const boundaries = new Map() + const read = (key: string) => journal.items.get(key) ?? null + for (const { source, items } of candidates) { + check() + const owners = new Set() + let heldTurn: AgentJournalRenderItem | undefined + for (const candidate of items) { + check() + const ownKey = agentJournalItemKey(candidate.identity) + const held = read(ownKey) + if (candidate.body.kind === 'turn') { + heldTurn = held ?? undefined + const native = held && nativeIndex.get(held.itemId) + if (native !== undefined && native !== null) { + owners.add(native) + } + } else if (acpHistoryHasOutput(candidate.body) && held?.turnScope?.kind === 'turn') { + const native = nativeIndex.get(held.turnScope.turnItemId) + if (native !== undefined) { + owners.add(native) + } + } + } + const native = [...owners][0] + const historical = + heldTurn && isHistoricalProviderTimelineItem(heldTurn, read) ? heldTurn : undefined + if (owners.size > 1 || (historical && native !== undefined)) { + throw new Error('conflicting-history-owners') + } + if (native !== undefined || (scope.identity === 'prompt-id' && requests.has(source.turn))) { + kind.set(source.turn, 'native') + if (native !== undefined) { + matched.set(source.turn, native) + } + } else if (historical) { + kind.set(source.turn, 'imported') + boundaries.set(source.turn, historical.sequence) + } else if (heldTurn) { + throw new Error('unowned-history-turn') + } else { + const adoptedSequence = source.messageIds.flatMap((id) => + ['', 'opencode:'].flatMap((prefix) => + ['', ':reasoning', ':omission'].flatMap((suffix) => { + const sequence = adopted.get(`${prefix}${id}${suffix}`) + return sequence === undefined ? [] : [sequence] + }) + ) + ) + if ( + adoptedSequence.length || + (scope.identity === 'prompt-id' && + importedAt !== null && + source.startedAt !== null && + source.startedAt <= importedAt) + ) { + kind.set(source.turn, 'adopted') + if (scope.identity === 'user-message-id') { + let sequence = -Infinity + for (const value of adoptedSequence) { + check() + sequence = Math.max(sequence, value) + } + boundaries.set(source.turn, sequence) + } + } + } + } + if (scope.identity === 'user-message-id') { + const used = new Set(matched.values()) + for (const [id, sequence] of boundaries) { + let low = 0 + let high = natives.length + while (low < high) { + check() + const middle = Math.floor((low + high) / 2) + if (natives[middle].sequence < sequence) { + low = middle + 1 + } else { + high = middle + } + } + boundaries.set(id, low) + } + const occurrences = new Map() + const byText = new Map() + const positions: number[] = [] + for (const [index, { source }] of candidates.entries()) { + check() + const held = occurrences.get(source.turn) ?? [] + held.push(index) + occurrences.set(source.turn, held) + positions.push(index) + } + const anchors = new OrderedAvailability(positions, check, false) + for (const [index, { source }] of candidates.entries()) { + check() + if (matched.has(source.turn) || boundaries.has(source.turn)) { + anchors.set(index, true) + } + } + for (const [index, native] of natives.entries()) { + check() + if (!native.output && native.text !== null && !used.has(index)) { + const ranks = byText.get(native.text) ?? [] + ranks.push(index) + byText.set(native.text, ranks) + } + } + const available = new Map() + for (const [text, ranks] of byText) { + check() + available.set(text, new OrderedAvailability(ranks, check)) + } + let floor = -1 + for (const [index, candidate] of candidates.entries()) { + check() + const id = candidate.source.turn + const exact = matched.get(id) + const edge = boundaries.get(id) + if (exact !== undefined) { + floor = exact + continue + } + if (edge !== undefined) { + floor = edge - 1 + continue + } + if (kind.has(id) || candidate.items.some((item) => acpHistoryHasOutput(item.body))) { + continue + } + const text = candidate.source.textOnly ? promptText(candidate.source.user.blocks) : null + if (text === null) { + continue + } + const nextAnchor = anchors.after(index) + const nextId = nextAnchor && candidates[nextAnchor.value]?.source.turn + const ceiling = + nextId === undefined + ? natives.length + : (matched.get(nextId) ?? boundaries.get(nextId) ?? natives.length) + const group = available.get(text) + const next = group?.after(floor) + if (next && next.value < ceiling) { + kind.set(id, 'native') + matched.set(id, next.value) + used.add(next.value) + floor = next.value + group?.set(next.index, false) + for (const position of occurrences.get(id) ?? []) { + check() + if (position > index) { + anchors.set(position, true) + } + } + } + } + } + return { natives, kind, matched } +} + +function promptText(blocks: readonly NativeChatBlock[]): string | null { + return blocks.every((block) => block.type === 'text') + ? blocks.flatMap((block) => (block.type === 'text' ? [block.text] : [])).join('\n') + : null +} diff --git a/src/main/acp/acp-launch-specs.ts b/src/main/acp/acp-launch-specs.ts index 281751cbd00..bce0f200471 100644 --- a/src/main/acp/acp-launch-specs.ts +++ b/src/main/acp/acp-launch-specs.ts @@ -1,3 +1,4 @@ +import type { AcpHistoryTurnIdentity } from './acp-history-replay' import { join } from 'node:path' import type { AcpDialect } from './acp-dialects/acp-dialect' import { GROK_ACP_DIALECT } from './acp-dialects/grok-dialect' @@ -125,6 +126,9 @@ export type AcpLaunchSpec = { compaction?: true /** Rejects copy arguments that route the CLI away from the copy's store. */ assertCopyLaunchArgs?: (args: readonly string[]) => void + /** How the agent names each turn its reopen replays, when a reopened chat catches up on the + * turns made outside it; absent, the replay is only dropped. */ + historyCatchUp?: AcpHistoryTurnIdentity modelDiscovery: AcpModelDiscovery visualsSkill?: AcpVisualsSkillLoader transcriptImport: StructuredAgentTranscriptImport | null @@ -147,6 +151,7 @@ const GROK_LAUNCH_SPEC: AcpLaunchSpec = { ], env: {}, dialect: GROK_ACP_DIALECT, + historyCatchUp: 'prompt-id', loginCommand: agentSessionSignInFor('grok')?.loginCommand ?? [], // Prefer the launch environment's API key, then cached sign-in; never prompt interactively. authMethod: ({ advertised, env }) => @@ -184,6 +189,7 @@ function openCodeLaunchSpec( env: { OPENCODE_CLIENT: 'acp', OPENCODE_ENABLE_QUESTION_TOOL: 'false' }, scrubEnvironment: scrubOpenCodeAcpEnvironment, dialect: OPENCODE_ACP_DIALECT, + historyCatchUp: 'user-message-id', loginCommand: agentSessionSignInFor(row.agent)?.loginCommand ?? [], account: openCodeAcpAccountBinding(), installDirectories: ({ homePath }) => [join(homePath, '.opencode', 'bin')], diff --git a/src/main/acp/acp-structured-acquire.ts b/src/main/acp/acp-structured-acquire.ts index 198de112f0b..c48b59dbd08 100644 --- a/src/main/acp/acp-structured-acquire.ts +++ b/src/main/acp/acp-structured-acquire.ts @@ -22,6 +22,7 @@ import { NO_ACP_CHILD_STOPS } from './acp-structured-child-stop' import { prepareAcpStructuredLaunch } from './acp-structured-launch-preparation' import { AcpStructuredOptions } from './acp-structured-options' import { AcpStructuredPrompts } from './acp-structured-prompts' +import { AcpLoadHistory } from './acp-history-replay' import { asReattachHistory, routeAcpSessionEvent } from './acp-structured-session' import type { AcpStructuredSessionAdapterDeps } from './acp-structured-session-adapter-deps' import type { AcpStructuredTurnsDeps } from './acp-structured-turns' @@ -49,6 +50,7 @@ export async function acquireAcpStructuredSession(input: { const { acquire, deps, generation } = input const { spec } = deps const agentName = acpAgentName(spec.agent) + const logger = deps.logger ?? createStructuredAgentSessionLogger() const sessionId = acquire.identity.sessionId const now = deps.now ?? Date.now const events = acquire.events @@ -65,10 +67,11 @@ export async function acquireAcpStructuredSession(input: { const launch = await prepareAcpStructuredLaunch(deps, acquire, closedBeforeSpawn) let session: AcpStructuredChild | null = null // Use a slot so closure writes do not conflict with control-flow narrowing. - const slot: { lane: AcpStructuredLane | null; reattaching: boolean } = { - lane: null, - reattaching: false - } + const slot: { + lane: AcpStructuredLane | null + reattaching: boolean + history: AcpLoadHistory | null + } = { lane: null, reattaching: false, history: null } const options = new AcpStructuredOptions(launch.spec.dialect) // Agent-initiated turns cannot open interactive requests; settled request content may still show. const prompts = new AcpStructuredPrompts( @@ -121,7 +124,10 @@ export async function acquireAcpStructuredSession(input: { }, onRequest: (method, params, context) => prompts.handle(method, params, context), onExtensionNotification: (method, params) => - whenLane(() => + whenLane(() => { + if (slot.reattaching) { + slot.history?.keep(method, params) + } slot.lane?.apply( slot.lane.translator.notification( method, @@ -129,7 +135,7 @@ export async function acquireAcpStructuredSession(input: { now() ) ) - ), + }), onDiagnostic: (message) => deps.logger?.warn('ACP agent protocol diagnostic', { scope: 'acp-diagnostic', @@ -147,6 +153,12 @@ export async function acquireAcpStructuredSession(input: { connection.subscribe((event: AcpSessionEvent) => whenLane(() => { if (slot.lane) { + if (slot.reattaching) { + slot.history?.keep( + 'session/update', + event.kind === 'known' ? event.notification : event.raw + ) + } routeAcpSessionEvent({ lane: slot.lane, options }, event, now(), slot.reattaching) } if ( @@ -183,7 +195,7 @@ export async function acquireAcpStructuredSession(input: { now, childStops: () => session?.childStops ?? NO_ACP_CHILD_STOPS, ...acpLaneChildWorkDelivery(deps, sessionId, events), - logger: deps.logger ?? createStructuredAgentSessionLogger(), + logger, onInputAccepted: (clientMessageId) => { if (isAcpReadySession(session)) { session.turns?.accept(clientMessageId) @@ -193,6 +205,17 @@ export async function acquireAcpStructuredSession(input: { }) slot.lane = lane slot.reattaching = attaching + // What the agent replays is kept for catch-up, never written as it arrives. + slot.history = + attaching && spec.historyCatchUp + ? new AcpLoadHistory(providerSessionId, () => + logger.warn(`${agentName} chat did not catch up on turns made outside it`, { + scope: 'acp-history-catch-up', + sessionId, + reason: 'replay-too-large' + }) + ) + : null if (attaching) { lane.translator.beginLoad() } @@ -241,9 +264,13 @@ export async function acquireAcpStructuredSession(input: { create: makeLane, clear: () => { slot.lane = null + slot.history = null }, finishLoad: () => { slot.reattaching = false + const replayed = slot.history?.frames() ?? null + slot.history = null + return replayed } }, onSettled: input.onSettled, diff --git a/src/main/acp/acp-structured-handshake.ts b/src/main/acp/acp-structured-handshake.ts index 5b80495e13f..d71e92cac93 100644 --- a/src/main/acp/acp-structured-handshake.ts +++ b/src/main/acp/acp-structured-handshake.ts @@ -1,4 +1,8 @@ -// Suppress replayed history during reattach; preserve context usage. +// An ACP child's handshake, run after the host has published it: initialize, reopen the chat's +// session or start a new one, then return its protocol facts to the host. A saved session the +// agent cannot reopen is replaced by a new one, with a warning row that any later start writes if +// this one never did; while a reopened session replays its history, nothing it sends is written +// except context usage, and afterwards the turns of it the chat never saw are added once. import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle' import type { StructuredAgentSessionAcquireInput } from '../native-chat/agent-session-wire/structured-agent-session-adapter' @@ -8,6 +12,8 @@ import { acpReopenTakeover, acpSessionNotRestoredRow } from './acp-session-reopen-failure' +import { catchUpAcpHistory } from './acp-history-catch-up' +import type { AcpLoadHistoryFrame } from './acp-history-replay' import { ACP_HANDLE_TRANSPORT, acpAgentName } from './acp-structured-agent-definitions' import { probeAcpChildStops } from './acp-structured-child-stop' import type { AcpStructuredConnection } from './acp-structured-connection' @@ -36,7 +42,8 @@ type AcpHandshakeInput = { lanes: { create: (providerSessionId: string, attaching?: boolean) => AcpStructuredLane clear: () => void - finishLoad: () => void + /** The attach window is over: what the agent replayed in it, when it was kept. */ + finishLoad: () => AcpLoadHistoryFrame[] | null } onSettled: AcpStructuredTurnsDeps['settle'] onSessionLost: AcpStructuredTurnsDeps['sessionLost'] @@ -64,6 +71,7 @@ export async function initializeAcpStructuredSession( let started: Awaited> | null = null let liveLane: AcpStructuredLane | null = null let takeover: ReturnType | null = null + let replayed: AcpLoadHistoryFrame[] | null = null if (resume) { const attaching = input.lanes.create(resume.sessionId, true) liveLane = attaching @@ -75,7 +83,7 @@ export async function initializeAcpStructuredSession( ...setup }) assertStarting() - input.lanes.finishLoad() + replayed = input.lanes.finishLoad() attaching.translator.finishLoad() } catch (error) { assertStarting() @@ -93,6 +101,29 @@ export async function initializeAcpStructuredSession( assertStarting() liveLane = input.lanes.create(started.sessionId) } + // Only a load that reopened the chat's own conversation catches up, and only from what that + // conversation replayed; a replacement session has no history of the chat's to add. + if ( + replayed && + spec.historyCatchUp && + !takeover && + started.kind !== 'new' && + started.sessionId === resume?.sessionId && + acquire.events + ) { + await catchUpAcpHistory({ + frames: replayed, + identity: spec.historyCatchUp, + sink: acquire.events, + sessionId, + agent: spec.agent, + providerSessionId: started.sessionId, + dialect: spec.dialect, + assertCurrent: assertStarting, + ...(deps.logger ? { logger: deps.logger } : {}) + }) + assertStarting() + } const childStops = await probeAcpChildStops(connection, spec.dialect) assertStarting() options.adoptSession(started.response, started.kind === 'new' ? 'new' : 'loaded') diff --git a/src/main/acp/acp-structured-session-adapter-resume.test.ts b/src/main/acp/acp-structured-session-adapter-resume.test.ts index b35d3b5d7f2..cb7dd89c0ca 100644 --- a/src/main/acp/acp-structured-session-adapter-resume.test.ts +++ b/src/main/acp/acp-structured-session-adapter-resume.test.ts @@ -64,6 +64,17 @@ function sendsWhileAttaching(agent: AcpScriptedAgent, reply: string): void { }) } +/** Grok saved the exchange only once the chat sent it: the first open has nothing to replay. */ +function onReopen(): (replay: () => void) => void { + let loads = 0 + return (replay) => { + if (loads++ > 0) { + replay() + } + } +} + +/** Orca's send of `hello` as `m1` and Grok's reply, ended or cut off. */ async function exchange(rig: AcpAdapterRig, reply: string, end: boolean): Promise { await rig.rig.eventSink.appendItem({ provider: 'orca', clientMessageId: 'm1' }, hello, { turnScope: { kind: 'thread' } @@ -211,12 +222,13 @@ it("starts in the chat's own permission mode, asks when none is named, and fails describe('reattaching a Grok chat the journal holds', () => { it('loads with session/load though Grok also resumes, and writes nothing it replays', async () => { + const reopen = onReopen() const rig = await openAcpAdapterRig({ ...resume, initialize: RESUMES, script: (agent) => agent.on('session/load', (frame) => { - sendsWhileAttaching(agent, 'hi') + reopen(() => sendsWhileAttaching(agent, 'hi')) agent.reply(frame, { configOptions: GROK_CONFIG_OPTIONS }) }) }) @@ -237,11 +249,12 @@ describe('reattaching a Grok chat the journal holds', () => { }) it('loads an agent that only loads, writing none of what it replays', async () => { + const reopen = onReopen() const rig = await openAcpAdapterRig({ ...resume, script: (agent) => agent.on('session/load', (frame) => { - sendsWhileAttaching(agent, 'hi') + reopen(() => sendsWhileAttaching(agent, 'hi')) agent.reply(frame, { configOptions: GROK_CONFIG_OPTIONS }) }) }) @@ -257,11 +270,12 @@ describe('reattaching a Grok chat the journal holds', () => { }) it('leaves a reply cut off mid-turn as it was cut, never completed from what Grok saved', async () => { + const reopen = onReopen() const rig = await openAcpAdapterRig({ ...resume, script: (agent) => agent.on('session/load', (frame) => { - sendsWhileAttaching(agent, 'complete saved reply') + reopen(() => sendsWhileAttaching(agent, 'complete saved reply')) agent.reply(frame, { configOptions: GROK_CONFIG_OPTIONS }) }) }) @@ -277,13 +291,14 @@ describe('reattaching a Grok chat the journal holds', () => { ]) }) - it('writes nothing of the reattach into a sink the host binds only after the start', async () => { + it('writes no replay rows when the host binds the sink between acquire and open', async () => { + const reopen = onReopen() const rig = await openAcpAdapterRig({ ...resume, initialize: RESUMES, script: (agent) => agent.on('session/load', (frame) => { - sendsWhileAttaching(agent, 'hi') + reopen(() => sendsWhileAttaching(agent, 'hi')) agent.reply(frame, { configOptions: GROK_CONFIG_OPTIONS }) }) }) @@ -291,7 +306,7 @@ describe('reattaching a Grok chat the journal holds', () => { await exchange(rig, 'hi', true) await rig.adapter.closeSession(SESSION) const before = await rig.rig.rows() - // Bind a fresh acquisition sink only after startup succeeds. + // The host binds after acquiring the child, before opening its provider session. const deferred = createDeferredStructuredAgentSessionEventSink(testEventSinkLogging(SESSION)) try { const acquired = await rig.adapter.acquire({ @@ -306,6 +321,7 @@ describe('reattaching a Grok chat the journal holds', () => { spawnToken: 'spawn-2', events: deferred.sink }) + deferred.bind({ journal: rig.rig.journal, fence: 2, publish: () => {} }) await rig.adapter.open({ sessionId: SESSION, fence: 2, @@ -313,14 +329,13 @@ describe('reattaching a Grok chat the journal holds', () => { signal: new AbortController().signal }) expect(rig.sent('session/load')).toHaveLength(1) - deferred.bind({ journal: rig.rig.journal, fence: 2, publish: () => {} }) expect(await deferred.drained()).toMatchObject({ ok: true }) expect((await rig.rig.rows()).map((row) => row.itemId)).toEqual( before.map((row) => row.itemId) ) - await rig.adapter.closeSession(SESSION) } finally { deferred.close() + await rig.adapter.closeSession(SESSION) } }) diff --git a/src/main/acp/acp-timeline-dialect.test.ts b/src/main/acp/acp-timeline-dialect.test.ts index db7d632835d..8be8071cb9c 100644 --- a/src/main/acp/acp-timeline-dialect.test.ts +++ b/src/main/acp/acp-timeline-dialect.test.ts @@ -39,6 +39,32 @@ const question = { } describe('ACP dialect request and turn boundaries', () => { + it('separates anonymous output across a suppressed user echo within one prompt', async () => { + const { rig, translator, apply, notification } = await dialectRig() + apply(translator.openPrompt('send-1', 1000).events) + for (const [sessionUpdate, text] of [ + ['user_message_chunk', 'root echo'], + ['agent_message_chunk', 'A'], + ['user_message_chunk', 'followup echo'], + ['agent_message_chunk', 'B'] + ]) { + notification('session/update', { + sessionId: 'provider-1', + _meta: { promptId: 'prompt:send-1' }, + update: { sessionUpdate, content: { type: 'text', text } } + }) + } + apply(translator.promptResult('send-1', { stopReason: 'end_turn' }, 1200)) + const messages = (await rig.rows()).flatMap((row) => + row.body.kind === 'message' ? [{ role: row.body.role, text: messageText(row.body) }] : [] + ) + expect(messages).toEqual([ + { role: 'assistant', text: 'A' }, + { role: 'assistant', text: 'B' } + ]) + expect(await rig.turns()).toHaveLength(1) + }) + it('preserves provider permission labels, validates selected ids, and cancels unanswered rows on turn end', async () => { const { rig, translator, apply } = await dialectRig() apply(translator.openPrompt('send-1', 1000).events) diff --git a/src/main/acp/acp-timeline-translator.ts b/src/main/acp/acp-timeline-translator.ts index 7937098c53b..1a364dfa665 100644 --- a/src/main/acp/acp-timeline-translator.ts +++ b/src/main/acp/acp-timeline-translator.ts @@ -48,7 +48,7 @@ export class AcpTimelineTranslator { private readonly failures: AcpTurnFailures private loading = false private readonly context = new AcpContextTimeline() - private activeTurn?: string + protected activeTurn?: string constructor(private readonly options: AcpTimelineTranslatorOptions) { this.dialect = options.dialect ?? GENERIC_ACP_DIALECT @@ -310,7 +310,7 @@ export class AcpTimelineTranslator { return events } - private start(turn: string, at: number): ProviderTimelineEvent[] { + protected start(turn: string, at: number): ProviderTimelineEvent[] { if (turn === this.prompts.current?.turn) { const events = this.prompts.start(turn, at) // Late frames must not reopen or retarget this prompt. diff --git a/src/main/acp/acp-turn-messages.ts b/src/main/acp/acp-turn-messages.ts index ec101af2c5e..28b166a4c2b 100644 --- a/src/main/acp/acp-turn-messages.ts +++ b/src/main/acp/acp-turn-messages.ts @@ -51,7 +51,11 @@ export class AcpTurnMessages { update.sessionUpdate !== 'agent_message_chunk' && update.sessionUpdate !== 'agent_thought_chunk' ) { - if (['tool_call', 'tool_call_update', 'plan'].includes(update.sessionUpdate)) { + if ( + ['user_message_chunk', 'tool_call', 'tool_call_update', 'plan'].includes( + update.sessionUpdate + ) + ) { state.channel = undefined } return undefined diff --git a/src/main/acp/fixtures/grok-v1-reopen-outside.jsonl b/src/main/acp/fixtures/grok-v1-reopen-outside.jsonl new file mode 100644 index 00000000000..66829c4a62b --- /dev/null +++ b/src/main/acp/fixtures/grok-v1-reopen-outside.jsonl @@ -0,0 +1,12 @@ +{"direction":"out","process":"A","message":{"jsonrpc":"2.0","id":3,"method":"session/prompt","params":{"sessionId":"session-1","prompt":[{"type":"text","text":"Reply with exactly the word ONE."}],"_meta":{"promptId":"prompt:A:3"}}}} +{"direction":"in","process":"A","message":{"jsonrpc":"2.0","method":"_x.ai/session_notification","params":{"sessionId":"session-1","update":{"sessionUpdate":"retry_state","type":"failed","error_type":"api","message":"API error (status 402 Payment Required): Grok Build usage balance exhausted"},"_meta":{"eventId":"session-1-4","agentTimestampMs":1792606352991}}}} +{"direction":"in","process":"A","message":{"jsonrpc":"2.0","method":"_x.ai/session_notification","params":{"sessionId":"session-1","update":{"sessionUpdate":"turn_completed","prompt_id":"prompt:A:3","stop_reason":"error","agent_result":"API error (status 402 Payment Required): Grok Build usage balance exhausted","elapsed_ms":455},"_meta":{"eventId":"session-1-5","agentTimestampMs":1792606353077}}}} +{"direction":"in","process":"A","message":{"jsonrpc":"2.0","id":3,"error":{"code":-32603,"message":"Internal error","data":{"message":"API error (status 402 Payment Required): Grok Build usage balance exhausted","http_status":402}}}} +{"direction":"out","process":"C","message":{"jsonrpc":"2.0","id":2,"method":"session/load","params":{"sessionId":"session-1","cwd":"/workspace","mcpServers":[]}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"session-1","update":{"sessionUpdate":"user_message_chunk","content":{"type":"text","text":"Reply with exactly the word ONE."},"_meta":{"modelId":"grok-4.7","promptIndex":0}},"_meta":{"eventId":"session-1-2","agentTimestampMs":1792606352634,"isReplay":true}}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","method":"_x.ai/session/update","params":{"sessionId":"session-1","update":{"sessionUpdate":"retry_state","type":"failed","error_type":"api","message":"API error (status 402 Payment Required): Grok Build usage balance exhausted"},"_meta":{"eventId":"session-1-4","agentTimestampMs":1792606352991,"isReplay":true}}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","method":"_x.ai/session/update","params":{"sessionId":"session-1","update":{"sessionUpdate":"turn_completed","prompt_id":"prompt:A:3","stop_reason":"error","agent_result":"API error (status 402 Payment Required): Grok Build usage balance exhausted","elapsed_ms":455},"_meta":{"eventId":"session-1-5","agentTimestampMs":1792606353077,"isReplay":true}}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"session-1","update":{"sessionUpdate":"user_message_chunk","content":{"type":"text","text":"Run the shell command: echo TWO-TOOL. Then reply with exactly the word TWO."},"_meta":{"modelId":"grok-4.7","promptIndex":1}},"_meta":{"eventId":"session-1-10","agentTimestampMs":1792606356454,"isReplay":true}}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","method":"_x.ai/session/update","params":{"sessionId":"session-1","update":{"sessionUpdate":"retry_state","type":"failed","error_type":"api","message":"API error (status 402 Payment Required): Grok Build usage balance exhausted"},"_meta":{"eventId":"session-1-12","agentTimestampMs":1792606356711,"isReplay":true}}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","method":"_x.ai/session/update","params":{"sessionId":"session-1","update":{"sessionUpdate":"turn_completed","prompt_id":"afd209ad-e5c3-431d-9e84-9c2a775b3bd2","stop_reason":"error","agent_result":"API error (status 402 Payment Required): Grok Build usage balance exhausted","elapsed_ms":320},"_meta":{"eventId":"session-1-13","agentTimestampMs":1792606356766,"isReplay":true}}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","id":2,"result":{}}} diff --git a/src/main/acp/fixtures/opencode-v2-reopen-outside.jsonl b/src/main/acp/fixtures/opencode-v2-reopen-outside.jsonl new file mode 100644 index 00000000000..adc5d2be86a --- /dev/null +++ b/src/main/acp/fixtures/opencode-v2-reopen-outside.jsonl @@ -0,0 +1,6 @@ +{"direction":"out","process":"A","message":{"jsonrpc":"2.0","id":3,"method":"session/prompt","params":{"sessionId":"session-1","prompt":[{"type":"text","text":"Reply with exactly the word ONE."}]}}} +{"direction":"in","process":"A","message":{"jsonrpc":"2.0","id":3,"error":{"code":-32603,"message":"Internal error: Rate limit exceeded. Please try again later.","data":{"service":"session","errorName":"provider.quota"}}}} +{"direction":"out","process":"C","message":{"jsonrpc":"2.0","id":2,"method":"session/load","params":{"sessionId":"session-1","cwd":"/workspace","mcpServers":[]}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"session-1","update":{"sessionUpdate":"user_message_chunk","messageId":"msg_15fad6c9f001d4ix8WtRMUSWtI","content":{"type":"text","text":"Reply with exactly the word ONE."}}}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"session-1","update":{"sessionUpdate":"user_message_chunk","messageId":"msg_15fad7241001viLg4AlvQLn3N6","content":{"type":"text","text":"Run the shell command: echo TWO-TOOL. Then reply with exactly the word TWO."}}}}} +{"direction":"in","process":"C","message":{"jsonrpc":"2.0","id":2,"result":{}}} diff --git a/src/main/agent-hooks/first-work-branch-rename.test.ts b/src/main/agent-hooks/first-work-branch-rename.test.ts index 54b9b919d96..c7e5fb813a4 100644 --- a/src/main/agent-hooks/first-work-branch-rename.test.ts +++ b/src/main/agent-hooks/first-work-branch-rename.test.ts @@ -112,9 +112,12 @@ describe('maybeAutoRenameBranchOnFirstWork', () => { const items: AgentJournalRenderItem[] = [] // Advance journal sequence so projection caching cannot hide a rename event. let sequence = 0 - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the status feed reads only cursor(), lastActivityAt(), stopMarks, snapshot(), the provenance reads and runningTurn() of a journal. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the status feed uses these held rows and provenance reads; the item lookup is checked against the real journal contract. const journal = { snapshot: () => ({ items, submissions: [] }), + item: ((itemId) => + items.find((item) => item.itemId === itemId) ?? + null) satisfies AgentSessionJournal['item'], stopMarks: { latest: () => null, revision: () => 0 }, lastActivityAt: () => 1, cursor: () => ({ epoch: 1, sequence: (sequence += 1) }), @@ -205,24 +208,34 @@ describe('maybeAutoRenameBranchOnFirstWork', () => { const { deps, setDisplayName, setRenameError } = makeDeps({ getRepo: () => ({ id: REPO_ID, kind: 'folder', path: '/workspace/platform' }) as Repo }) - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the status feed reads only cursor(), lastActivityAt(), stopMarks, snapshot(), the provenance reads and runningTurn() of a journal. + const items: AgentJournalRenderItem[] = [ + { + itemId: 'user-1', + sequence: 1, + revision: 1, + observedAt: 1, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'Fix auth' }] } + }, + { + itemId: 'turn-1', + sequence: 2, + revision: 1, + observedAt: 1, + body: { + kind: 'status', + text: 'Working', + turnLifecycle: { turnId: 'turn-1', state: 'running' } + } + } + ] + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the status feed uses these held rows and provenance reads; the item lookup is checked against the real journal contract. const journal = { lastActivityAt: () => 1, stopMarks: { latest: () => null, revision: () => 0 }, cursor: () => ({ epoch: 1, sequence: 1 }), - snapshot: () => ({ - items: [ - { body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'Fix auth' }] } }, - { - body: { - kind: 'status', - text: 'Working', - turnLifecycle: { turnId: 'turn-1', state: 'running' } - } - } - ], - submissions: [] - }), + snapshot: () => ({ items, submissions: [] }), + item: ((itemId) => + items.find((item) => item.itemId === itemId) ?? null) satisfies AgentSessionJournal['item'], itemFence: () => undefined, wroteBeforeOpen: () => false, runningTurn: () => null diff --git a/src/main/native-chat/agent-session-timeline/provider-timeline-history.ts b/src/main/native-chat/agent-session-timeline/provider-timeline-history.ts new file mode 100644 index 00000000000..828c8c06dca --- /dev/null +++ b/src/main/native-chat/agent-session-timeline/provider-timeline-history.ts @@ -0,0 +1,54 @@ +import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' + +type ItemReader = (itemId: string) => AgentJournalRenderItem | null + +/** Only replay produces this reciprocal generated-user/turn relation; live user echoes are ignored. */ +export function isHistoricalProviderTimelineItem( + item: AgentJournalRenderItem, + read: ItemReader +): boolean { + const turnItem = readAgentJournalTurn(item.body) + ? item + : item.turnScope?.kind === 'turn' + ? read(item.turnScope.turnItemId) + : null + const turn = readAgentJournalTurn(turnItem?.body) + const identity = turnItem && parseAgentJournalItemKey(turnItem.itemId) + if ( + !turnItem || + !turn?.userItemId || + turn.userItemId === turnItem.itemId || + identity?.provider !== 'legacy' || + !['grok', 'opencode', 'opencode2'].includes(identity.agent) || + identity.recordId !== `turn-lifecycle:${turn.turnId}` + ) { + return false + } + const namespace = /^p:([^:]+):[^:]+$/.exec(turn.turnId)?.[1] + const opener = read(turn.userItemId) + const source = opener && parseAgentJournalItemKey(opener.itemId) + return ( + namespace !== undefined && + opener?.itemId === turn.userItemId && + opener.body.kind === 'message' && + opener.body.role === 'user' && + opener.turnScope?.kind === 'turn' && + opener.turnScope.turnItemId === turnItem.itemId && + source?.provider === 'legacy' && + source.agent === identity.agent && + source.sessionId === identity.sessionId && + source.recordId.startsWith(`item:p:${namespace}:${namespace}/user%3A`) && + source.recordId.length > `item:p:${namespace}:${namespace}/user%3A`.length + ) +} + +export function providerTimelineExecutionItems( + items: readonly AgentJournalRenderItem[] +): AgentJournalRenderItem[] { + const byId = new Map(items.map((item) => [item.itemId, item])) + return items.filter( + (item) => !isHistoricalProviderTimelineItem(item, (key) => byId.get(key) ?? null) + ) +} diff --git a/src/main/native-chat/agent-session-timeline/provider-timeline-settlement.ts b/src/main/native-chat/agent-session-timeline/provider-timeline-settlement.ts index 537883415c1..9ded05c364e 100644 --- a/src/main/native-chat/agent-session-timeline/provider-timeline-settlement.ts +++ b/src/main/native-chat/agent-session-timeline/provider-timeline-settlement.ts @@ -1,3 +1,4 @@ +import { isHistoricalProviderTimelineItem } from './provider-timeline-history' // Resolve unsettled work from journal bodies, using the same terminal forms as dead-generation recovery. // Preserve client answers and other writers’ settlements; repeated settlement writes nothing. @@ -76,6 +77,10 @@ export function providerTimelineSettlement( const mutations: JournalLifecycleMutationInput[] = [] journal.visitItemsWithLinkage((itemId, _sequence, body, attribution) => { const turnScope = attribution.turnScope ?? AGENT_JOURNAL_THREAD_SCOPE + const historical = scope === 'session' && journal.item(itemId) + if (historical && isHistoricalProviderTimelineItem(historical, (key) => journal.item(key))) { + return + } const covered = scope === 'session' || (turnScope.kind === 'turn' && turnScope.turnItemId === scope.turnItemId) @@ -110,7 +115,8 @@ export function runningProviderTimelineTurns( journal.visitItems((itemId, _sequence, body) => { const identity = readAgentJournalTurn(body)?.state === 'running' ? parseAgentJournalItemKey(itemId) : null - if (identity) { + const item = journal.item(itemId) + if (identity && item && !isHistoricalProviderTimelineItem(item, (key) => journal.item(key))) { running.push({ identity, itemId }) } }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-current-work.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-current-work.ts index b60fb66aa11..7ccbd6d5497 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-current-work.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-current-work.ts @@ -1,3 +1,4 @@ +import { isHistoricalProviderTimelineItem } from '../agent-session-timeline/provider-timeline-history' import { currentRequestsProjection } from './agent-session-current-requests' // Derive operational work from lease, first-hand exit, replaced-runtime and journal evidence. // Cleanup writes never gate current work; readers share this projection rather than maintaining rules. @@ -178,6 +179,10 @@ export class StructuredAgentSessionCurrentWork { if (this.liveFence === null) { return false } + const item = this.journal.item(itemId) + if (item && isHistoricalProviderTimelineItem(item, (key) => this.journal.item(key))) { + return false + } const provenance = this.journal.itemFence(itemId) return provenance === undefined || provenance >= this.liveFence } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts index ae31f716d71..a94f3006e56 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts @@ -1,3 +1,4 @@ +import { providerTimelineExecutionItems } from '../agent-session-timeline/provider-timeline-history' import { agentSessionFailureFact, MAX_PROVIDER_DIAGNOSTIC_CHARS, @@ -94,7 +95,7 @@ export function planStructuredAgentSessionDeadGeneration( unrun ? { rejection: unrun } : { reason: input.pendingSubmissionReason } ) ] - const allItems = input.journal.snapshot().items + const allItems = providerTimelineExecutionItems(input.journal.snapshot().items) const items = allItems.filter((item) => owns(input.journal.itemFence(item.itemId))) const proven = watchedExitRevisions(items, input.exit, input.journal) const mutations: JournalLifecycleMutationInput[] = [] diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-test-harness.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-test-harness.ts index ca1a01a44c2..78313b7dcc0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-test-harness.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-resume-test-harness.ts @@ -6,6 +6,8 @@ import type { import type { AgentSessionRecord } from '../../../shared/agent-session-record' import type { AgentSessionResumeMarker } from '../../../shared/agent-session-resume-marker' import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' +import { runningStructuredAgentSessionTurnItemBySequence } from '../../../shared/structured-agent-session-live-turn' +import type { StructuredAgentSessionCurrentWorkJournal } from './structured-agent-session-current-work' import { structuredAgentSessionResumableSet } from './structured-agent-session-restart-resume-set' import { claudeProviderHandle, @@ -143,13 +145,24 @@ export function journal( items: AgentJournalRenderItem[], submissions: AgentJournalSubmission[] = [] ) { - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the code under test calls only snapshot(), submissions(), appendItem() and the provenance reads; a real AgentSessionJournal needs an on-disk SQLite store. - return { - snapshot: () => ({ items, submissions, cursor: { epoch: EPOCH, sequence: items.length } }), + const currentWork: StructuredAgentSessionCurrentWorkJournal = { + item: (itemId) => items.find((item) => item.itemId === itemId) ?? null, + cursor: () => ({ epoch: EPOCH, sequence: items.length }), + visitItems: (visit) => { + for (const item of items) { + visit(item.itemId, item.sequence, item.body) + } + }, + runningTurn: () => runningStructuredAgentSessionTurnItemBySequence(items), submissions: () => submissions, - appendItem: async () => undefined, itemFence: () => undefined, wroteBeforeOpen: () => false + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: current-work reads are checked above; the remaining exercised methods use the held fixture, without an on-disk SQLite store. + return { + ...currentWork, + snapshot: () => ({ items, submissions, cursor: { epoch: EPOCH, sequence: items.length } }), + appendItem: async () => undefined } as never } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-state-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-state-settlement.ts index 60d7f055040..a10d8bbad16 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-state-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-state-settlement.ts @@ -1,3 +1,4 @@ +import { providerTimelineExecutionItems } from '../agent-session-timeline/provider-timeline-history' // Plan against queued journal state; death/replacement proof interrupts only matching owners, // otherwise settle unverifiable and revise when later proof arrives. @@ -87,7 +88,9 @@ export function planStaleStructuredAgentSessionState( const { journal, below } = input const ended = (fence: number | undefined) => below === undefined || (fence !== undefined && fence < below) - const items = journal.snapshot().items.filter((item) => ended(journal.itemFence(item.itemId))) + const items = providerTimelineExecutionItems(journal.snapshot().items).filter((item) => + ended(journal.itemFence(item.itemId)) + ) const dispatches = below === undefined ? [] diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unfinished-work.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unfinished-work.ts index e87c8403bee..782be0853d9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unfinished-work.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unfinished-work.ts @@ -1,3 +1,4 @@ +import { providerTimelineExecutionItems } from '../agent-session-timeline/provider-timeline-history' import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import { requiresTerminalSettlement } from '../agent-session-journal/journal-terminal-settlement' @@ -24,7 +25,7 @@ export function captureUnfinishedStructuredAgentSessionWork( journal: DeadGenerationJournal ): StructuredAgentSessionUnfinishedWork { return { - items: journal.snapshot().items.filter(isUnfinishedItem), + items: providerTimelineExecutionItems(journal.snapshot().items).filter(isUnfinishedItem), hadUnsettledSubmissions: hasUnsettledSubmission(journal) } } @@ -37,7 +38,8 @@ export function hasUnfinishedStructuredAgentSessionWork( return ( work.hadUnsettledSubmissions || work.items.length > 0 || - watchedExitRevisions(journal.snapshot().items, exit, journal).length > 0 + watchedExitRevisions(providerTimelineExecutionItems(journal.snapshot().items), exit, journal) + .length > 0 ) } @@ -48,7 +50,7 @@ export function unfinishedStructuredAgentSessionWorkWasInterrupted( exit?: StructuredAgentSessionWatchedExit, heldSends?: ReadonlySet ): boolean { - const currentSnapshot = journal.snapshot() + const currentItemsSnapshot = providerTimelineExecutionItems(journal.snapshot().items) if ( journal .submissions?.() @@ -59,16 +61,16 @@ export function unfinishedStructuredAgentSessionWorkWasInterrupted( !isQueuedAgentJournalSubmission(send) && (send.dispatchState === 'pending' || send.dispatchState === 'unknown') ) || - currentSnapshot.items.some(isInProgressStructuredAgentSessionItem) + currentItemsSnapshot.some(isInProgressStructuredAgentSessionItem) ) { return true } // Stream closure may mark a turn unverifiable before the child’s exit arrives. - if (watchedExitRevisions(currentSnapshot.items, exit, journal).length > 0) { + if (watchedExitRevisions(currentItemsSnapshot, exit, journal).length > 0) { return true } if ( - currentSnapshot.items.some((item) => { + currentItemsSnapshot.some((item) => { const turn = readAgentJournalTurn(item.body) return turn?.state === 'interrupted' && turn.completedAt === observedExitAt }) @@ -79,7 +81,7 @@ export function unfinishedStructuredAgentSessionWorkWasInterrupted( if (inProgressBefore.length === 0) { return false } - const currentItems = new Map(currentSnapshot.items.map((item) => [item.itemId, item])) + const currentItems = new Map(currentItemsSnapshot.map((item) => [item.itemId, item])) const runningTurns = inProgressBefore.filter( (item) => readAgentJournalTurn(item.body)?.state === 'running' ) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts b/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts index 7e0cdce1c95..89535fd4c94 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-rpc.test-fixture.ts @@ -100,12 +100,15 @@ function statusFeed(): StructuredAgentSessionStatusFeed { [ STATUS_SESSION, { - // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the status feed reads only cursor(), lastActivityAt(), stopMarks, snapshot(), itemFence() and runningTurn() of a journal with no sends. + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the status feed uses these held rows and provenance reads; the item lookup is checked against the real journal contract. journal: { cursor: () => ({ epoch: 'epoch-status', sequence: 2 }), lastActivityAt: () => 2, stopMarks: { latest: () => null, revision: () => 0 }, snapshot: () => ({ items: STATUS_ITEMS, submissions: [] }), + item: ((itemId) => + STATUS_ITEMS.find((item) => item.itemId === itemId) ?? + null) satisfies AgentSessionJournal['item'], itemFence: () => undefined, runningTurn: () => null } as unknown as AgentSessionJournal, @@ -255,10 +258,8 @@ export function dispatcher( ...runtimeCalls, ...runtimeOverrides } - return new RpcDispatcher({ - runtime: runtime as unknown as OrcaRuntimeService, - methods - }) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: this partial runtime supplies the dispatcher methods under test and runtimeOverrides; a missing access fails the test. + return new RpcDispatcher({ runtime: runtime as unknown as OrcaRuntimeService, methods }) } /** Reply transport carries the negotiated client identity the capability gate reads. */ diff --git a/src/shared/structured-agent-session-message-projection.test.ts b/src/shared/structured-agent-session-message-projection.test.ts index 3a32bb5c059..8c3c7a2c852 100644 --- a/src/shared/structured-agent-session-message-projection.test.ts +++ b/src/shared/structured-agent-session-message-projection.test.ts @@ -9,7 +9,10 @@ import type { } from './agent-session-journal-types' import { agentJournalSubmissionKey } from './agent-session-journal-item-key' import { structuredAgentSessionSendBody } from './structured-agent-session-send-mutation' -import { projectStructuredAgentSessionMessages } from './structured-agent-session-message-projection' +import { + projectStructuredAgentSessionMessages, + structuredAgentSessionRejectedShownInPlace +} from './structured-agent-session-message-projection' import { projectNativeChatTranscriptMessages } from './native-chat-transcript-projection' import { DISPATCH_REJECTED_CANCELLED, @@ -263,3 +266,41 @@ describe('a not-sent row', () => { expect(shown.at(-1)).toMatchObject({ queued: true }) }) }) + +// Later evidence uses its maximum time, independent of its own display eligibility. +it('keeps rejected-copy scope, equality, withdrawal and forward order with nonmonotone times', () => { + const copy = (id: string, fingerprint: string, fields: Partial = {}) => + rejected({ clientMessageId: id, payloadFingerprint: fingerprint, resolvedAt: 80, ...fields }) + const submissions = [ + copy('old-a', 'a'), + copy('old-b', 'b'), + copy('old-c', 'c', { acceptedSequence: 50 }), + copy('old-d', 'd', { resolvedAt: 100 }), + copy('old-e', 'e'), + copy('unresolved', 'f', { resolvedAt: null }), + copy('near-a', 'a', { submittedAt: 10, dispatchState: 'accepted' }), + copy('far-a', 'a', { submittedAt: 90, keptAsQueuedMessageId: 'card' }), + copy('withdrawn-b', 'b', { submittedAt: 100, rejection: { kind: 'cancelled' } }), + copy('new-context-c', 'c', { + submittedAt: 100, + acceptedSequence: 51, + dispatchState: 'accepted' + }), + copy('command-d', 'd', { submittedAt: 100 }), + copy('queued-e', 'e', { submittedAt: 100, queuedMessageId: 'queued' }), + copy('later-f', 'f', { submittedAt: 100, dispatchState: 'accepted' }) + ] + const before = structuredClone(submissions) + expect([ + ...structuredAgentSessionRejectedShownInPlace( + submissions, + new Set([agentJournalSubmissionKey('command-d')]), + [50] + ) + ]).toEqual([ + agentJournalSubmissionKey('old-b'), + agentJournalSubmissionKey('old-c'), + agentJournalSubmissionKey('unresolved') + ]) + expect(submissions).toEqual(before) +}) diff --git a/src/shared/structured-agent-session-message-projection.ts b/src/shared/structured-agent-session-message-projection.ts index 4e4d403b0ff..22e52159d7c 100644 --- a/src/shared/structured-agent-session-message-projection.ts +++ b/src/shared/structured-agent-session-message-projection.ts @@ -48,22 +48,19 @@ export function structuredAgentSessionRejectedShownInPlace( commandItemIds: ReadonlySet, clearSequences: readonly number[] = [] ): Set { - // Track copies by body/scope; withdrawals supersede no failed copy. - const copies = new Map() + const latest = new Map() + const later = Array.from({ length: submissions.length }) const scopeKey = (submission: AgentJournalSubmission) => JSON.stringify([ agentSessionContextSequenceFor(submission.acceptedSequence ?? 0, clearSequences), submission.payloadFingerprint ]) - for (const [index, submission] of submissions.entries()) { + for (let index = submissions.length - 1; index >= 0; index--) { + const submission = submissions[index] + const key = scopeKey(submission) + later[index] = latest.get(key) if (!dispatchWasWithdrawn(submission)) { - const copy = { index, submittedAt: submission.submittedAt } - const same = copies.get(scopeKey(submission)) - if (same) { - same.push(copy) - } else { - copies.set(scopeKey(submission), [copy]) - } + latest.set(key, Math.max(latest.get(key) ?? -Infinity, submission.submittedAt)) } } const shown = new Set() @@ -76,12 +73,10 @@ export function structuredAgentSessionRejectedShownInPlace( // Persist card ownership on the send so editing/deleting the card cannot resurrect its row. submission.keptAsQueuedMessageId !== undefined || commandItemIds.has(agentJournalSubmissionKey(submission.clientMessageId)) || - // Collapse rejected retries only when resent after known rejection; earlier intentional repeats remain visible. - // Both client Retry and host redelivery mint fresh IDs. - (resolvedAt !== null && - (copies.get(scopeKey(submission)) ?? []).some( - (copy) => copy.index > index && copy.submittedAt >= resolvedAt - )) + // Collapses resends of a rejected message: past Retries resent it under a new id, and the + // host re-delivers its own messages under new ids. Only a later copy sent once the rejection + // was known counts, so a repeat sent before it is kept. + (resolvedAt !== null && later[index] !== undefined && later[index] >= resolvedAt) ) { continue }