diff --git a/mobile/src/session/MobileNativeChatMessage.test.ts b/mobile/src/session/MobileNativeChatMessage.test.ts index 67e901328667..c8193641fefb 100644 --- a/mobile/src/session/MobileNativeChatMessage.test.ts +++ b/mobile/src/session/MobileNativeChatMessage.test.ts @@ -3,6 +3,8 @@ import { act, create, type ReactTestInstance, type ReactTestRenderer } from 'rea import { afterEach, describe, expect, it, vi } from 'vitest' import { MAX_TOOL_DETAIL_LENGTH } from '../../../src/shared/native-chat-tool-summary' import type { NativeChatMessage } from '../../../src/shared/native-chat-types' +import { AGENT_SESSION_HOST_STATUS_COPY } from '../../../src/shared/agent-session-host-status-rows' +import { colors } from '../theme/mobile-theme' vi.mock('react-native', async () => { const React = await import('react') @@ -60,6 +62,7 @@ describe('MobileNativeChatMessage', () => { function render( message: NativeChatMessage, props: { + fontScale?: number toolsExpanded?: boolean structuredActivityUi?: boolean activeTurnIsWorking?: boolean @@ -81,6 +84,38 @@ describe('MobileNativeChatMessage', () => { const textIn = (node: ReactTestInstance): string[] => node.findAllByType('Text' as never).map((text) => String(text.children.join(''))) + it.each(['system', 'user'] as const)( + 'renders a %s host notice as selectable muted text rather than a markdown answer', + (role) => { + const tree = render( + { + id: 'notice', + role, + timestamp: 1, + blocks: [ + { type: 'text', text: 'provider fallback', presentation: 'history-item-too-large' } + ] + }, + { fontScale: 1.5 } + ) + expect(tree.root.findAll((node) => String(node.type) === 'MobileMarkdown')).toHaveLength(0) + const text = tree.root.find((node) => String(node.type) === 'Text') + expect(text.props.children).toBe(AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large']) + expect(text.props.selectable).toBe(true) + expect(Object.assign({}, ...text.props.style)).toMatchObject({ + color: colors.textMuted, + fontSize: 25.5 + }) + } + ) + + it('preserves an ordinary assistant answer without interpreting its text as a host notice', () => { + const tree = render(toolMessage([{ type: 'text', text: 'provider fallback' }])) + expect(tree.root.find((node) => String(node.type) === 'MobileMarkdown').props.content).toBe( + 'provider fallback' + ) + }) + it('renders a loadable preview URI as an image thumbnail', () => { const tree = render(userMessage([{ type: 'image-ref', url: 'file:///a.jpg', alt: 'a photo' }])) const image = tree.root.findByType('Image' as never) diff --git a/mobile/src/session/MobileNativeChatMessage.tsx b/mobile/src/session/MobileNativeChatMessage.tsx index 9c894e3cd7fc..40635166281f 100644 --- a/mobile/src/session/MobileNativeChatMessage.tsx +++ b/mobile/src/session/MobileNativeChatMessage.tsx @@ -4,6 +4,10 @@ import { Image, Text as NativeText, View } from 'react-native' import { splitNativeChatBlocks } from '../../../src/shared/native-chat-tool-fold' import { selectActiveToolCall } from '../../../src/shared/native-chat-tool-activity' import { isImageRefBlock, isTextBlock } from '../../../src/shared/native-chat-types' +import { + AGENT_SESSION_HOST_STATUS_COPY, + isAgentSessionHostStatusPresentation +} from '../../../src/shared/agent-session-host-status-rows' import type { NativeChatBlock, NativeChatMessage } from '../../../src/shared/native-chat-types' import { MobileMarkdown } from '../components/MobileMarkdown' import { MobileNativeChatTurnStatus } from './MobileNativeChatTurnStatus' @@ -24,6 +28,13 @@ function Prose({ onOpenFile?: (relativePath: string) => void }): React.JSX.Element | null { if (isTextBlock(block)) { + if (isAgentSessionHostStatusPresentation(block.presentation)) { + return ( + + {AGENT_SESSION_HOST_STATUS_COPY[block.presentation]} + + ) + } // Inverted (user) bubbles use a fixed dark-on-light text rather than the // markdown renderer's light-on-dark palette. if (invert) { diff --git a/mobile/src/session/mobile-native-chat-message-styles.ts b/mobile/src/session/mobile-native-chat-message-styles.ts index 51ff9ab1f36c..225d664e65e4 100644 --- a/mobile/src/session/mobile-native-chat-message-styles.ts +++ b/mobile/src/session/mobile-native-chat-message-styles.ts @@ -29,6 +29,11 @@ export const styles = StyleSheet.create({ lineHeight: TEXT_SIZE + 6, fontWeight: '500' }, + hostNotice: { + color: colors.textMuted, + fontSize: TEXT_SIZE, + lineHeight: TEXT_SIZE + 6 + }, reasoning: { opacity: 0.7 }, diff --git a/mobile/src/session/mobile-native-chat-pair-window.test.ts b/mobile/src/session/mobile-native-chat-pair-window.test.ts new file mode 100644 index 000000000000..fb2bd15605c2 --- /dev/null +++ b/mobile/src/session/mobile-native-chat-pair-window.test.ts @@ -0,0 +1,134 @@ +import { describe, expect, it } from 'vitest' +import { createNativeChatMerger, replaceList } from '../../../src/shared/native-chat-merge' +import type { NativeChatMessage } from '../../../src/shared/native-chat-types' +import { applyMobileNativeChatStreamFrame } from './mobile-native-chat-stream-frame' + +function row(id: string, role: NativeChatMessage['role'] = 'assistant'): NativeChatMessage { + return { id, role, source: 'transcript', timestamp: 1, blocks: [{ type: 'text', text: id }] } +} + +function pair(id: string): NativeChatMessage[] { + return [row(`${id}:reasoning`, 'reasoning'), row(id)] +} + +describe('mobile complete-pair stream windows', () => { + it.each(['appended', 'snapshot'])( + 'keeps a pair at the %s cutoff without moving the cursor', + (type) => { + const merger = createNativeChatMerger() + const base = [ + ...pair('opencode:oldest'), + ...Array.from({ length: 38 }, (_, i) => row(`m${i}`)) + ] + replaceList(merger, base) + const result = applyMobileNativeChatStreamFrame({ + merger, + frame: { + type, + messages: type === 'snapshot' ? [...base.slice(-1), row('latest')] : [row('latest')], + hasMore: true, + beforeOffset: 1 + }, + limit: 40, + replaceSnapshot: false + }) + expect(result).toEqual({ kind: 'messages', messages: [...base, row('latest')] }) + expect(merger.indexById.get('opencode:oldest:reasoning')).toBe(0) + } + ) + + it.each(['appended', 'snapshot'])( + 'keeps reasoning, answer and omission from one raw row at the %s cutoff', + (type) => { + const group = [...pair('opencode:oldest'), row('opencode:oldest:omission', 'system')].map( + (message) => ({ ...message, transcriptOffset: 7 }) + ) + const base = [...group, ...Array.from({ length: 38 }, (_, i) => row(`m${i}`))] + const merger = createNativeChatMerger() + replaceList(merger, base) + const result = applyMobileNativeChatStreamFrame({ + merger, + frame: { + type, + messages: type === 'snapshot' ? [...base.slice(-1), row('latest')] : [row('latest')], + hasMore: true, + beforeOffset: 7 + }, + limit: 40, + replaceSnapshot: false + }) + expect(result).toEqual({ kind: 'messages', messages: [...base, row('latest')] }) + expect(merger.list.slice(0, 3)).toEqual(group) + expect(merger.list).toHaveLength(42) + } + ) + + it('invalidates the cursor when the oldest complete pair leaves', () => { + const merger = createNativeChatMerger() + const base = [...pair('opencode:oldest'), ...Array.from({ length: 39 }, (_, i) => row(`m${i}`))] + replaceList(merger, base) + const result = applyMobileNativeChatStreamFrame({ + merger, + frame: { type: 'appended', messages: [row('latest')] }, + limit: 40, + replaceSnapshot: false + }) + expect(result).toEqual({ + kind: 'messages', + messages: [...base.slice(2), row('latest')], + cursorInvalidated: true + }) + expect(merger.indexById.has('opencode:oldest')).toBe(false) + expect(merger.indexById.has('opencode:oldest:reasoning')).toBe(false) + }) + + it('retains a host page with an extra pair row across a user append followed by a pair append', () => { + const merger = createNativeChatMerger() + const base = [ + ...pair('opencode:first'), + ...Array.from({ length: 13 }, (_, i) => [ + row(`user-${i}`, 'user'), + ...pair(`opencode:${i}`) + ]).flat() + ] + applyMobileNativeChatStreamFrame({ + merger, + frame: { type: 'snapshot', messages: base, hasMore: true, beforeOffset: 123 }, + limit: 40, + replaceSnapshot: true + }) + applyMobileNativeChatStreamFrame({ + merger, + frame: { type: 'appended', messages: [row('latest-user', 'user')] }, + limit: 40, + replaceSnapshot: false + }) + const result = applyMobileNativeChatStreamFrame({ + merger, + frame: { type: 'appended', messages: pair('opencode:latest') }, + limit: 40, + replaceSnapshot: false + }) + expect(result).toMatchObject({ kind: 'messages', cursorInvalidated: true }) + expect(merger.list).toHaveLength(41) + expect(merger.list.slice(0, 2)).toEqual(pair('opencode:0')) + expect(merger.list.slice(-2)).toEqual(pair('opencode:latest')) + }) + + it('keeps legacy peers with independent reasoning IDs at the raw row limit', () => { + const merger = createNativeChatMerger() + replaceList(merger, [row('claude-thinking', 'reasoning'), row('claude-answer')]) + const result = applyMobileNativeChatStreamFrame({ + merger, + frame: { type: 'appended', messages: [row('latest')] }, + limit: 2, + replaceSnapshot: false + }) + expect(result).toEqual({ + kind: 'messages', + messages: [row('claude-answer'), row('latest')], + cursorInvalidated: true + }) + expect(merger.list).toHaveLength(2) + }) +}) diff --git a/src/main/ai-vault/session-scanner-opencode-native-worker.test.ts b/src/main/ai-vault/session-scanner-opencode-native-worker.test.ts index 9e8511967f2e..9b41f700b245 100644 --- a/src/main/ai-vault/session-scanner-opencode-native-worker.test.ts +++ b/src/main/ai-vault/session-scanner-opencode-native-worker.test.ts @@ -73,8 +73,13 @@ describe('OpenCode reads through the production shared worker', () => { throw new Error('Expected the bounded native page') } expect(page.items.map((item) => item.rowid)).toEqual([2, 3]) + expect(page.items[0]?.message.role).toBe('system') expect(page.items[0]?.message.blocks).toEqual([ - { type: 'text', text: AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'] } + { + type: 'text', + text: AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'], + presentation: 'history-item-too-large' + } ]) const repeat = await client.readNativeChat(args) expect(repeat).toEqual(page) diff --git a/src/main/native-chat/transcript-opencode-part-blocks.ts b/src/main/native-chat/transcript-opencode-part-blocks.ts index 572851b34ae2..ec8ca0cd0460 100644 --- a/src/main/native-chat/transcript-opencode-part-blocks.ts +++ b/src/main/native-chat/transcript-opencode-part-blocks.ts @@ -1,6 +1,11 @@ import { fileURLToPath } from 'node:url' -import type { NativeChatBlock, NativeChatImageRefBlock } from '../../shared/native-chat-types' -import { AGENT_SESSION_HOST_STATUS_COPY } from '../../shared/agent-session-host-status-rows' +import type { + NativeChatBlock, + NativeChatImageRefBlock, + NativeChatMessage, + NativeChatTextBlock +} from '../../shared/native-chat-types' +import { agentSessionHostStatusBody } from '../../shared/agent-session-host-status-rows' import { asRecord, extractString, parseJsonObject } from '../ai-vault/session-scanner-values' // query module so each stays under the repo's file-size cap. Electron-free: // runs on the OpenCode SQLite worker thread (#8864). @@ -13,11 +18,50 @@ export type OpenCodePartRow = { export const OPENCODE_TRANSCRIPT_MAX_ROW_BYTES = 2 * 1024 * 1024 -export function opencodeMessageBlocks(partRows: OpenCodePartRow[]): NativeChatBlock[] { +export function opencodeMessages( + message: Pick, + partRows: OpenCodePartRow[], + additionalBlocks: NativeChatBlock[] = [] +): NativeChatMessage[] { + const { blocks, reasoning, notices } = opencodeMessageContent(partRows) + blocks.push(...additionalBlocks) + const messages: NativeChatMessage[] = [] + if (reasoning.length > 0) { + messages.push({ + ...message, + id: blocks.length > 0 || notices.length > 0 ? `${message.id}:reasoning` : message.id, + role: 'reasoning', + blocks: reasoning, + source: 'transcript' + }) + } + if (blocks.length > 0) { + messages.push({ ...message, blocks, source: 'transcript' }) + } + if (notices.length > 0) { + messages.push({ + ...message, + id: messages.length > 0 ? `${message.id}:omission` : message.id, + role: 'system', + blocks: notices, + source: 'transcript' + }) + } + return messages +} + +function opencodeMessageContent(partRows: OpenCodePartRow[]): { + blocks: NativeChatBlock[] + reasoning: NativeChatBlock[] + notices: NativeChatTextBlock[] +} { const blocks: NativeChatBlock[] = [] + const reasoning: NativeChatBlock[] = [] + const notices: NativeChatTextBlock[] = [] for (const partRow of partRows) { if (partRow.data === null) { - blocks.push({ type: 'text', text: AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'] }) + const { text, presentation } = agentSessionHostStatusBody('history-item-too-large') + notices.push({ type: 'text', text, presentation }) continue } const part = parseJsonObject(partRow.data) @@ -38,7 +82,7 @@ export function opencodeMessageBlocks(partRows: OpenCodePartRow[]): NativeChatBl case 'reasoning': { const text = extractString(part.text) if (text) { - blocks.push({ type: 'text', text }) + reasoning.push({ type: 'text', text }) } break } @@ -46,6 +90,18 @@ export function opencodeMessageBlocks(partRows: OpenCodePartRow[]): NativeChatBl blocks.push(...opencodeToolBlocks(part)) break } + case 'patch': { + const files = Array.isArray(part.files) + ? part.files.filter((file): file is string => typeof file === 'string') + : [] + blocks.push({ + type: 'tool-call', + name: 'patch', + state: 'completed', + input: { hash: extractString(part.hash), files } + }) + break + } case 'file': { const block = opencodeFileBlock(part) if (block) { @@ -58,7 +114,7 @@ export function opencodeMessageBlocks(partRows: OpenCodePartRow[]): NativeChatBl break } } - return blocks + return { blocks, reasoning, notices } } function opencodeFileBlock(part: Record): NativeChatImageRefBlock | null { diff --git a/src/main/native-chat/transcript-opencode-read-bounds.test.ts b/src/main/native-chat/transcript-opencode-read-bounds.test.ts new file mode 100644 index 000000000000..46f4ace57c1a --- /dev/null +++ b/src/main/native-chat/transcript-opencode-read-bounds.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, it, vi } from 'vitest' +import type { NativeChatMessage } from '../../shared/native-chat-types' +import { + readOpenCodeNativeChatTranscriptFull, + readOpenCodeNativeChatTranscriptTail +} from './transcript-opencode' + +vi.mock('../managed-data-accounts/service', () => ({ getManagedDataAccountService: vi.fn() })) + +describe('OpenCode full native read bounds', () => { + it('counts reasoning, answer and image bytes cumulatively across pages', async () => { + let calls = 0 + const chunk = 'x'.repeat(1_000_000) + const messages: NativeChatMessage[] = Array.from({ length: 12 }, (_, index) => ({ + id: String(index), + role: index % 2 === 0 ? 'reasoning' : 'assistant', + timestamp: 1, + source: 'transcript', + blocks: [{ type: 'text', text: chunk }] + })) + messages.push({ + id: 'image', + role: 'user', + timestamp: 1, + source: 'transcript', + blocks: [{ type: 'image-ref', url: `data:image/png;base64,${chunk}` }] + }) + const result = await readOpenCodeNativeChatTranscriptFull('session', { + resolveDbPath: async () => 'private-db', + readPage: async () => ({ + items: messages.map((message) => ({ rowid: 100 - calls, fingerprint: '', message })), + hasMore: true, + beforeMessageRowId: 100 - ++calls + }) + }) + expect(result).toEqual({ error: 'OpenCode transcript exceeds its full read limit' }) + expect(calls).toBe(6) + }) + + it.each(['tail', 'full'] as const)( + 'keeps cancellation attached to %s discovery and reads', + async (kind) => { + const controller = new AbortController() + const resolveDbPath = vi.fn(async (_sessionId?: string, signal?: AbortSignal) => { + expect(signal).toBe(controller.signal) + return 'private-db' + }) + const readPage = vi.fn(async (_args: unknown, signal?: AbortSignal) => { + expect(signal).toBe(controller.signal) + controller.abort(new Error('Cancelled private read')) + signal?.throwIfAborted() + return null + }) + const deps = { resolveDbPath, readPage } + const result = + kind === 'tail' + ? await readOpenCodeNativeChatTranscriptTail( + { sessionId: 'session', limit: 1 }, + deps, + controller.signal + ) + : await readOpenCodeNativeChatTranscriptFull('session', deps, controller.signal) + expect(result).toEqual({ error: 'Cancelled private read' }) + expect(readPage).toHaveBeenCalledOnce() + } + ) +}) diff --git a/src/main/native-chat/transcript-opencode-semantic-fidelity.test.ts b/src/main/native-chat/transcript-opencode-semantic-fidelity.test.ts new file mode 100644 index 000000000000..aa30a4b743a5 --- /dev/null +++ b/src/main/native-chat/transcript-opencode-semantic-fidelity.test.ts @@ -0,0 +1,331 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { NativeChatMessage } from '../../shared/native-chat-types' +import { AGENT_SESSION_HOST_STATUS_COPY } from '../../shared/agent-session-host-status-rows' +import Database from '../sqlite/sync-database' +import { opencodeMessages } from './transcript-opencode-part-blocks' +import { + readOpenCodeTranscriptPage, + readOpenCodeTranscriptSignal +} from './transcript-opencode-sqlite-query' +import { readOpenCodeNativeChatTranscriptFull } from './transcript-opencode' +import { subscribeOpenCodeNativeChatTranscript } from './transcript-opencode-subscribe' + +vi.mock('../managed-data-accounts/service', () => ({ getManagedDataAccountService: vi.fn() })) + +const fixtures: { db: Database.Database; root: string }[] = [] +afterEach(() => { + for (const { db, root } of fixtures.splice(0)) { + db.close() + rmSync(root, { recursive: true, force: true }) + } +}) + +function fixture(version: 'v1' | 'v2') { + const root = mkdtempSync(join(tmpdir(), 'orca-opencode-semantics-')) + const path = join(root, 'opencode.db') + const db = new Database(path) + fixtures.push({ db, root }) + if (version === 'v1') { + db.exec(`CREATE TABLE session (id TEXT PRIMARY KEY); + CREATE TABLE message (id TEXT, session_id TEXT, time_created INTEGER, time_updated INTEGER, data TEXT); + CREATE TABLE part (id TEXT, message_id TEXT, session_id TEXT, time_updated INTEGER, data TEXT); + INSERT INTO session VALUES ('session');`) + } else { + db.exec(`CREATE TABLE session_v2 (id TEXT PRIMARY KEY); + CREATE TABLE session_message (id TEXT, session_id TEXT, type TEXT, seq INTEGER, + data TEXT, time_created INTEGER, time_updated INTEGER); + INSERT INTO session_v2 VALUES ('session');`) + } + let sequence = 0 + const insert = (id: string, parts: unknown[]) => { + sequence += 7 + if (version === 'v1') { + db.prepare('INSERT INTO message VALUES (?, ?, ?, ?, ?)').run( + id, + 'session', + sequence, + sequence, + JSON.stringify({ role: 'assistant' }) + ) + parts.forEach((part, index) => { + db.prepare('INSERT INTO part VALUES (?, ?, ?, ?, ?)').run( + `${id}:${index}`, + id, + 'session', + sequence, + JSON.stringify(part) + ) + }) + } else { + db.prepare('INSERT INTO session_message VALUES (?, ?, ?, ?, ?, ?, ?)').run( + id, + 'session', + 'assistant', + sequence, + JSON.stringify({ content: parts }), + sequence, + sequence + ) + } + } + const prefix = version === 'v1' ? '' : 'opencode:' + return { db, path, insert, prefix } +} + +const mixed = (answer = 'ANSWER_SENTINEL') => [ + { type: 'reasoning', text: 'REASONING_SENTINEL' }, + { type: 'text', text: answer } +] + +it('keeps reasoning apart from prose, image refs, tools, and applied patch records', () => { + // Contracts captured through isolated 2.0.16 and official 1.18.30 CLI ingestion. + const parts = [ + ...mixed(), + { type: 'text', text: 'injected context', synthetic: true }, + { type: 'step-finish' }, + { type: 'file', mime: 'image/png', url: 'data:image/png;base64,YQ==' }, + { type: 'patch', hash: 'snapshot-before-edit', files: ['sentinel.txt', 3] } + ].map((part) => ({ message_id: 'row', time_updated: 1, data: JSON.stringify(part) })) + const messages = opencodeMessages({ id: 'row', role: 'assistant', timestamp: 1 }, parts) + expect(messages).toEqual([ + { + id: 'row:reasoning', + role: 'reasoning', + timestamp: 1, + source: 'transcript', + blocks: [{ type: 'text', text: 'REASONING_SENTINEL' }] + }, + { + id: 'row', + role: 'assistant', + timestamp: 1, + source: 'transcript', + blocks: [ + { type: 'text', text: 'ANSWER_SENTINEL' }, + { type: 'image-ref', url: 'data:image/png;base64,YQ==' }, + { + type: 'tool-call', + name: 'patch', + state: 'completed', + input: { hash: 'snapshot-before-edit', files: ['sentinel.txt'] } + } + ] + } + ]) + expect( + opencodeMessages({ id: 'thinking', role: 'assistant', timestamp: null }, parts.slice(0, 1)) + ).toEqual([ + { + id: 'thinking', + role: 'reasoning', + timestamp: null, + source: 'transcript', + blocks: [{ type: 'text', text: 'REASONING_SENTINEL' }] + } + ]) +}) + +describe.each(['v1', 'v2'] as const)('%s semantic pagination', (version) => { + it.each([1, 2, 3])('reconstructs sparse mixed history once with a page limit of %i', (limit) => { + const { path, insert, prefix } = fixture(version) + const expected: string[] = [] + for (let index = 0; index < 12; index++) { + const id = String(index) + if (index % 3 === 1) { + insert(id, [{ type: 'step-start' }]) + } else if (index % 3 === 2) { + insert(id, mixed()) + expected.push(`${prefix}${id}:reasoning`, `${prefix}${id}`) + } else { + insert(id, [{ type: 'text', text: id }]) + expected.push(`${prefix}${id}`) + } + } + const pages: NativeChatMessage[][] = [] + let cursor: number | undefined + for (let pageIndex = 0; pageIndex < 30; pageIndex++) { + const page = readOpenCodeTranscriptPage({ + dbPath: path, + sessionId: 'session', + limit, + beforeMessageRowId: cursor + })! + expect(page.items.length).toBeLessThanOrEqual(limit + 1) + for (const item of page.items.filter((item) => item.message.role === 'reasoning')) { + expect( + page.items.some( + (other) => other.rowid === item.rowid && other.message.role === 'assistant' + ) + ).toBe(true) + } + pages.push(page.items.map((item) => item.message)) + if (!page.hasMore) { + break + } + expect(page.beforeMessageRowId).toBeLessThan(cursor ?? Number.MAX_SAFE_INTEGER) + cursor = page.beforeMessageRowId ?? undefined + } + expect( + pages + .toReversed() + .flat() + .map((message) => message.id) + ).toEqual(expected) + }) + + it('reconstructs a full read beyond its 500-message window without losing a split row', async () => { + const { path, insert, prefix } = fixture(version) + const expected: string[] = [] + for (let index = 0; index < 260; index++) { + insert(String(index), mixed()) + expected.push(`${prefix}${index}:reasoning`, `${prefix}${index}`) + } + const result = await readOpenCodeNativeChatTranscriptFull('session', { + resolveDbPath: async () => path, + readPage: async (args) => readOpenCodeTranscriptPage(args) + }) + expect('messages' in result && result.messages.map((message) => message.id)).toEqual(expected) + }) + + it('keeps both presentations on streaming mutation and new rows', async () => { + const { db, path, insert, prefix } = fixture(version) + insert('reply', mixed()) + const snapshots: NativeChatMessage[][] = [] + const replacements: NativeChatMessage[][] = [] + const appends: NativeChatMessage[][] = [] + let visible: NativeChatMessage[] = [] + const subscription = subscribeOpenCodeNativeChatTranscript( + { + agent: 'opencode', + sessionId: 'session', + initialLimit: 1, + resolvePollIntervalMs: 5, + onAppend: (messages) => { + appends.push(messages) + visible.push(...messages) + }, + onInitialSnapshot: (messages) => { + snapshots.push(messages) + visible = [...messages] + }, + onReplace: (messages) => { + replacements.push(messages) + visible = [...messages] + } + }, + undefined, + { + resolveDbPath: async () => path, + readSignal: async (dbPath, sessionId) => readOpenCodeTranscriptSignal(dbPath, sessionId), + readPage: async (args) => readOpenCodeTranscriptPage(args) + } + ) + try { + await expect.poll(() => snapshots.length).toBe(1) + expect(snapshots[0].map((message) => message.role)).toEqual(['reasoning', 'assistant']) + if (version === 'v1') { + db.prepare("UPDATE part SET data = ?, time_updated = 100 WHERE id = 'reply:1'").run( + JSON.stringify({ type: 'text', text: 'complete' }) + ) + } else { + db.prepare('UPDATE session_message SET data = ?, time_updated = 100').run( + JSON.stringify({ content: mixed('complete') }) + ) + } + await expect.poll(() => replacements.length).toBe(1) + expect(replacements[0].map((message) => message.role)).toEqual(['reasoning', 'assistant']) + expect(replacements[0][1].blocks).toEqual([{ type: 'text', text: 'complete' }]) + insert('next', mixed('next answer')) + await expect.poll(() => appends.length + replacements.length).toBe(2) + expect(visible.map((message) => message.id)).toEqual( + ['reply:reasoning', 'reply', 'next:reasoning', 'next'].map((id) => `${prefix}${id}`) + ) + expect(visible.map((message) => message.role)).toEqual([ + 'reasoning', + 'assistant', + 'reasoning', + 'assistant' + ]) + expect(visible[1].blocks).toEqual([{ type: 'text', text: 'complete' }]) + expect(visible[3].blocks).toEqual([{ type: 'text', text: 'next answer' }]) + } finally { + subscription.unsubscribe() + } + }) + + it('keeps complete pairs around an omitted oversized history row', () => { + const { path, insert, prefix } = fixture(version) + insert('before', mixed('BEFORE')) + insert('large', [ + ...mixed('MIDDLE'), + { type: 'file', mime: 'image/png', url: '😀'.repeat(2 * 1024 * 1024) } + ]) + insert('after', mixed('AFTER')) + const messages: NativeChatMessage[] = [] + const cursors = new Set() + let beforeMessageRowId: number | undefined + while (true) { + const page = readOpenCodeTranscriptPage({ + dbPath: path, + sessionId: 'session', + limit: 1, + beforeMessageRowId + }) + expect(page).not.toBeNull() + if (!page) { + throw new Error('History page unavailable') + } + messages.unshift(...page.items.map((item) => item.message)) + if (!page.hasMore) { + break + } + expect(page.beforeMessageRowId).not.toBeNull() + if (page.beforeMessageRowId === null) { + throw new Error('History cursor unavailable') + } + expect(cursors.has(page.beforeMessageRowId)).toBe(false) + cursors.add(page.beforeMessageRowId) + beforeMessageRowId = page.beforeMessageRowId + } + const middleIds = version === 'v1' ? ['large:reasoning', 'large', 'large:omission'] : ['large'] + expect(messages.map((message) => message.id)).toEqual( + ['before:reasoning', 'before', ...middleIds, 'after:reasoning', 'after'].map( + (id) => `${prefix}${id}` + ) + ) + const notices = messages.flatMap((message) => + message.blocks.filter( + (block) => + block.type === 'text' && + block.text === AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'] + ) + ) + expect(notices).toHaveLength(1) + expect(notices[0]).toMatchObject({ presentation: 'history-item-too-large' }) + const noticeRow = messages.find((message) => + message.blocks.some( + (block) => block.type === 'text' && block.presentation === 'history-item-too-large' + ) + ) + expect(noticeRow?.role).toBe('system') + expect(noticeRow?.transcriptOffset).toBeTypeOf('number') + expect(Buffer.byteLength(JSON.stringify(messages))).toBeLessThan(4096) + }) + + it('enforces one page byte budget across multiple mixed rows', () => { + const { path, insert } = fixture(version) + const large = 'x'.repeat(1_000_000) + for (let index = 0; index < 20; index++) { + insert(String(index), [ + { type: 'reasoning', text: large }, + { type: 'text', text: large } + ]) + } + expect(() => + readOpenCodeTranscriptPage({ dbPath: path, sessionId: 'session', limit: 40 }) + ).toThrow('read limit') + }) +}) diff --git a/src/main/native-chat/transcript-opencode-sqlite-query.test.ts b/src/main/native-chat/transcript-opencode-sqlite-query.test.ts index d82613650729..6e500b7b9748 100644 --- a/src/main/native-chat/transcript-opencode-sqlite-query.test.ts +++ b/src/main/native-chat/transcript-opencode-sqlite-query.test.ts @@ -115,7 +115,11 @@ describe('readOpenCodeTranscriptPage', () => { const page = readOpenCodeTranscriptPage(args) expect(page?.items.map((item) => item.message.id)).toEqual(['msg-2', 'msg-3']) expect(page?.items[0]?.message.blocks).toEqual([ - { type: 'text', text: AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'] } + { + type: 'text', + text: AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'], + presentation: 'history-item-too-large' + } ]) expect(page?.items[0]?.rowid).toBe(messageRowid(db, 'msg-2')) expect(readOpenCodeTranscriptPage(args)?.items[0]?.fingerprint).toBe( @@ -152,13 +156,17 @@ describe('readOpenCodeTranscriptPage', () => { time: 1, data: { type: 'file', mime: 'image/png', url: '😀'.repeat(600_000) } }) - expect( - readOpenCodeTranscriptPage({ dbPath: path, sessionId: 'ses-1', limit: 1 })?.items[0]?.message - .blocks - ).toEqual([ - { type: 'text', text: 'see this' }, - { type: 'text', text: AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'] } + const page = readOpenCodeTranscriptPage({ dbPath: path, sessionId: 'ses-1', limit: 1 }) + expect(page?.items.map((item) => item.message)).toMatchObject([ + { id: 'msg', role: 'user', blocks: [{ type: 'text', text: 'see this' }] }, + { + id: 'msg:omission', + role: 'system', + blocks: [{ type: 'text', presentation: 'history-item-too-large' }] + } ]) + expect(page?.items[0]?.rowid).toBe(page?.items[1]?.rowid) + expect(page?.items[0]?.message.transcriptOffset).toBe(page?.items[1]?.message.transcriptOffset) }) it('returns null when the session row does not exist', () => { diff --git a/src/main/native-chat/transcript-opencode-sqlite-query.ts b/src/main/native-chat/transcript-opencode-sqlite-query.ts index 31729111c2ee..d245020b619a 100644 --- a/src/main/native-chat/transcript-opencode-sqlite-query.ts +++ b/src/main/native-chat/transcript-opencode-sqlite-query.ts @@ -11,7 +11,7 @@ import type SyncDatabase from '../sqlite/sync-database' type BindValue = SyncDatabase.BindValue type SqliteStatement = SyncDatabase.Statement import { - opencodeMessageBlocks, + opencodeMessages, OPENCODE_TRANSCRIPT_MAX_ROW_BYTES } from './transcript-opencode-part-blocks' // Cursors are opaque provider order: SQLite rowid in v1, session sequence in v2. @@ -166,8 +166,16 @@ export function readOpenCodeTranscriptPage(args: { break } } - const overshot = collected.length > limit - const trimmed = overshot ? collected.slice(0, limit) : collected + let retained = Math.min(limit, collected.length) + // A raw-row cursor cannot resume inside a reasoning/answer pair. + while ( + retained < collected.length && + collected[retained].rowid === collected[retained - 1].rowid + ) { + retained++ + } + const overshot = retained < collected.length + const trimmed = overshot ? collected.slice(0, retained) : collected const items = trimmed.toReversed() return { items, @@ -232,27 +240,25 @@ function mapMessageRows( const items: OpenCodeTranscriptItem[] = [] for (const row of rows) { const partList = partsByMessage.get(row.id) ?? [] - const blocks = opencodeMessageBlocks( + const record = row.data === null ? null : parseJsonObject(row.data) + const role = extractString(record?.role) + const messages = opencodeMessages( + { + id: row.id, + role: role === 'user' ? 'user' : role === 'assistant' ? 'assistant' : 'system', + timestamp: Number.isFinite(row.time_created) ? row.time_created : null + }, row.data === null ? [{ message_id: row.id, time_updated: row.time_updated, data: null }] : partList ) - if (blocks.length === 0) { - continue + for (const message of messages.toReversed()) { + items.push({ + rowid: row.message_rowid, + fingerprint: `${row.time_updated}:${partList.length}:${maxPartTimeUpdated(partList)}`, + message: { ...message, transcriptOffset: row.message_rowid } + }) } - const record = row.data === null ? null : parseJsonObject(row.data) - const role = extractString(record?.role) - items.push({ - rowid: row.message_rowid, - fingerprint: `${row.time_updated}:${partList.length}:${maxPartTimeUpdated(partList)}`, - message: { - id: row.id, - role: role === 'user' ? 'user' : role === 'assistant' ? 'assistant' : 'system', - blocks, - timestamp: Number.isFinite(row.time_created) ? row.time_created : null, - source: 'transcript' - } - }) } return items } diff --git a/src/main/native-chat/transcript-opencode2-sqlite.test.ts b/src/main/native-chat/transcript-opencode2-sqlite.test.ts index 333788c82bbd..7ddd653071d1 100644 --- a/src/main/native-chat/transcript-opencode2-sqlite.test.ts +++ b/src/main/native-chat/transcript-opencode2-sqlite.test.ts @@ -129,9 +129,16 @@ describe('OpenCode 2 native transcript', () => { expect(page?.items[0]).toMatchObject({ rowid: 2, message: { - role: 'user', + role: 'system', timestamp: 1, - blocks: [{ type: 'text', text: AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'] }] + transcriptOffset: 2, + blocks: [ + { + type: 'text', + text: AGENT_SESSION_HOST_STATUS_COPY['history-item-too-large'], + presentation: 'history-item-too-large' + } + ] } }) expect(readOpenCodeTranscriptPage(args)?.items[0]?.fingerprint).toBe( diff --git a/src/main/native-chat/transcript-opencode2-sqlite.ts b/src/main/native-chat/transcript-opencode2-sqlite.ts index 2ebfe950e10a..1b4f01033e9c 100644 --- a/src/main/native-chat/transcript-opencode2-sqlite.ts +++ b/src/main/native-chat/transcript-opencode2-sqlite.ts @@ -4,9 +4,10 @@ import { openCodeTranscriptPageLimit } from '../../shared/opencode-transcript-pa import { columnExists, tableExists } from '../opencode-usage/schema-helpers' import { asRecord, parseJsonObject } from '../ai-vault/session-scanner-values' import { - opencodeMessageBlocks, + opencodeMessages, OPENCODE_TRANSCRIPT_MAX_ROW_BYTES } from './transcript-opencode-part-blocks' +import type { NativeChatBlock } from '../../shared/native-chat-types' import type { OpenCodeTranscriptPage, OpenCodeTranscriptSignal, @@ -55,7 +56,7 @@ export function readOpenCode2TranscriptSignal( } } -function messageItem(value: unknown): OpenCodeTranscriptItem | null { +function messageItems(value: unknown): OpenCodeTranscriptItem[] { const row = asRecord(value) if ( typeof row?.id !== 'string' || @@ -65,30 +66,30 @@ function messageItem(value: unknown): OpenCodeTranscriptItem | null { ) { throw new Error('OpenCode transcript message is invalid') } + const messageCursor = row.cursor + const updatedAt = row.time_updated if ( row.data === null || (typeof row.data === 'string' && Buffer.byteLength(row.data) > OPENCODE_TRANSCRIPT_MAX_ROW_BYTES) ) { - return { - rowid: row.cursor, - fingerprint: `${row.time_updated}:omitted`, - message: { + return opencodeMessages( + { id: `opencode:${row.id}`, - role: row.type === 'user' ? 'user' : row.type === 'assistant' ? 'assistant' : 'system', - blocks: opencodeMessageBlocks([ - { message_id: row.id, time_updated: row.time_updated, data: null } - ]), - timestamp: row.time_created, - source: 'transcript' - } - } + role: 'system', + timestamp: row.time_created + }, + [{ message_id: row.id, time_updated: row.time_updated, data: null }] + ).map((message) => ({ + rowid: messageCursor, + fingerprint: `${updatedAt}:omitted`, + message: { ...message, transcriptOffset: messageCursor } + })) } if (typeof row.data !== 'string') { throw new Error('OpenCode transcript message is invalid') } const messageId = row.id - const updatedAt = row.time_updated const record = parseJsonObject(row.data) if (!record) { throw new Error('OpenCode transcript message contains invalid JSON') @@ -140,7 +141,7 @@ function messageItem(value: unknown): OpenCodeTranscriptItem | null { } ] }) - const blocks = opencodeMessageBlocks(parts) + const blocks: NativeChatBlock[] = [] const error = asRecord(record.error) if (typeof error?.message === 'string') { blocks.push({ type: 'text', text: error.message }) @@ -149,25 +150,28 @@ function messageItem(value: unknown): OpenCodeTranscriptItem | null { blocks.push({ type: 'text', text: 'Conversation interrupted' }) } if ( - blocks.length === 0 || - (row.type !== 'user' && - row.type !== 'assistant' && - row.type !== 'system' && - row.type !== 'idle') + row.type !== 'user' && + row.type !== 'assistant' && + row.type !== 'system' && + row.type !== 'idle' ) { - return null + return [] } - return { - rowid: row.cursor, - fingerprint: `${updatedAt}:${createHash('sha256').update(row.data).digest('hex')}`, - message: { + const messages = opencodeMessages( + { id: `opencode:${row.id}`, role: row.type === 'idle' ? 'system' : row.type, - blocks, - timestamp: row.time_created, - source: 'transcript' - } - } + timestamp: row.time_created + }, + parts, + blocks + ) + const fingerprint = `${updatedAt}:${createHash('sha256').update(row.data).digest('hex')}` + return messages.toReversed().map((message) => ({ + rowid: messageCursor, + fingerprint, + message: { ...message, transcriptOffset: messageCursor } + })) } export function readOpenCode2TranscriptPage( @@ -209,9 +213,9 @@ export function readOpenCode2TranscriptPage( if (++scannedRows > MAX_SCAN_ROWS || bytes > MAX_PAGE_BYTES) { throw new Error('OpenCode transcript page exceeds its read limit') } - const item = messageItem(value) + const mapped = messageItems(value) cursor = row.cursor - if (item) { + for (const item of mapped) { pageBytes += Buffer.byteLength(JSON.stringify(item)) if (pageBytes > MAX_PAGE_BYTES) { throw new Error('OpenCode transcript page exceeds its read limit') diff --git a/src/main/runtime/rpc/methods/native-chat-opencode-semantic-pages.test.ts b/src/main/runtime/rpc/methods/native-chat-opencode-semantic-pages.test.ts new file mode 100644 index 000000000000..358b723e1ac3 --- /dev/null +++ b/src/main/runtime/rpc/methods/native-chat-opencode-semantic-pages.test.ts @@ -0,0 +1,144 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import Database from '../../../sqlite/sync-database' +import { asRecord } from '../../../ai-vault/session-scanner-values' +import { readOpenCodeTranscriptPage } from '../../../native-chat/transcript-opencode-sqlite-query' +import { readOpenCodeNativeChatTranscriptTail } from '../../../native-chat/transcript-opencode' +import type { SubscribeNativeChatTranscriptArgs } from '../../../native-chat/transcript-watch-contract' +import { buildRegistry, isStreamingMethod, type RpcContext } from '../core' + +const state: { path: string; watcher: SubscribeNativeChatTranscriptArgs | null } = vi.hoisted( + () => ({ + path: '', + watcher: null + }) +) +vi.mock('../../../managed-data-accounts/service', () => ({ getManagedDataAccountService: vi.fn() })) +vi.mock('../../../native-chat/transcript-watch', () => ({ + readNativeChatTranscriptTail: (args: { + sessionId: string + limit: number + beforeOffset?: number + }) => + readOpenCodeNativeChatTranscriptTail(args, { + resolveDbPath: async () => state.path, + readPage: async (page) => readOpenCodeTranscriptPage(page) + }), + subscribeNativeChatTranscript: async (args: SubscribeNativeChatTranscriptArgs) => { + state.watcher = args + return { watching: true, unsubscribe: vi.fn() } + } +})) +import { NATIVE_CHAT_METHODS } from './native-chat' + +const roots: string[] = [] +afterEach(() => { + state.watcher = null + for (const root of roots.splice(0)) { + rmSync(root, { recursive: true, force: true }) + } +}) + +function fixture() { + const root = mkdtempSync(join(tmpdir(), 'orca-opencode-paired-pages-')) + roots.push(root) + state.path = join(root, 'opencode.db') + const db = new Database(state.path) + db.exec(`CREATE TABLE session_v2 (id TEXT PRIMARY KEY); + CREATE TABLE session_message (id TEXT, session_id TEXT, type TEXT, seq INTEGER, + data TEXT, time_created INTEGER, time_updated INTEGER); + INSERT INTO session_v2 VALUES ('session');`) + for (let index = 1; index <= 3; index++) { + db.prepare('INSERT INTO session_message VALUES (?, ?, ?, ?, ?, ?, ?)').run( + String(index), + 'session', + 'assistant', + index * 7, + JSON.stringify({ + content: [ + { type: 'reasoning', text: `thinking ${index}` }, + { type: 'text', text: `answer ${index}` } + ] + }), + 1, + 1 + ) + } + db.close() +} + +function context(): RpcContext { + const runtime = { registerSubscriptionCleanup: vi.fn(), cleanupSubscription: vi.fn() } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: These handlers only register or invoke subscription cleanup on the runtime. + return { runtime: runtime as unknown as RpcContext['runtime'], clientKind: 'mobile' } +} + +describe.each(['opencode', 'opencode2'])('%s paired semantic pages', (agent) => { + it('reconstructs reasoning and answers across raw-row cursors at limit 1', async () => { + fixture() + const method = buildRegistry(NATIVE_CHAT_METHODS).get('nativeChat.readSession')! + if (isStreamingMethod(method)) { + throw new Error('Expected a page read') + } + const pages: string[][] = [] + let beforeOffset: number | undefined + for (let index = 0; index < 4; index++) { + const params = method.params?.parse({ agent, sessionId: 'session', limit: 1, beforeOffset }) + const result = asRecord(await method.handler(params, context())) + expect(result?.hasMore).toBe(index < 2) + const messages = Array.isArray(result?.messages) ? result.messages : [] + expect(messages.map((message) => asRecord(message)?.role)).toEqual(['reasoning', 'assistant']) + pages.push(messages.map((message) => String(asRecord(message)?.id))) + if (result?.hasMore !== true) { + break + } + expect(typeof result.beforeOffset).toBe('number') + beforeOffset = typeof result.beforeOffset === 'number' ? result.beforeOffset : undefined + } + expect(pages.toReversed().flat()).toEqual([ + 'opencode:1:reasoning', + 'opencode:1', + 'opencode:2:reasoning', + 'opencode:2', + 'opencode:3:reasoning', + 'opencode:3' + ]) + }) + + it('keeps the pair in snapshot and replacement frames while sanitizing mobile text', async () => { + fixture() + const method = buildRegistry(NATIVE_CHAT_METHODS).get('nativeChat.subscribe')! + if (!isStreamingMethod(method)) { + throw new Error('Expected a subscription') + } + const frames: unknown[] = [] + await method.handler( + method.params?.parse({ agent, sessionId: 'session', limit: 1 }), + context(), + (frame) => frames.push(frame) + ) + const page = readOpenCodeTranscriptPage({ dbPath: state.path, sessionId: 'session', limit: 1 })! + const messages = page.items.map((item) => item.message) + messages[0].blocks = [{ type: 'text', text: 'x'.repeat(100_000) }] + const watcher = state.watcher + if (!watcher) { + throw new Error('Expected a native watcher') + } + watcher.onInitialSnapshot?.(messages, page.hasMore, page.beforeMessageRowId ?? 0) + watcher.onReplace?.(messages, page.hasMore, page.beforeMessageRowId ?? 0) + expect(frames).toHaveLength(2) + for (const frame of frames) { + const record = asRecord(frame) + const published = Array.isArray(record?.messages) ? record.messages : [] + expect(published.map((message) => asRecord(message)?.role)).toEqual([ + 'reasoning', + 'assistant' + ]) + expect(record?.beforeOffset).toBe(page.beforeMessageRowId) + expect(JSON.stringify(published).length).toBeLessThan(100_000) + } + expect(messages[0].blocks[0]).toEqual({ type: 'text', text: 'x'.repeat(100_000) }) + }) +}) diff --git a/src/main/runtime/rpc/methods/native-chat-opencode-wire-page-budget.test.ts b/src/main/runtime/rpc/methods/native-chat-opencode-wire-page-budget.test.ts new file mode 100644 index 000000000000..ec9d872d5d34 --- /dev/null +++ b/src/main/runtime/rpc/methods/native-chat-opencode-wire-page-budget.test.ts @@ -0,0 +1,333 @@ +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { createHash } from 'node:crypto' +import { afterEach, describe, expect, it, vi } from 'vitest' +import Database from '../../../sqlite/sync-database' +import { asRecord } from '../../../ai-vault/session-scanner-values' +import { readOpenCodeNativeChatTranscriptTail } from '../../../native-chat/transcript-opencode' +import { readOpenCodeTranscriptPage } from '../../../native-chat/transcript-opencode-sqlite-query' +import type { SubscribeNativeChatTranscriptArgs } from '../../../native-chat/transcript-watch-contract' +import { createDispatcherStreamingFeatureEmitter } from '../dispatcher-streaming-feature-emitter' +import { mobileE2EETextPayloadAdmissionBytes } from '../mobile-e2ee-outbound-admission' +import { REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES } from '../../../../shared/remote-runtime-memory-limits' +import { buildRegistry, isStreamingMethod, type RpcContext } from '../core' + +const state: { path: string; watcher: SubscribeNativeChatTranscriptArgs | null } = vi.hoisted( + () => ({ path: '', watcher: null }) +) +vi.mock('../../../managed-data-accounts/service', () => ({ getManagedDataAccountService: vi.fn() })) +vi.mock('../../../native-chat/transcript-watch', () => ({ + readNativeChatTranscriptTail: (args: { + sessionId: string + limit: number + beforeOffset?: number + }) => + readOpenCodeNativeChatTranscriptTail(args, { + resolveDbPath: async () => state.path, + readPage: async (page) => readOpenCodeTranscriptPage(page) + }), + subscribeNativeChatTranscript: async (args: SubscribeNativeChatTranscriptArgs) => { + state.watcher = args + return { watching: true, unsubscribe: vi.fn() } + } +})) +import { NATIVE_CHAT_METHODS } from './native-chat' +import { + boundNativeChatRpcPageByBytes, + nativeChatRpcAppendBatches +} from './native-chat-rpc-page-bounds' +import type { NativeChatMessage } from '../../../../shared/native-chat-types' + +const roots: string[] = [] +afterEach(() => { + state.watcher = null + for (const root of roots.splice(0)) { + rmSync(root, { recursive: true, force: true }) + } +}) + +function denseFixture(version: 'v1' | 'v2'): string { + const root = mkdtempSync(join(tmpdir(), 'orca-opencode-wire-budget-')) + roots.push(root) + const path = join(root, 'opencode.db') + state.path = path + const db = new Database(path) + if (version === 'v2') { + db.exec(`CREATE TABLE session_v2 (id TEXT PRIMARY KEY); + CREATE TABLE session_message (id TEXT, session_id TEXT, type TEXT, seq INTEGER, + data TEXT, time_created INTEGER, time_updated INTEGER); + INSERT INTO session_v2 VALUES ('session');`) + } else { + db.exec(`CREATE TABLE session (id TEXT PRIMARY KEY); + CREATE TABLE message (id TEXT, session_id TEXT, time_created INTEGER, time_updated INTEGER, data TEXT); + CREATE TABLE part (id TEXT, message_id TEXT, session_id TEXT, time_updated INTEGER, data TEXT); + INSERT INTO session VALUES ('session');`) + } + db.exec('BEGIN') + const insert = (id: string, role: 'user' | 'assistant', cursor: number, content: unknown[]) => { + if (version === 'v2') { + db.prepare('INSERT INTO session_message VALUES (?, ?, ?, ?, ?, ?, ?)').run( + id, + 'session', + role, + cursor, + JSON.stringify({ content }), + cursor, + cursor + ) + } else { + db.prepare('INSERT INTO message VALUES (?, ?, ?, ?, ?)').run( + id, + 'session', + cursor, + cursor, + JSON.stringify({ role }) + ) + for (const [index, part] of content.entries()) { + db.prepare('INSERT INTO part VALUES (?, ?, ?, ?, ?)').run( + `${id}:${index}`, + id, + 'session', + cursor, + JSON.stringify(part) + ) + } + } + } + insert('oldest', 'user', 1, [{ type: 'text', text: 'Earlier history' }]) + for (let index = 1; index <= 1200; index++) { + insert(String(index), 'assistant', index * 7, [ + { type: 'reasoning', text: 'Я'.repeat(900) }, + { type: 'text', text: 'Я'.repeat(900) } + ]) + } + insert('latest', 'user', 1201 * 7, [{ type: 'text', text: 'Я'.repeat(900) }]) + db.exec('COMMIT') + db.close() + return path +} + +function context(clientKind: RpcContext['clientKind'] = 'mobile'): RpcContext { + const runtime = { registerSubscriptionCleanup: vi.fn(), cleanupSubscription: vi.fn() } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: Native-chat only uses these subscription cleanup methods; its feature emitter records no interaction. + return { runtime: runtime as unknown as RpcContext['runtime'], clientKind } +} + +describe.each(['v1', 'v2'] as const)('%s byte-bounded semantic pages', (version) => { + it.each(['mobile', 'runtime'] as const)( + 'keeps an actual serialized %s replacement within E2EE admission and pages all groups', + async (clientKind) => { + const dbPath = denseFixture(version) + const page = readOpenCodeTranscriptPage({ dbPath, sessionId: 'session', limit: 2400 })! + expect(page.items).toHaveLength(2401) + expect(page.items.every((item) => Number.isFinite(item.message.transcriptOffset))).toBe(true) + expect(page.items.every((item) => item.message.transcriptOffset === item.rowid)).toBe(true) + const method = buildRegistry(NATIVE_CHAT_METHODS).get('nativeChat.subscribe')! + if (!isStreamingMethod(method)) { + throw new Error('Expected a subscription') + } + const ctx = context(clientKind) + const replies: string[] = [] + const request = { id: 'budget-control', authToken: 'fixture', method: 'nativeChat.subscribe' } + const emitter = createDispatcherStreamingFeatureEmitter( + ctx.runtime, + request, + { runtimeId: 'private-budget-control' }, + (reply) => replies.push(reply) + ) + await method.handler( + method.params?.parse({ agent: 'opencode2', sessionId: 'session', limit: 2000 }), + ctx, + emitter.emit + ) + const watcher = state.watcher + if (!watcher) { + throw new Error('Expected a native watcher') + } + watcher.onReplace?.( + page.items.map((item) => item.message), + page.hasMore, + page.beforeMessageRowId ?? 0 + ) + const reply = replies[0] + const admission = mobileE2EETextPayloadAdmissionBytes(reply) + const response = asRecord(JSON.parse(reply)) + const result = asRecord(response?.result) + const messages = Array.isArray(result?.messages) ? result.messages : [] + const receipt = process.env.ORCA_NATIVE_CHAT_BUDGET_RECEIPT + if (receipt && version === 'v2' && clientKind === 'mobile') { + writeFileSync(`${receipt}.payload.json`, reply) + writeFileSync( + receipt, + `${JSON.stringify( + { + source: process.env.ORCA_NATIVE_CHAT_BUDGET_SOURCE, + kind: 'Controlled synthetic SQLite stress; actual production parser/RPC serializer/E2EE admission, no genuine CLI claim', + inputMessages: page.items.length, + outputMessages: messages.length, + serializedBytes: Buffer.byteLength(reply), + sha256: createHash('sha256').update(reply).digest('hex'), + maximumBytes: REMOTE_RUNTIME_MAX_OUTBOUND_JSON_BYTES, + admission: Number.isFinite(admission) ? admission : 'Infinity', + hasMore: result?.hasMore, + beforeOffset: result?.beforeOffset, + first: messages[0], + last: messages.at(-1) + }, + null, + 2 + )}\n` + ) + } + expect(Number.isFinite(admission)).toBe(true) + expect(messages.length).toBeLessThan(page.items.length) + expect(result?.hasMore).toBe(true) + const first = asRecord(messages[0]) + expect(result?.beforeOffset).toBe(first?.transcriptOffset) + expect(messages.map((message) => asRecord(message)?.role).at(0)).toBe('reasoning') + expect(messages.map((message) => asRecord(message)?.role).at(1)).toBe('assistant') + expect(new Set(messages.map((message) => asRecord(message)?.id)).size).toBe(messages.length) + const read = buildRegistry(NATIVE_CHAT_METHODS).get('nativeChat.readSession')! + if (isStreamingMethod(read)) { + throw new Error('Expected a page read') + } + const pages = [messages] + let cursor = typeof result?.beforeOffset === 'number' ? result.beforeOffset : undefined + let hasMore = result?.hasMore === true + while (hasMore) { + const earlier = asRecord( + await read.handler( + read.params?.parse({ + agent: 'opencode', + sessionId: 'session', + limit: 2000, + beforeOffset: cursor + }), + ctx + ) + ) + const entries = Array.isArray(earlier?.messages) ? earlier.messages : [] + expect(entries.length).toBeGreaterThan(0) + const before = typeof earlier?.beforeOffset === 'number' ? earlier.beforeOffset : undefined + expect(before).toBeLessThan(cursor!) + cursor = before + hasMore = earlier?.hasMore === true + pages.unshift(entries) + } + const all = pages.flat() + const ids = all.map((message) => asRecord(message)?.id) + expect(all).toHaveLength(2402) + expect(new Set(ids).size).toBe(2402) + const prefix = version === 'v2' ? 'opencode:' : '' + expect(ids).toEqual([ + `${prefix}oldest`, + ...Array.from({ length: 1200 }, (_, index) => [ + `${prefix}${index + 1}:reasoning`, + `${prefix}${index + 1}` + ]).flat(), + `${prefix}latest` + ]) + } + ) +}) + +function message(id: string, offset: number, text: string): NativeChatMessage { + return { + id, + role: 'assistant', + source: 'transcript', + timestamp: 10, + transcriptOffset: offset, + blocks: [{ type: 'text', text }] + } +} + +it('replaces an unsendable newest raw group with one stable localized notice', () => { + const large = [ + message('row:reasoning', 42, 'Я'.repeat(1_000_000)), + message('row', 42, 'Я'.repeat(1_000_000)) + ] + large[0].role = 'reasoning' + const page = boundNativeChatRpcPageByBytes([message('older', 2, 'older'), ...large], false, 2) + expect(page).toEqual({ + messages: [ + { + id: 'row', + role: 'system', + source: 'transcript', + timestamp: 10, + transcriptOffset: 42, + blocks: [ + { + type: 'text', + text: 'This part of the chat was too large to show.', + presentation: 'history-item-too-large' + } + ] + } + ], + hasMore: true, + beforeOffset: 42 + }) + expect(Number.isFinite(mobileE2EETextPayloadAdmissionBytes(JSON.stringify(page)))).toBe(true) + const only = boundNativeChatRpcPageByBytes(large, false, 42) + expect(only.hasMore).toBe(false) + expect(only.beforeOffset).toBe(42) +}) + +it('keeps append batches complete and admits every serialized batch', () => { + const messages = Array.from({ length: 12 }, (_, index) => [ + message(`${index}:reasoning`, index, 'Я'.repeat(100_000)), + message(String(index), index, 'Я'.repeat(100_000)) + ]).flat() + for (let index = 0; index < messages.length; index += 2) { + messages[index].role = 'reasoning' + } + const batches = nativeChatRpcAppendBatches(messages) + expect(batches.length).toBeGreaterThan(1) + expect(batches.flat()).toEqual(messages) + for (const batch of batches) { + expect(batch[0].role).toBe('reasoning') + expect(batch.at(-1)?.role).toBe('assistant') + expect( + Number.isFinite( + mobileE2EETextPayloadAdmissionBytes(JSON.stringify({ type: 'appended', messages: batch })) + ) + ).toBe(true) + } +}) + +it('does not manufacture a paging cursor for legacy messages without optional offsets', () => { + const messages = [ + message('old', 1, 'Я'.repeat(700_000)), + message('new', 2, 'Я'.repeat(700_000)) + ].map(({ transcriptOffset: _offset, ...row }) => row) + const result = boundNativeChatRpcPageByBytes(messages, true, 90) + expect(result.hasMore).toBe(true) + expect(result.beforeOffset).toBe(90) + expect(result.messages).toHaveLength(1) + expect(result.messages[0].role).toBe('system') + expect(result.messages[0].blocks[0]).toMatchObject({ presentation: 'history-item-too-large' }) +}) + +it('preserves empty append lifecycle frames', async () => { + const method = buildRegistry(NATIVE_CHAT_METHODS).get('nativeChat.subscribe')! + if (!isStreamingMethod(method)) { + throw new Error('Expected a subscription') + } + const frames: unknown[] = [] + await method.handler( + method.params?.parse({ agent: 'opencode2', sessionId: 'session' }), + context(), + (frame) => frames.push(frame) + ) + state.watcher?.onAppend([], { state: 'completed', turnId: 'turn', timestamp: 10 }) + expect(frames).toEqual([ + { + type: 'appended', + messages: [], + lifecycle: { state: 'completed', turnId: 'turn', timestamp: 10 } + } + ]) +}) diff --git a/src/main/runtime/rpc/methods/native-chat-rpc-page-bounds.ts b/src/main/runtime/rpc/methods/native-chat-rpc-page-bounds.ts new file mode 100644 index 000000000000..1ebdf1056dc1 --- /dev/null +++ b/src/main/runtime/rpc/methods/native-chat-rpc-page-bounds.ts @@ -0,0 +1,100 @@ +import { + nativeChatMessagesShareTranscriptRow, + nativeChatSemanticRowId, + type NativeChatMessage +} from '../../../../shared/native-chat-types' +import { agentSessionHostStatusBody } from '../../../../shared/agent-session-host-status-rows' +import { boundJournalKeyComponent } from '../../../../shared/agent-session-journal-item-key' +import { HISTORY_PAGE_CONTENT_BUDGET_BYTES } from '../../../native-chat/agent-session-wire/agent-session-history-page-bounds' + +function oversizedRow(message: NativeChatMessage): NativeChatMessage { + const { text, presentation } = agentSessionHostStatusBody('history-item-too-large') + return { + id: boundJournalKeyComponent(nativeChatSemanticRowId(message)), + role: 'system', + blocks: [{ type: 'text', text, presentation }], + timestamp: message.timestamp, + source: message.source, + ...(message.transcriptOffset !== undefined + ? { transcriptOffset: message.transcriptOffset } + : {}) + } +} + +function groupBytes(group: readonly NativeChatMessage[]): number { + return group.reduce((bytes, message) => bytes + Buffer.byteLength(JSON.stringify(message)) + 1, 0) +} + +function rowGroups(messages: readonly NativeChatMessage[]): NativeChatMessage[][] { + const groups: NativeChatMessage[][] = [] + for (const message of messages) { + const last = groups.at(-1) + if (last && nativeChatMessagesShareTranscriptRow(last.at(-1)!, message)) { + last.push(message) + } else { + groups.push([message]) + } + } + return groups +} + +export function boundNativeChatRpcPageByBytes( + messages: readonly NativeChatMessage[], + hasMore: boolean, + beforeOffset: number +): { messages: NativeChatMessage[]; hasMore: boolean; beforeOffset: number } { + const groups = rowGroups(messages) + const kept: NativeChatMessage[][] = [] + let bytes = 2 + let retained = 0 + for (const group of groups.toReversed()) { + const size = groupBytes(group) + if (kept.length === 0 && size > HISTORY_PAGE_CONTENT_BUDGET_BYTES) { + kept.push([oversizedRow(group[0])]) + retained += group.length + break + } + if (bytes + size > HISTORY_PAGE_CONTENT_BUDGET_BYTES) { + break + } + kept.push(group) + retained += group.length + bytes += size + } + const selected = kept.toReversed().flat() + const dropped = retained < messages.length + const cursor = selected[0]?.transcriptOffset + // A legacy projection has no row cursor: explicitly omit the page rather than skip unseen history. + if (dropped && cursor === undefined) { + return { messages: messages.length ? [oversizedRow(messages[0])] : [], hasMore, beforeOffset } + } + return { + messages: selected, + hasMore: hasMore || dropped, + beforeOffset: dropped ? (cursor ?? beforeOffset) : beforeOffset + } +} + +export function nativeChatRpcAppendBatches( + messages: readonly NativeChatMessage[] +): NativeChatMessage[][] { + const batches: NativeChatMessage[][] = [] + let batch: NativeChatMessage[] = [] + let bytes = 2 + for (const group of rowGroups(messages)) { + const bounded = + groupBytes(group) > HISTORY_PAGE_CONTENT_BUDGET_BYTES ? [oversizedRow(group[0])] : group + const size = groupBytes(bounded) + if (batch.length > 0 && bytes + size > HISTORY_PAGE_CONTENT_BUDGET_BYTES) { + batches.push(batch) + batch = [] + bytes = 2 + } + batch.push(...bounded) + bytes += size + } + if (batch.length > 0) { + batches.push(batch) + } + return batches +} diff --git a/src/main/runtime/rpc/methods/native-chat.ts b/src/main/runtime/rpc/methods/native-chat.ts index 05f8f7f87261..99c0178ea0cf 100644 --- a/src/main/runtime/rpc/methods/native-chat.ts +++ b/src/main/runtime/rpc/methods/native-chat.ts @@ -1,4 +1,5 @@ import type { NativeChatMessage } from '../../../../shared/native-chat-types' +import { resolveNativeChatTranscriptAgent } from '../../../../shared/native-chat-agent-support' import { readNativeChatTranscriptTail, subscribeNativeChatTranscript, @@ -7,6 +8,10 @@ import { } from '../../../native-chat/transcript-watch' import { defineMethod, defineStreamingMethod, type RpcContext } from '../core' import { sanitizeNativeChatRpcBlock } from './native-chat-rpc-block-sanitize' +import { + boundNativeChatRpcPageByBytes, + nativeChatRpcAppendBatches +} from './native-chat-rpc-page-bounds' import { MOBILE_NATIVE_CHAT_MAX_WINDOW, NativeChatSession, @@ -51,16 +56,21 @@ function windowTranscript( return messages.length > window ? messages.slice(-window) : messages.slice() } -/** Apply the windowed slice and keep inline image bytes off every RPC transport. - * Mobile clients additionally receive bounded text and tool bodies; runtime - * clients keep those bodies intact. */ -function windowForClient( +function pageForClient( messages: readonly NativeChatMessage[], + hasMore: boolean, + beforeOffset: number, clientKind: RpcContext['clientKind'], - limit = MOBILE_NATIVE_CHAT_DEFAULT_WINDOW -): NativeChatMessage[] { - const windowed = windowTranscript(messages, limit) - return windowed.map((message) => sanitizeMessage(message, clientKind)) + limit = MOBILE_NATIVE_CHAT_DEFAULT_WINDOW, + agent?: string +): { messages: NativeChatMessage[]; hasMore: boolean; beforeOffset: number } { + const isOpenCode = resolveNativeChatTranscriptAgent(agent) === 'opencode' + const sanitized = (isOpenCode ? messages : windowTranscript(messages, limit)).map((message) => + sanitizeMessage(message, clientKind) + ) + return isOpenCode + ? boundNativeChatRpcPageByBytes(sanitized, hasMore, beforeOffset) + : { messages: sanitized, hasMore, beforeOffset } } export const NATIVE_CHAT_METHODS = [ @@ -81,9 +91,14 @@ export const NATIVE_CHAT_METHODS = [ ) return 'messages' in result ? { - messages: windowForClient(result.messages, clientKind, limit), - hasMore: result.hasMore, - beforeOffset: result.beforeOffset, + ...pageForClient( + result.messages, + result.hasMore, + result.beforeOffset, + clientKind, + limit, + params.agent + ), ...(result.lifecycle ? { lifecycle: result.lifecycle } : {}) } : result @@ -144,9 +159,7 @@ export const NATIVE_CHAT_METHODS = [ // instead of stranding the view at 'loading' when the read keeps throwing. emit({ type: 'snapshot', - messages: windowForClient(messages, clientKind, limit), - hasMore, - beforeOffset, + ...pageForClient(messages, hasMore, beforeOffset, clientKind, limit, params.agent), ...(error ? { error } : {}), ...(lifecycle ? { lifecycle } : {}) }) @@ -166,9 +179,7 @@ export const NATIVE_CHAT_METHODS = [ } emit({ type: 'replacement', - messages: windowForClient(messages, clientKind, limit), - hasMore, - beforeOffset, + ...pageForClient(messages, hasMore, beforeOffset, clientKind, limit, params.agent), ...(lifecycle ? { lifecycle } : {}) }) }, @@ -176,11 +187,18 @@ export const NATIVE_CHAT_METHODS = [ if (closed) { return } - emit({ - type: 'appended', - messages: sanitizeAppendForClient(messages, clientKind), - ...(lifecycle ? { lifecycle } : {}) - }) + const sanitized = sanitizeAppendForClient(messages, clientKind) + const batches = + sanitized.length > 0 && resolveNativeChatTranscriptAgent(params.agent) === 'opencode' + ? nativeChatRpcAppendBatches(sanitized) + : [sanitized] + for (const batch of batches) { + emit({ + type: 'appended', + messages: batch, + ...(lifecycle && batch === batches.at(-1) ? { lifecycle } : {}) + }) + } } } let subscription: NativeChatTranscriptSubscription diff --git a/src/shared/native-chat-merge-pairs.test.ts b/src/shared/native-chat-merge-pairs.test.ts new file mode 100644 index 000000000000..fe59e24334a4 --- /dev/null +++ b/src/shared/native-chat-merge-pairs.test.ts @@ -0,0 +1,114 @@ +import { describe, expect, it } from 'vitest' +import { + applyAppend, + boundNativeChatWindow, + createNativeChatMerger, + replaceList +} from './native-chat-merge' +import type { NativeChatMessage } from './native-chat-types' + +function row(id: string, overrides: Partial = {}): NativeChatMessage { + return { + id, + role: 'assistant', + source: 'transcript', + timestamp: 1, + blocks: [{ type: 'text', text: id }], + ...overrides + } +} + +function pair(id: string): NativeChatMessage[] { + return [row(`${id}:reasoning`, { role: 'reasoning' }), row(id)] +} + +describe('native chat complete-pair windows', () => { + it.each(['msg-provider', 'opencode:msg-provider'])('keeps the cutoff pair for %s', (id) => { + const messages = [row('older'), ...pair(id), row('latest')] + expect(boundNativeChatWindow(messages, 2)).toEqual(messages.slice(1)) + expect(messages.map((message) => message.id)).toEqual([ + 'older', + `${id}:reasoning`, + id, + 'latest' + ]) + }) + + it('keeps the newest complete pair even at limit 1', () => { + const messages = [...pair('older'), ...pair('latest')] + expect(boundNativeChatWindow(messages, 1)).toEqual(messages.slice(2)) + }) + + it('does not group unrelated reasoning, suffixes, or user rows', () => { + const unrelated = [row('thinking', { role: 'reasoning' }), row('answer'), row('latest')] + expect(boundNativeChatWindow(unrelated, 2)).toEqual(unrelated.slice(1)) + const suffixOnly = [row('answer:reasoning'), row('answer'), row('latest')] + expect(boundNativeChatWindow(suffixOnly, 2)).toEqual(suffixOnly.slice(1)) + const user = [row('prompt:reasoning', { role: 'reasoning' }), row('prompt', { role: 'user' })] + expect(boundNativeChatWindow(user, 1)).toEqual(user.slice(1)) + }) + + it('keeps ordering, source precedence and cached indexes after paired trims', () => { + const merger = createNativeChatMerger() + const [reasoning, answer] = pair('opencode:first') + if (!reasoning || !answer) { + throw new Error('Expected a pair') + } + replaceList(merger, [row('older'), reasoning, { ...answer, source: 'hook' }]) + const authoritative = { ...answer, blocks: [{ type: 'text' as const, text: 'final answer' }] } + const newest = row('newest') + expect(applyAppend(merger, [authoritative, newest], 2)).toEqual([ + reasoning, + authoritative, + newest + ]) + expect([...merger.indexById]).toEqual([ + [reasoning.id, 0], + [answer.id, 1], + ['newest', 2] + ]) + expect(applyAppend(merger, [{ ...answer, source: 'scrape' }], 2)[1]).toBe(authoritative) + expect(applyAppend(merger, pair('opencode:next'), 2).map((message) => message.id)).toEqual([ + 'opencode:next:reasoning', + 'opencode:next' + ]) + expect([...merger.indexById]).toEqual([ + ['opencode:next:reasoning', 0], + ['opencode:next', 1] + ]) + }) + + it.each([1, 40, 2000])('stays at most one row above limit %i across long appends', (limit) => { + const merger = createNativeChatMerger() + for (let index = 0; index < limit + 20; index++) { + const messages = applyAppend( + merger, + index % 2 === 0 ? pair(`opencode:${index}`) : [row(`user-${index}`, { role: 'user' })], + limit + ) + expect(messages.length).toBeLessThanOrEqual(limit + 1) + expect(messages[0]?.role).not.toBe('assistant') + expect(merger.indexById.size).toBe(messages.length) + } + for (let at = 0; at < merger.list.length; at++) { + const message = merger.list[at] + expect(merger.indexById.get(message.id)).toBe(at) + if (message.role === 'reasoning') { + expect(merger.list[at + 1]?.id).toBe(message.id.slice(0, -':reasoning'.length)) + } else if (message.role === 'assistant') { + expect(merger.list[at - 1]?.id).toBe(`${message.id}:reasoning`) + } + } + }) +}) + +it('retains reasoning, answer and omission projections sharing an optional raw-row cursor', () => { + const messages = [ + row('older'), + ...pair('mixed').map((message) => ({ ...message, transcriptOffset: 42 })), + row('mixed:omission', { role: 'system', transcriptOffset: 42 }), + row('latest', { transcriptOffset: 43 }) + ] + expect(boundNativeChatWindow(messages, 2)).toEqual(messages.slice(1)) + expect(boundNativeChatWindow(messages, 1)).toEqual(messages.slice(-1)) +}) diff --git a/src/shared/native-chat-merge.ts b/src/shared/native-chat-merge.ts index a2e12ad8302b..310634af4b3a 100644 --- a/src/shared/native-chat-merge.ts +++ b/src/shared/native-chat-merge.ts @@ -1,6 +1,7 @@ // Pure id-dedup and windowing for both desktop and mobile native-chat streams. import { + nativeChatMessagesShareTranscriptRow, NATIVE_CHAT_SOURCE_PRIORITY, type NativeChatMessage, type NativeChatSource @@ -35,10 +36,7 @@ export function mergeNativeChatMessages( return mergeNativeChatMessagesWith(existing, incoming, NATIVE_CHAT_SOURCE_PRIORITY) } -/** Cap a message list to its most-recent `limit` entries. The base read is - * already windowed; this keeps the live-append tail bounded to the same window - * so a long run can't grow the list without limit. A non-positive limit means - * "no cap". Returns the input reference when no trim is needed. */ +/** Keep the recent tail without splitting a provider row. A non-positive limit means no cap. */ export function boundNativeChatWindow( messages: readonly NativeChatMessage[], limit: number @@ -46,7 +44,11 @@ export function boundNativeChatWindow( if (limit <= 0 || messages.length <= limit) { return messages as NativeChatMessage[] } - return messages.slice(messages.length - limit) + let start = messages.length - limit + while (start > 0 && nativeChatMessagesShareTranscriptRow(messages[start - 1], messages[start])) { + start -= 1 + } + return messages.slice(start) } /** Stateful id-dedup merger that caches the id→index map across appends so a diff --git a/src/shared/native-chat-reasoning-order.test.ts b/src/shared/native-chat-reasoning-order.test.ts new file mode 100644 index 000000000000..ea3fd56305b4 --- /dev/null +++ b/src/shared/native-chat-reasoning-order.test.ts @@ -0,0 +1,46 @@ +import { expect, it } from 'vitest' +import type { NativeChatMessage } from './native-chat-types' +import { + compareNativeChatTranscriptMessages, + projectNativeChatTranscriptMessages +} from './native-chat-transcript-projection' + +function row(id: string, role: NativeChatMessage['role'], timestamp = 1): NativeChatMessage { + return { id, role, timestamp, source: 'transcript', blocks: [{ type: 'text', text: id }] } +} + +it('keeps derived reasoning immediately before its answer without changing provider timestamps', () => { + const messages = [ + row('opencode:msg_b', 'assistant'), + row('opencode:msg_a:other', 'assistant'), + row('opencode:msg_a', 'assistant'), + row('opencode:msg_b:reasoning', 'reasoning'), + row('opencode:msg_a:reasoning', 'reasoning') + ] + const sorted = [...messages].sort(compareNativeChatTranscriptMessages) + expect(sorted.map((message) => message.id)).toEqual([ + 'opencode:msg_a:reasoning', + 'opencode:msg_a', + 'opencode:msg_a:other', + 'opencode:msg_b:reasoning', + 'opencode:msg_b' + ]) + expect(projectNativeChatTranscriptMessages(messages)).toEqual(sorted) + expect(messages.every((message) => message.timestamp === 1)).toBe(true) + for (let first = 0; first < sorted.length; first++) { + for (let second = first + 1; second < sorted.length; second++) { + expect(compareNativeChatTranscriptMessages(sorted[first], sorted[second])).toBeLessThan(0) + expect(compareNativeChatTranscriptMessages(sorted[second], sorted[first])).toBeGreaterThan(0) + } + } +}) + +it('keeps timestamp and journal authority ahead of a derived reasoning key', () => { + const answer = row('row', 'assistant', 1) + const laterReasoning = row('row:reasoning', 'reasoning', 2) + expect(compareNativeChatTranscriptMessages(answer, laterReasoning)).toBeLessThan(0) + answer.journalPosition = { sequence: 1, index: 0 } + laterReasoning.journalPosition = { sequence: 2, index: 0 } + laterReasoning.timestamp = 0 + expect(compareNativeChatTranscriptMessages(answer, laterReasoning)).toBeLessThan(0) +}) diff --git a/src/shared/native-chat-transcript-projection.ts b/src/shared/native-chat-transcript-projection.ts index d826ecaf552e..3ab78007edb0 100644 --- a/src/shared/native-chat-transcript-projection.ts +++ b/src/shared/native-chat-transcript-projection.ts @@ -7,7 +7,7 @@ // conversation: each subagent's rows are that subagent's, kept apart and folded on // their own, for the surface that shows them beside the spawn that ran it. -import type { NativeChatMessage } from './native-chat-types' +import { nativeChatSemanticRowId, type NativeChatMessage } from './native-chat-types' import { compareAgentJournalPositions } from './agent-session-journal-position' import { agentJournalItemSubagentId } from './agent-session-journal-producer' import { stripNoiseMessages } from './native-chat-noise' @@ -25,13 +25,18 @@ export function compareNativeChatMessagesByTime( if (at !== bt) { return at - bt } - if (a.id < b.id) { + // Split reasoning shares its provider row's key, before that row's answer. + const aId = nativeChatSemanticRowId(a) + const bId = nativeChatSemanticRowId(b) + const aReasoning = aId !== a.id + const bReasoning = bId !== b.id + if (aId < bId) { return -1 } - if (a.id > b.id) { + if (aId > bId) { return 1 } - return 0 + return Number(bReasoning) - Number(aReasoning) } /** Rows the journal holds read in the journal's own order, never its clock: a diff --git a/src/shared/native-chat-types.ts b/src/shared/native-chat-types.ts index db7b7024cdc1..127a9837910b 100644 --- a/src/shared/native-chat-types.ts +++ b/src/shared/native-chat-types.ts @@ -205,6 +205,8 @@ export type NativeChatMessage = AgentJournalProducerLinkage & { * supply one (e.g. some scrape segments). Null sorts before any timestamp. */ timestamp: number | null source: NativeChatSource + /** Optional provider row cursor; split projections share it for whole-row paging. */ + transcriptOffset?: number /** Optional explicit turn key. When present, two messages with the same * `turnId` are treated as the same turn for dedup regardless of `id`. */ turnId?: string @@ -223,6 +225,28 @@ export type NativeChatMessage = AgentJournalProducerLinkage & { foldedJournalPosition?: AgentJournalPosition } +/** Split reasoning and its answer share the provider's row identity. */ +export function nativeChatSemanticRowId(message: NativeChatMessage): string { + return message.role === 'reasoning' && message.id.endsWith(':reasoning') + ? message.id.slice(0, -':reasoning'.length) + : message.id +} + +/** New hosts expose the cursor; older messages retain their reasoning/answer id convention. */ +export function nativeChatMessagesShareTranscriptRow( + first: NativeChatMessage, + second: NativeChatMessage +): boolean { + if (typeof first.transcriptOffset === 'number' && typeof second.transcriptOffset === 'number') { + return first.transcriptOffset === second.transcriptOffset + } + return ( + first.role === 'reasoning' && + second.role === 'assistant' && + nativeChatSemanticRowId(first) === second.id + ) +} + export const NATIVE_CHAT_TURN_LIFECYCLE_STATES = ['working', 'completed', 'interrupted'] as const export type NativeChatTurnLifecycleState = (typeof NATIVE_CHAT_TURN_LIFECYCLE_STATES)[number]