diff --git a/src/main/codex/codex-structured-item-streams.ts b/src/main/codex/codex-structured-item-streams.ts new file mode 100644 index 00000000000..9eb2204b72b --- /dev/null +++ b/src/main/codex/codex-structured-item-streams.ts @@ -0,0 +1,176 @@ +import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types' +import { + createAgentSessionDeltaCoalescer, + type AgentSessionDeltaCoalescerDeps +} from '../native-chat/agent-session-wire/agent-session-delta-coalescer' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { + codexJournalItem, + codexStreamingJournalItem, + type CodexThreadItem +} from './codex-structured-item-translation' + +const CODEX_ITEM_STREAM_TYPES = { + 'item/agentMessage/delta': 'agentMessage', + 'item/plan/delta': 'plan', + 'item/commandExecution/outputDelta': 'commandExecution', + 'item/fileChange/outputDelta': 'fileChange', + 'item/reasoning/summaryTextDelta': 'reasoning', + 'item/reasoning/textDelta': 'reasoning' +} as const + +const PATCH_UPDATED_METHOD = 'item/fileChange/patchUpdated' +const REASONING_PART_METHOD = 'item/reasoning/summaryPartAdded' +const TERMINAL_INTERACTION_METHOD = 'item/commandExecution/terminalInteraction' + +type CodexItemStreamDeps = { + sink: StructuredAgentSessionEventSink + identityFor: ( + threadId: string, + params: unknown, + item: CodexThreadItem + ) => AgentJournalItemIdentity + coalesceMs?: number + schedule?: AgentSessionDeltaCoalescerDeps['schedule'] +} + +type StreamState = { identity: AgentJournalItemIdentity; item: CodexThreadItem } + +export type CodexStructuredItemStreams = { + track: (threadId: string, item: CodexThreadItem, identity: AgentJournalItemIdentity) => void + handle: (threadId: string, method: string, params: unknown) => boolean + forget: (threadId: string, itemId: string) => void + flush: () => void + dispose: () => void +} + +function readRecord(value: unknown): Record { + return typeof value === 'object' && value !== null ? (value as Record) : {} +} + +function readString(source: Record, key: string): string | null { + const value = source[key] + return typeof value === 'string' && value.length > 0 ? value : null +} + +export function codexStructuredItemKey(threadId: string, itemId: string): string { + return `${encodeURIComponent(threadId)}:${encodeURIComponent(itemId)}` +} + +export function createCodexStructuredItemStreams( + deps: CodexItemStreamDeps +): CodexStructuredItemStreams { + const states = new Map() + const latestText = new Map() + const checkpointLengths = new Map() + + const append = (state: StreamState, text: string): void => { + const translated = codexStreamingJournalItem(state.item, text) + if (!translated.body) { + return + } + deps.sink.appendItem(state.identity, translated.body, translated.blobs) + deps.sink.publish() + } + + const persist = (key: string, text: string, force: boolean): void => { + latestText.set(key, text) + const checkpointLength = checkpointLengths.get(key) ?? 0 + const nextLength = Math.max(checkpointLength + 32, Math.ceil(checkpointLength * 1.125)) + if (!force && checkpointLength > 0 && text.length < nextLength) { + return + } + checkpointLengths.set(key, text.length) + const state = states.get(key) + if (state) { + append(state, text) + } + } + + const coalescer = createAgentSessionDeltaCoalescer({ + windowMs: deps.coalesceMs, + schedule: deps.schedule, + emit: (key, text) => persist(key, text, false) + }) + + const ensureState = ( + threadId: string, + itemId: string, + type: string, + params: unknown + ): StreamState => { + const key = codexStructuredItemKey(threadId, itemId) + const existing = states.get(key) + if (existing) { + return existing + } + const item = { type, id: itemId } + const state = { item, identity: deps.identityFor(threadId, params, item) } + states.set(key, state) + return state + } + + const flush = (): void => { + coalescer.flushAll() + for (const [key, text] of latestText) { + if (checkpointLengths.get(key) !== text.length) { + persist(key, text, true) + } + } + } + + return { + track: (threadId, item, identity) => { + states.set(codexStructuredItemKey(threadId, item.id), { item, identity }) + }, + handle: (threadId, method, params) => { + const paramsRecord = readRecord(params) + const itemId = readString(paramsRecord, 'itemId') + if (method === PATCH_UPDATED_METHOD) { + if (!itemId || !Array.isArray(paramsRecord.changes)) { + return true + } + const key = codexStructuredItemKey(threadId, itemId) + coalescer.flush(key) + const state = ensureState(threadId, itemId, 'fileChange', params) + state.item = { ...state.item, changes: paramsRecord.changes } + const translated = codexJournalItem(state.item) + if (translated.body) { + deps.sink.appendItem(state.identity, translated.body, translated.blobs) + deps.sink.publish() + } + return true + } + if (method === TERMINAL_INTERACTION_METHOD) { + return true + } + const type = CODEX_ITEM_STREAM_TYPES[method as keyof typeof CODEX_ITEM_STREAM_TYPES] + if (!type && method !== REASONING_PART_METHOD) { + return false + } + if (!itemId) { + return true + } + const state = ensureState(threadId, itemId, type ?? 'reasoning', params) + const delta = method === REASONING_PART_METHOD ? '\n' : paramsRecord.delta + if (typeof delta === 'string') { + coalescer.append(codexStructuredItemKey(threadId, state.item.id), delta) + } + return true + }, + forget: (threadId, itemId) => { + const key = codexStructuredItemKey(threadId, itemId) + coalescer.forget(key) + states.delete(key) + latestText.delete(key) + checkpointLengths.delete(key) + }, + flush, + dispose: () => { + coalescer.dispose() + states.clear() + latestText.clear() + checkpointLengths.clear() + } + } +} diff --git a/src/main/codex/codex-structured-item-translation.test.ts b/src/main/codex/codex-structured-item-translation.test.ts index 36b811fba7e..44266204405 100644 --- a/src/main/codex/codex-structured-item-translation.test.ts +++ b/src/main/codex/codex-structured-item-translation.test.ts @@ -171,6 +171,24 @@ describe('codex item bodies', () => { ).toMatchObject({ state: 'failed' }) }) + it('maps file changes to one bounded diff item', () => { + expect( + codexItemBody({ + type: 'fileChange', + id: 'patch-1', + status: 'completed', + changes: [ + { path: 'src/a.ts', diff: '@@ a @@' }, + { path: 'src/b.ts', diff: '@@ b @@' } + ] + }) + ).toMatchObject({ + kind: 'diff', + path: '2 files', + patch: { head: '@@ a @@\n@@ b @@', truncated: false } + }) + }) + it('renders reasoning as status and exposes an unknown item as a provider frame', () => { expect(codexItemBody({ type: 'reasoning', id: 'r', text: 'thinking' })).toEqual({ kind: 'status', diff --git a/src/main/codex/codex-structured-item-translation.ts b/src/main/codex/codex-structured-item-translation.ts index abdd9b75864..31e64334acb 100644 --- a/src/main/codex/codex-structured-item-translation.ts +++ b/src/main/codex/codex-structured-item-translation.ts @@ -74,6 +74,10 @@ export class CodexTurnOrdinals { } } +function readRecord(value: unknown): Record { + return typeof value === 'object' && value !== null ? (value as Record) : {} +} + /** * Durable identity for a Codex item, or null for one that has none. * @@ -196,6 +200,40 @@ function commandItem(item: CodexThreadItem): CodexJournalItem { } } +function fileChangeItem(item: CodexThreadItem): CodexJournalItem { + const changes = Array.isArray(item.changes) + ? item.changes.flatMap((change) => { + const record = typeof change === 'object' && change !== null ? readRecord(change) : {} + const path = readString(record, 'path') + const diff = readString(record, 'diff') + return path && diff ? [{ path, diff }] : [] + }) + : [] + if (changes.length === 0) { + return { + body: { + kind: 'tool-call', + name: 'apply_patch', + input: { changes: item.changes ?? null }, + state: commandState(item) + }, + blobs: [], + handled: true + } + } + const patch = changes.map((change) => change.diff).join('\n') + const bounded = boundInlineText(patch, DEFAULT_JOURNAL_PAYLOAD_LIMITS).bounded + return { + body: { + kind: 'diff', + path: changes.length === 1 ? changes[0]!.path : `${changes.length} files`, + patch: bounded + }, + blobs: bounded.truncated ? [{ digest: bounded.digest, payload: patch }] : [], + handled: true + } +} + /** * Journal body for a Codex item, or null for one with nothing to render. * @@ -218,18 +256,9 @@ export function codexJournalItem(item: CodexThreadItem): CodexJournalItem { return commandItem(item) } if (item.type === 'fileChange') { - return { - body: { - kind: 'tool-call', - name: 'apply_patch', - input: { changes: item.changes ?? null }, - state: commandState(item) - }, - blobs: [], - handled: true - } + return fileChangeItem(item) } - if (item.type === 'reasoning') { + if (item.type === 'reasoning' || item.type === 'plan') { const text = readTextContent(item, 'text') ?? readTextContent(item, 'summary') ?? @@ -255,3 +284,26 @@ export function codexItemBody(item: CodexThreadItem): AgentJournalItemBody | nul export function codexStreamingMessageBody(text: string): AgentJournalItemBody { return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text }] } } + +/** Snapshot body for any item-level stream, keyed onto its parent item. */ +export function codexStreamingJournalItem(item: CodexThreadItem, text: string): CodexJournalItem { + if (item.type === 'agentMessage') { + return { body: codexStreamingMessageBody(text), blobs: [], handled: true } + } + if (item.type === 'commandExecution') { + return commandItem({ ...item, aggregatedOutput: text }) + } + if (item.type === 'fileChange') { + const path = Array.isArray(item.changes) + ? readString(readRecord(item.changes[0]), 'path') + : null + const bounded = boundInlineText(text, DEFAULT_JOURNAL_PAYLOAD_LIMITS).bounded + return { + body: { kind: 'diff', path: path ?? 'pending patch', patch: bounded }, + blobs: bounded.truncated ? [{ digest: bounded.digest, payload: text }] : [], + handled: true + } + } + const bounded = boundInlineText(text, DEFAULT_JOURNAL_PAYLOAD_LIMITS) + return { body: { kind: 'status', text: bounded.text }, blobs: [], handled: true } +} diff --git a/src/main/codex/codex-structured-journal-translation.test.ts b/src/main/codex/codex-structured-journal-translation.test.ts index 8b33c9f4b9e..b27eafec7fb 100644 --- a/src/main/codex/codex-structured-journal-translation.test.ts +++ b/src/main/codex/codex-structured-journal-translation.test.ts @@ -6,7 +6,10 @@ import type { import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { projectStructuredItemsToNativeChat } from '../../shared/structured-agent-session-projection' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' -import { createCodexJournalTranslator } from './codex-structured-journal-translation' +import { + createCodexJournalTranslator, + MAX_CODEX_GENERIC_ROWS_PER_TURN +} from './codex-structured-journal-translation' import { CODEX_COMMAND_APPROVAL_METHOD, CODEX_USER_INPUT_METHOD @@ -388,6 +391,66 @@ describe('codex journal translation', () => { }) }) + it('folds long-running command output into one exec item and zero generic rows', () => { + const { translator, tap, window } = translatorWith() + translator.handle(TURN_STARTED) + translator.handle( + notification('item/started', { + item: { type: 'commandExecution', id: 'exec-1', command: 'long-task', status: 'inProgress' } + }) + ) + + for (let index = 0; index < 512; index += 1) { + translator.handle( + notification('item/commandExecution/outputDelta', { itemId: 'exec-1', delta: 'x' }) + ) + window.fire() + } + translator.flush() + + expect(new Set(tap.rows.map((row) => row.key))).toEqual( + new Set(['orca:codex-item%3Athread-abc%3Aexec-1']) + ) + expect(tap.rows.every((row) => row.body.kind === 'tool-call')).toBe(true) + expect(tap.rows.length).toBeLessThan(40) + expect(tap.rows.at(-1)?.body).toMatchObject({ + kind: 'tool-call', + output: { head: 'x'.repeat(512) } + }) + }) + + it('folds reasoning and patch streams into their parent rows', () => { + const { translator, tap, window } = translatorWith() + translator.handle(TURN_STARTED) + translator.handle(notification('item/started', { item: { type: 'reasoning', id: 'r-1' } })) + translator.handle( + notification('item/reasoning/summaryTextDelta', { itemId: 'r-1', delta: 'thinking' }) + ) + translator.handle( + notification('item/started', { + item: { type: 'fileChange', id: 'patch-1', changes: [], status: 'inProgress' } + }) + ) + translator.handle( + notification('item/fileChange/patchUpdated', { + itemId: 'patch-1', + changes: [{ path: 'src/app.ts', kind: { type: 'update' }, diff: '@@ -1 +1 @@' }] + }) + ) + window.fire() + + const reduced = new Map(tap.rows.map((row) => [row.key, row.body])) + expect(reduced.get('orca:codex-item%3Athread-abc%3Ar-1')).toEqual({ + kind: 'status', + text: 'thinking' + }) + expect(reduced.get('orca:codex-item%3Athread-abc%3Apatch-1')).toMatchObject({ + kind: 'diff', + path: 'src/app.ts', + patch: { head: '@@ -1 +1 @@' } + }) + }) + it('publishes after every write so a subscriber never trails the journal', () => { const { translator, tap } = translatorWith() @@ -399,7 +462,7 @@ describe('codex journal translation', () => { expect(tap.publishes()).toBe(1) }) - it('journals malformed item events and deltas instead of hiding them', () => { + it('journals malformed item events but never malformed deltas', () => { const { translator, tap, window } = translatorWith() translator.handle(TURN_STARTED) @@ -411,10 +474,6 @@ describe('codex journal translation', () => { expect.objectContaining({ kind: 'status', providerFrame: expect.objectContaining({ kind: 'notification:item/completed' }) - }), - expect.objectContaining({ - kind: 'status', - providerFrame: expect.objectContaining({ kind: 'notification:item/agentMessage/delta' }) }) ]) }) @@ -443,6 +502,28 @@ describe('codex journal translation', () => { ).toEqual(['notification:future/notification', 'request:future/request', 'frame:unclassified']) }) + it('bounds generic rows per turn while preserving unknown non-delta evidence', () => { + const { translator, tap } = translatorWith() + translator.handle(TURN_STARTED) + for (let index = 0; index < MAX_CODEX_GENERIC_ROWS_PER_TURN + 20; index += 1) { + translator.handle(notification('future/notification', { value: index })) + } + translator.handle(notification('item/future/outputDelta', { itemId: 'future', delta: 'x' })) + + expect(tap.rows).toHaveLength(MAX_CODEX_GENERIC_ROWS_PER_TURN) + expect(tap.rows[0]?.body).toMatchObject({ + kind: 'status', + providerFrame: { kind: 'notification:future/notification' } + }) + expect( + tap.rows.some( + (row) => + row.body.kind === 'status' && + row.body.providerFrame?.kind === 'notification:item/future/outputDelta' + ) + ).toBe(false) + }) + it('keeps a fresh session timeline empty through startup and status notifications', () => { const { translator, tap } = translatorWith() diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index dafa86af4eb..fa227901765 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -1,19 +1,19 @@ import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types' import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' -import { - createAgentSessionDeltaCoalescer, - type AgentSessionDeltaCoalescerDeps -} from '../native-chat/agent-session-wire/agent-session-delta-coalescer' +import type { AgentSessionDeltaCoalescerDeps } from '../native-chat/agent-session-wire/agent-session-delta-coalescer' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { unhandledProviderFrameJournalItem } from '../native-chat/agent-session-wire/unhandled-provider-frame' import type { CodexStructuredSessionEvent } from './codex-structured-session-adapter' import { codexItemIdentity, codexJournalItem, - codexStreamingMessageBody, CodexTurnOrdinals, readCodexThreadItem } from './codex-structured-item-translation' +import { + codexStructuredItemKey, + createCodexStructuredItemStreams +} from './codex-structured-item-streams' import { codexApprovalItem, codexPromptIdentity, @@ -29,7 +29,7 @@ import { readCodexTurnId } from './codex-structured-thread-facts' // per-session and per-acquisition — a new lease gets a new translator and a new // sink, so a superseded child cannot keep writing. -const AGENT_MESSAGE_DELTA_METHOD = 'item/agentMessage/delta' +export const MAX_CODEX_GENERIC_ROWS_PER_TURN = 8 export type CodexJournalTranslatorDeps = { sink: StructuredAgentSessionEventSink @@ -65,15 +65,21 @@ export function createCodexJournalTranslator( /** What each announced item is, so an approval can name what it approves. */ const details = new Map() const currentTurnIds = new Map() - const latestStreamText = new Map() - const checkpointLengths = new Map() + const genericRowsByTurn = new Map() let fallbackSequence = 0 - const appendUnhandled = (kind: string, payload: unknown): void => { + const appendUnhandled = (kind: string, payload: unknown, threadId = 'session'): void => { const translated = unhandledProviderFrameJournalItem('codex', kind, payload) if (!translated) { return } + const turnId = readCodexTurnId(payload) ?? currentTurnIds.get(threadId) ?? 'outside-turn' + const bucket = `${encodeURIComponent(threadId)}:${encodeURIComponent(turnId)}` + const rowCount = genericRowsByTurn.get(bucket) ?? 0 + if (rowCount >= MAX_CODEX_GENERIC_ROWS_PER_TURN) { + return + } + genericRowsByTurn.set(bucket, rowCount + 1) fallbackSequence += 1 deps.sink.appendItem( { provider: 'orca', clientMessageId: `provider-frame:codex:${fallbackSequence}` }, @@ -110,52 +116,12 @@ export function createCodexJournalTranslator( deps.sink.publish() } - const itemKey = (threadId: string, codexItemId: string): string => - `${encodeURIComponent(threadId)}:${encodeURIComponent(codexItemId)}` - - const persistStream = (key: string, text: string, force: boolean): void => { - latestStreamText.set(key, text) - const checkpointLength = checkpointLengths.get(key) ?? 0 - const nextLength = Math.max(checkpointLength + 32, Math.ceil(checkpointLength * 1.125)) - if (!force && checkpointLength > 0 && text.length < nextLength) { - return - } - checkpointLengths.set(key, text.length) - const identity = identities.get(key) - if (!identity) { - return - } - deps.sink.appendItem(identity, codexStreamingMessageBody(text)) - deps.sink.publish() - } - - const persistLatestStreams = (): void => { - for (const [key, text] of latestStreamText) { - if (checkpointLengths.get(key) !== text.length) { - persistStream(key, text, true) - } - } - } - - const flushStreams = (): void => { - coalescer.flushAll() - persistLatestStreams() - } - - const coalescer = createAgentSessionDeltaCoalescer({ - windowMs: deps.coalesceMs, - schedule: deps.schedule, - emit: (key, text) => { - persistStream(key, text, false) - } - }) - const identityFor = ( threadId: string, turnId: string | null, item: { type: string; id: string } ): AgentJournalItemIdentity => { - const key = itemKey(threadId, item.id) + const key = codexStructuredItemKey(threadId, item.id) const existing = identities.get(key) if (existing) { return existing @@ -165,6 +131,16 @@ export function createCodexJournalTranslator( return identity } + const streams = createCodexStructuredItemStreams({ + sink: deps.sink, + coalesceMs: deps.coalesceMs, + schedule: deps.schedule, + identityFor: (threadId, params, item) => { + const turnId = readCodexTurnId(params) ?? currentTurnIds.get(threadId) ?? null + return identityFor(threadId, turnId, item) + } + }) + const handleItemEvent = (event: { threadId: string method: string @@ -180,13 +156,13 @@ export function createCodexJournalTranslator( const translated = codexJournalItem(item) const command = readString(item, 'command') if (command) { - details.set(itemKey(event.threadId, item.id), command) + details.set(codexStructuredItemKey(event.threadId, item.id), command) } if (event.method === 'item/completed') { // The completed body is authoritative; the coalesced text is now stale. - coalescer.forget(itemKey(event.threadId, item.id)) - latestStreamText.delete(itemKey(event.threadId, item.id)) - checkpointLengths.delete(itemKey(event.threadId, item.id)) + streams.forget(event.threadId, item.id) + } else { + streams.track(event.threadId, item, identity) } if (!translated.body) { return true @@ -230,7 +206,7 @@ export function createCodexJournalTranslator( codexApprovalItem({ method: event.method, params: event.params, - detail: details.get(itemKey(event.threadId, event.codexItemId)) ?? null + detail: details.get(codexStructuredItemKey(event.threadId, event.codexItemId)) ?? null }) ) deps.bindPromptItemId?.(agentJournalItemKey(identity), event.threadId, event.promptKey) @@ -240,36 +216,31 @@ export function createCodexJournalTranslator( return { handle: (event) => { if (event.type === 'ended') { - flushStreams() + streams.flush() for (const [threadId, turnId] of currentTurnIds) { publishTurnLifecycle(event.sessionId, threadId, turnId, 'completed') } currentTurnIds.clear() return } - if (event.type === 'notification' && event.method === AGENT_MESSAGE_DELTA_METHOD) { - const params = readRecord(event.params) - const codexItemId = readString(params, 'itemId') - const delta = params.delta - if (codexItemId && typeof delta === 'string') { - coalescer.append(itemKey(event.threadId, codexItemId), delta) - } else { - appendUnhandled(`notification:${event.method}`, event.params) - } + if ( + event.type === 'notification' && + streams.handle(event.threadId, event.method, event.params) + ) { return } // Lifecycle bypass: nothing may be journaled ahead of the text it follows. - flushStreams() + streams.flush() if (event.type === 'prompt') { handlePrompt(event) return } if (event.type === 'server-request') { - appendUnhandled(`request:${event.method}`, event.params) + appendUnhandled(`request:${event.method}`, event.params, event.threadId) return } if (event.type === 'provider-frame') { - appendUnhandled(event.kind, event.payload) + appendUnhandled(event.kind, event.payload, event.threadId) return } if (event.method === 'turn/started') { @@ -292,20 +263,19 @@ export function createCodexJournalTranslator( } if (event.method === 'item/started' || event.method === 'item/completed') { if (!handleItemEvent(event)) { - appendUnhandled(`notification:${event.method}`, event.params) + appendUnhandled(`notification:${event.method}`, event.params, event.threadId) } return } - appendUnhandled(`notification:${event.method}`, event.params) + appendUnhandled(`notification:${event.method}`, event.params, event.threadId) }, - flush: flushStreams, + flush: streams.flush, dispose: () => { - coalescer.dispose() + streams.dispose() identities.clear() details.clear() currentTurnIds.clear() - latestStreamText.clear() - checkpointLengths.clear() + genericRowsByTurn.clear() } } } diff --git a/src/main/native-chat/agent-session-wire/provider-frame-disposition.test.ts b/src/main/native-chat/agent-session-wire/provider-frame-disposition.test.ts index 93444c638bd..299946ed226 100644 --- a/src/main/native-chat/agent-session-wire/provider-frame-disposition.test.ts +++ b/src/main/native-chat/agent-session-wire/provider-frame-disposition.test.ts @@ -1,7 +1,11 @@ import { describe, expect, it } from 'vitest' import { CODEX_APP_SERVER_NOTIFICATION_METHODS } from '../../codex/codex-app-server-notification-schema' import { CLAUDE_STREAM_JSON_FRAME_KINDS } from './claude-stream-json-frame-schema' -import { classifyProviderFrame, PROVIDER_FRAME_CLASSIFICATIONS } from './provider-frame-disposition' +import { + classifyProviderFrame, + isDeltaShapedProviderFrameKind, + PROVIDER_FRAME_CLASSIFICATIONS +} from './provider-frame-disposition' describe('provider frame classification catalog', () => { it('classifies every pinned Codex app-server notification method', () => { @@ -16,6 +20,19 @@ describe('provider frame classification catalog', () => { ]) }) + it('classifies every pinned delta kind as stream-into-item', () => { + const deltaKinds = [ + ...Object.keys(PROVIDER_FRAME_CLASSIFICATIONS.codex), + ...Object.keys(PROVIDER_FRAME_CLASSIFICATIONS.claude) + ].filter(isDeltaShapedProviderFrameKind) + + expect(deltaKinds.length).toBeGreaterThan(0) + for (const kind of deltaKinds) { + const provider = kind.startsWith('message:') ? 'claude' : 'codex' + expect(classifyProviderFrame(provider, kind, {}), kind).toBe('stream-into-item') + } + }) + it('suppresses benign hook lifecycle and Codex rate-limit frames', () => { expect(classifyProviderFrame('codex', 'notification:hook/started', {})).toBe( 'suppressed-benign' @@ -51,4 +68,11 @@ describe('provider frame classification catalog', () => { ) expect(classifyProviderFrame('claude', 'message:future_event', {})).toBe('timeline-substantive') }) + + it('structurally diverts unknown future delta kinds from generic rows', () => { + expect(classifyProviderFrame('codex', 'notification:item/newThing/outputDelta', {})).toBe( + 'stream-into-item' + ) + expect(classifyProviderFrame('claude', 'message:future_delta', {})).toBe('stream-into-item') + }) }) diff --git a/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts b/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts index be41fda8522..e7359b97ee1 100644 --- a/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts +++ b/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts @@ -3,6 +3,7 @@ import type { ClaudeStreamJsonFrameKind } from './claude-stream-json-frame-schem export type ProviderFrameClassification = | 'timeline-substantive' + | 'stream-into-item' | 'status-chrome' | 'suppressed-benign' | 'error-surface' @@ -41,15 +42,15 @@ export const PROVIDER_FRAME_CLASSIFICATIONS = { 'item/completed': 'timeline-substantive', 'rawResponseItem/completed': 'suppressed-benign', 'rawResponse/completed': 'suppressed-benign', - 'item/agentMessage/delta': 'timeline-substantive', - 'item/plan/delta': 'timeline-substantive', - 'command/exec/outputDelta': 'timeline-substantive', - 'process/outputDelta': 'timeline-substantive', + 'item/agentMessage/delta': 'stream-into-item', + 'item/plan/delta': 'stream-into-item', + 'command/exec/outputDelta': 'stream-into-item', + 'process/outputDelta': 'stream-into-item', 'process/exited': 'timeline-substantive', - 'item/commandExecution/outputDelta': 'timeline-substantive', - 'item/commandExecution/terminalInteraction': 'timeline-substantive', - 'item/fileChange/outputDelta': 'timeline-substantive', - 'item/fileChange/patchUpdated': 'timeline-substantive', + 'item/commandExecution/outputDelta': 'stream-into-item', + 'item/commandExecution/terminalInteraction': 'stream-into-item', + 'item/fileChange/outputDelta': 'stream-into-item', + 'item/fileChange/patchUpdated': 'stream-into-item', 'serverRequest/resolved': 'suppressed-benign', 'item/mcpToolCall/progress': 'status-chrome', 'mcpServer/oauthLogin/completed': 'status-chrome', @@ -61,9 +62,9 @@ export const PROVIDER_FRAME_CLASSIFICATIONS = { 'externalAgentConfig/import/progress': 'status-chrome', 'externalAgentConfig/import/completed': 'status-chrome', 'fs/changed': 'suppressed-benign', - 'item/reasoning/summaryTextDelta': 'timeline-substantive', - 'item/reasoning/summaryPartAdded': 'timeline-substantive', - 'item/reasoning/textDelta': 'timeline-substantive', + 'item/reasoning/summaryTextDelta': 'stream-into-item', + 'item/reasoning/summaryPartAdded': 'stream-into-item', + 'item/reasoning/textDelta': 'stream-into-item', 'thread/compacted': 'status-chrome', 'model/rerouted': 'status-chrome', 'model/verification': 'status-chrome', @@ -77,9 +78,9 @@ export const PROVIDER_FRAME_CLASSIFICATIONS = { 'fuzzyFileSearch/sessionCompleted': 'suppressed-benign', 'thread/realtime/started': 'status-chrome', 'thread/realtime/itemAdded': 'timeline-substantive', - 'thread/realtime/transcript/delta': 'timeline-substantive', + 'thread/realtime/transcript/delta': 'stream-into-item', 'thread/realtime/transcript/done': 'timeline-substantive', - 'thread/realtime/outputAudio/delta': 'timeline-substantive', + 'thread/realtime/outputAudio/delta': 'stream-into-item', 'thread/realtime/sdp': 'suppressed-benign', 'thread/realtime/error': 'error-surface', 'thread/realtime/closed': 'status-chrome', @@ -93,10 +94,10 @@ export const PROVIDER_FRAME_CLASSIFICATIONS = { 'message:result': 'status-chrome', 'message:system:init': 'status-chrome', 'message:stream_event:message_start': 'status-chrome', - 'message:stream_event:message_delta': 'status-chrome', + 'message:stream_event:message_delta': 'stream-into-item', 'message:stream_event:message_stop': 'status-chrome', 'message:stream_event:content_block_start': 'status-chrome', - 'message:stream_event:content_block_delta': 'timeline-substantive', + 'message:stream_event:content_block_delta': 'stream-into-item', 'message:stream_event:content_block_stop': 'status-chrome', 'message:system:compact_boundary': 'status-chrome', 'message:system:status': 'status-chrome', @@ -183,6 +184,10 @@ function notificationKind(kind: string): string { return kind.startsWith('notification:') ? kind.slice('notification:'.length) : kind } +export function isDeltaShapedProviderFrameKind(kind: string): boolean { + return notificationKind(kind).toLowerCase().endsWith('delta') +} + function catalogClassification( provider: string, kind: string @@ -203,6 +208,9 @@ export function classifyProviderFrame( kind: string, payload: unknown ): ProviderFrameClassification { + if (isDeltaShapedProviderFrameKind(kind)) { + return 'stream-into-item' + } if (hasProviderError(payload)) { return 'error-surface' } diff --git a/src/main/native-chat/agent-session-wire/unhandled-provider-frame.test.ts b/src/main/native-chat/agent-session-wire/unhandled-provider-frame.test.ts index 69a2ee4e87d..ed3dea75c21 100644 --- a/src/main/native-chat/agent-session-wire/unhandled-provider-frame.test.ts +++ b/src/main/native-chat/agent-session-wire/unhandled-provider-frame.test.ts @@ -72,6 +72,20 @@ describe('unhandled provider frame journal fallback', () => { ).toBeNull() }) + it('never creates generic rows for known or unknown delta-shaped frames', () => { + expect( + unhandledProviderFrameJournalItem('codex', 'notification:item/commandExecution/outputDelta', { + itemId: 'exec-1', + delta: 'x' + }) + ).toBeNull() + expect( + unhandledProviderFrameJournalItem('codex', 'notification:item/future/outputDelta', { + error: 'still a delta' + }) + ).toBeNull() + }) + it('renders codex systemError and Claude error result variants', () => { const codex = unhandledProviderFrameJournalItem('codex', 'notification:thread/status/changed', { threadId: 'thread-1', diff --git a/src/main/native-chat/agent-session-wire/unhandled-provider-frame.ts b/src/main/native-chat/agent-session-wire/unhandled-provider-frame.ts index 819989d61c1..d72ebe0d69d 100644 --- a/src/main/native-chat/agent-session-wire/unhandled-provider-frame.ts +++ b/src/main/native-chat/agent-session-wire/unhandled-provider-frame.ts @@ -28,7 +28,11 @@ export function unhandledProviderFrameJournalItem( limits: JournalPayloadLimits = DEFAULT_JOURNAL_PAYLOAD_LIMITS ): UnhandledProviderFrameJournalItem | null { const classification = classifyProviderFrame(provider, kind, payload) - if (classification === 'status-chrome' || classification === 'suppressed-benign') { + if ( + classification === 'stream-into-item' || + classification === 'status-chrome' || + classification === 'suppressed-benign' + ) { return null } const serialized = serializeProviderPayload(payload)