fix(native-chat): fold Codex item stream deltas

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<string, unknown> {
+  return typeof value === 'object' && value !== null ? (value as Record<string, unknown>) : {}
+}
+
+function readString(source: Record<string, unknown>, 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<string, StreamState>()
+  const latestText = new Map<string, string>()
+  const checkpointLengths = new Map<string, number>()
+
+  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()
+    }
+  }
+}

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',

src/main/codex/codex-structured-item-translation.ts

@@ -74,6 +74,10 @@ export class CodexTurnOrdinals {
   }
 }
 
+function readRecord(value: unknown): Record<string, unknown> {
+  return typeof value === 'object' && value !== null ? (value as Record<string, unknown>) : {}
+}
+
 /**
  * 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 }
+}

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()
 

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<string, string>()
   const currentTurnIds = new Map<string, string>()
-  const latestStreamText = new Map<string, string>()
-  const checkpointLengths = new Map<string, number>()
+  const genericRowsByTurn = new Map<string, number>()
   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()
     }
   }
 }

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')
+  })
 })

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'
   }

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',

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)