fix(native-chat): keep TUI transcripts readable

2026-10-22 08:03:37 +00:00 · 2026-08-11 21:37:39 -07:00
parent 7316ac3926
commit ed78cfd598
13 changed files with 467 additions and 22 deletions
@@ -10,7 +10,10 @@ import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-
 import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types'
 import { readJournalBlob } from './journal-blob-store'
 import { createLegacyIdentityTracker } from './journal-legacy-identity'
-import { importLegacyTranscriptIntoJournal } from './journal-legacy-import'
+import {
+  appendLegacyTranscriptMessages,
+  importLegacyTranscriptIntoJournal
+} from './journal-legacy-import'
 import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds'
 import { openAgentSessionJournal, type AgentSessionJournal } from './journal-store'

@@ -278,6 +281,46 @@ describe('claude import', () => {
 })

 describe('codex import', () => {
+  it('upserts live transcript messages without rolling the structured epoch', async () => {
+    const journal = await open('codex', CODEX_SESSION)
+    const epoch = journal.epoch
+    const message = {
+      id: 'live-tui-message',
+      role: 'assistant' as const,
+      blocks: [{ type: 'text' as const, text: 'first version' }],
+      timestamp: 1_800_000_000_000,
+      source: 'transcript' as const
+    }
+
+    await appendLegacyTranscriptMessages({
+      journal,
+      agent: 'codex',
+      sessionId: CODEX_SESSION,
+      fence: 2,
+      messages: [message]
+    })
+    await appendLegacyTranscriptMessages({
+      journal,
+      agent: 'codex',
+      sessionId: CODEX_SESSION,
+      fence: 2,
+      messages: [{ ...message, blocks: [{ type: 'text', text: 'final version' }] }]
+    })
+
+    expect(journal.epoch).toBe(epoch)
+    expect(journal.snapshot().items).toMatchObject([
+      {
+        itemId: legacyKey('live-tui-message'),
+        revision: 2,
+        body: {
+          kind: 'message',
+          role: 'assistant',
+          blocks: [{ type: 'text', text: 'final version' }]
+        }
+      }
+    ])
+  })
+
  it('keys rollout records in the import-scoped namespace, not as app-server ordinals', async () => {
    const filePath = await writeFixture('rollout.jsonl', CODEX_LINES)
    const journal = await open('codex', CODEX_SESSION)
@@ -47,6 +47,34 @@ export type LegacyImportResult =
  | { ok: true; epoch: string; cursor: AgentJournalCursor; imported: number }
  | { ok: false; error: string }

+export async function appendLegacyTranscriptMessages(input: {
+  journal: AgentSessionJournal
+  agent: AgentType
+  sessionId: string
+  fence: number
+  messages: NativeChatMessage[]
+}): Promise<number> {
+  let appended = 0
+  for (const message of input.messages) {
+    const mapped = legacyItemBody(message, DEFAULT_JOURNAL_PAYLOAD_LIMITS)
+    for (const blob of mapped.blobs) {
+      await putJournalBlob(input.journal.directory, blob.digest, blob.payload)
+    }
+    await input.journal.appendItem(
+      {
+        provider: 'legacy',
+        agent: input.agent,
+        sessionId: input.sessionId,
+        recordId: message.id
+      },
+      mapped.body,
+      { fence: input.fence, observedAt: message.timestamp ?? undefined }
+    )
+    appended += 1
+  }
+  return appended
+}
+
 export async function importLegacyTranscriptIntoJournal(input: {
  journal: AgentSessionJournal
  agent: AgentType
@@ -53,6 +53,7 @@ export async function handoffStructuredSessionToTui(
  context.publishStage(record, 'to-tui')
  let owner: StructuredTuiOwner | null = null
  try {
+    await deps.prepareTuiHistoryCatchup?.(sessionId, record.lease.runtimeFence)
    owner = await deps.transport!.launchTui({
      record,
      fence: record.lease.runtimeFence,
@@ -71,6 +72,7 @@ export async function handoffStructuredSessionToTui(
      now: deps.now()
    })
  } catch (error) {
+    deps.stopTuiHistoryCatchup?.(sessionId)
    if (owner) {
      if (!deps.transport?.stopFailedTuiLaunch) {
        await markManualRecovery(context, sessionId, operationId)
@@ -90,6 +92,7 @@ export async function handoffStructuredSessionToTui(
    throw error
  }
  context.retainOwner(sessionId, owner)
+  await deps.activateTuiHistoryCatchup?.(sessionId)
  context.setStatus(sessionId, {
    owner: 'tui',
    direction: null,
@@ -178,6 +178,7 @@ async function restoreRecoverableLiveTui(
    now: input.deps.now()
  })
  input.retainOwner(record.sessionId, owner)
+  await startRecoveredTuiCatchup(input, record)
  input.setStatus(record.sessionId, {
    owner: 'tui',
    direction: null,
@@ -192,6 +193,7 @@ async function restoreRecoverableLiveTui(
 async function restoreLiveTui(input: RestartAccess, record: AgentSessionRecord): Promise<void> {
  const owner = await input.deps.transport!.recoverTuiOwner(record)
  input.retainOwner(record.sessionId, owner)
+  await startRecoveredTuiCatchup(input, record)
  input.setStatus(record.sessionId, {
    owner: 'tui',
    direction: null,
@@ -239,6 +241,7 @@ async function restoreProving(input: RestartAccess, record: AgentSessionRecord):
      now: input.deps.now()
    })
    input.retainOwner(record.sessionId, reproved)
+    await startRecoveredTuiCatchup(input, record)
    input.setStatus(record.sessionId, {
      owner: 'tui',
      direction: null,
@@ -261,6 +264,14 @@ async function restoreProving(input: RestartAccess, record: AgentSessionRecord):
  await continueHandoff(input, stopped)
 }

+async function startRecoveredTuiCatchup(
+  input: RestartAccess,
+  record: AgentSessionRecord
+): Promise<void> {
+  await input.deps.prepareTuiHistoryCatchup?.(record.sessionId, record.lease.runtimeFence)
+  await input.deps.activateTuiHistoryCatchup?.(record.sessionId)
+}
+
 async function continueHandoff(input: RestartAccess, record: AgentSessionRecord): Promise<void> {
  const direction = record.lease.runtimeKind === 'native' ? 'to-tui' : 'to-native'
  const operationId = record.lease.handoffOperationId!
@@ -75,6 +75,7 @@ export async function handoffStructuredSessionToNative(
  ) {
    throw new Error('agent_session_operation_conflict')
  }
+  deps.stopTuiHistoryCatchup?.(sessionId)
  await deps.importTuiHistory({
    sessionId,
    fence: record.lease.runtimeFence,
@@ -53,6 +53,9 @@ export type StructuredAgentSessionHandoffDeps = {
    fence: number
    transcriptPath?: string
  }) => Promise<void>
+  prepareTuiHistoryCatchup?: (sessionId: string, fence: number) => Promise<void>
+  activateTuiHistoryCatchup?: (sessionId: string) => Promise<void>
+  stopTuiHistoryCatchup?: (sessionId: string) => void
  publish: (sessionId: string, status: AgentSessionHandoffStatus) => void
  schedule: (sessionId: string, task: () => Promise<void>) => Promise<void>
  now: () => number
@@ -55,6 +55,11 @@ let stopRecoveredOwner: ReturnType<
  typeof vi.fn<StructuredAgentSessionHandoffTransport['stopRecoveredOwner']>
 >
 let operations: number
+let prepareTuiHistoryCatchup: ReturnType<
+  typeof vi.fn<(sessionId: string, fence: number) => Promise<void>>
+>
+let activateTuiHistoryCatchup: ReturnType<typeof vi.fn<(sessionId: string) => Promise<void>>>
+let stopTuiHistoryCatchup: ReturnType<typeof vi.fn<(sessionId: string) => void>>

 function operationId(): string {
  operations += 1
@@ -220,6 +225,9 @@ function createCoordinator(): StructuredAgentSessionHandoffCoordinator {
        { fence, recovered: true }
      )
    },
+    prepareTuiHistoryCatchup,
+    activateTuiHistoryCatchup,
+    stopTuiHistoryCatchup,
    publish: (_sessionId, status) => statuses.push(status),
    schedule: async (_sessionId, task) => task(),
    now: () => NOW
@@ -252,6 +260,9 @@ beforeEach(async () => {
  importFailure = null
  nativeAcquireFailure = null
  acquireNativeCalls = 0
+  prepareTuiHistoryCatchup = vi.fn(async () => undefined)
+  activateTuiHistoryCatchup = vi.fn(async () => undefined)
+  stopTuiHistoryCatchup = vi.fn()
  stopRecoveredOwner = vi.fn(async () => undefined)
  store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
  await establishNativeOwner()
@@ -324,6 +335,12 @@ describe('structured session ownership handoff', () => {
      claimStatus: 'live',
      handoffStage: null
    })
+    expect(prepareTuiHistoryCatchup).toHaveBeenCalledWith(SESSION, forwardRecord.lease.runtimeFence)
+    expect(activateTuiHistoryCatchup).toHaveBeenCalledWith(SESSION)
+    expect(stopTuiHistoryCatchup).toHaveBeenCalledWith(SESSION)
+    expect(prepareTuiHistoryCatchup.mock.invocationCallOrder[0]).toBeLessThan(
+      launchTui.mock.invocationCallOrder[0] ?? Infinity
+    )
  })

  it('queues a busy native turn, supports cancel, and starts only after idle', async () => {
@@ -1,3 +1,4 @@
+import { join } from 'node:path'
 import type { AgentSessionRecord } from '../../../shared/agent-session-record'
 import { importLegacyTranscriptIntoJournal } from '../agent-session-journal/journal-legacy-import'
 import { journalIdentityFor } from './structured-agent-session-attach'
@@ -7,6 +8,7 @@ import type { StructuredAgentSessionHostDeps } from './structured-agent-session-
 import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types'
 import { StructuredAgentSessionHandoffCoordinator } from './structured-agent-session-handoff'
 import type { AgentSessionSubscribers } from './structured-agent-session-subscribers'
+import { StructuredTuiTranscriptCatchup } from './structured-tui-transcript-catchup'

 type HostHandoffAccess = {
  session: (sessionId: string) => StructuredAgentSessionHostSession
@@ -17,7 +19,9 @@ type HostHandoffAccess = {
  now: () => number
 }

-export type StructuredAgentSessionHostHandoff = StructuredAgentSessionHandoffCoordinator
+export type StructuredAgentSessionHostHandoff = StructuredAgentSessionHandoffCoordinator & {
+  stopTuiHistoryCatchup: () => void
+}

 export async function refreshRecoverableStructuredHandoffStatus(
  handoff: StructuredAgentSessionHostHandoff,
@@ -35,7 +39,17 @@ export function createStructuredAgentSessionHostHandoff(
  deps: StructuredAgentSessionHostDeps,
  host: HostHandoffAccess
 ): StructuredAgentSessionHostHandoff {
-  return new StructuredAgentSessionHandoffCoordinator({
+  const tuiHistoryCatchup = new StructuredTuiTranscriptCatchup({
+    store: deps.store,
+    session: host.session,
+    schedule: host.serialize,
+    publish: (sessionId) => {
+      const session = host.session(sessionId)
+      host.subscribers.publish(sessionId, session.journal)
+    },
+    ...(deps.onEventSinkError ? { onError: deps.onEventSinkError } : {})
+  })
+  const coordinator = new StructuredAgentSessionHandoffCoordinator({
    store: deps.store,
    claimKeyId: deps.claimKeyId,
    ...(deps.handoffTransport ? { transport: deps.handoffTransport } : {}),
@@ -49,6 +63,9 @@ export function createStructuredAgentSessionHostHandoff(
    acquireNativeStop: async (sessionId, turnId, fence) =>
      (await deps.adapter.cancelTurn({ sessionId, turnId, fence })).cancelled,
    importTuiHistory: (input) => importTuiHistory(deps, host, input),
+    prepareTuiHistoryCatchup: (sessionId, fence) => tuiHistoryCatchup.prepare(sessionId, fence),
+    activateTuiHistoryCatchup: (sessionId) => tuiHistoryCatchup.activate(sessionId),
+    stopTuiHistoryCatchup: (sessionId) => tuiHistoryCatchup.stop(sessionId),
    publish: (sessionId, status) => {
      const session = host.session(sessionId)
      const fence = deps.store.getRecord(sessionId)?.lease.runtimeFence ?? session.fence
@@ -57,6 +74,7 @@ export function createStructuredAgentSessionHostHandoff(
    schedule: host.serialize,
    now: host.now
  })
+  return Object.assign(coordinator, { stopTuiHistoryCatchup: () => tuiHistoryCatchup.stopAll() })
 }

 async function importTuiHistory(
@@ -75,7 +93,9 @@ async function importTuiHistory(
    agent: 'codex',
    sessionId: head.handle.threadId,
    fence: input.fence,
-    ...(input.transcriptPath ? { options: { filePath: input.transcriptPath } } : {})
+    options: input.transcriptPath
+      ? { filePath: input.transcriptPath }
+      : { codexSessionsDirs: [join(record.accountHome.path, 'sessions')] }
  })
  if (!imported.ok) {
    throw new Error(imported.error)
@@ -682,7 +682,10 @@ describe('restart', () => {
    await host.send(CALLER, { envelope: envelope('agentSession.send', { body }), body })
    await reboot(async () => ({ outcome: 'indeterminate', reason: 'read does not need ownership' }))
    acquire.mockClear()
+    const listRecords = vi.spyOn(store, 'listRecords')

+    await host.restoreReadableSessions()
+    const restoreReads = listRecords.mock.calls.length
    await host.restoreReadableSessions()

    expect(host.listSessionTabs()).toEqual([
@@ -691,6 +694,7 @@ describe('restart', () => {
    const history = host.history({ sessionId: SESSION, direction: 'tail' })
    expect(history.ok && history.page.items).not.toHaveLength(0)
    expect(acquire).not.toHaveBeenCalled()
+    expect(listRecords).toHaveBeenCalledTimes(restoreReads)
  })

  it('clears stale TUI recovery when restart successfully reacquires the native owner', async () => {
@@ -47,7 +47,6 @@ import {
  type AgentSessionSubscribeInput
 } from './structured-agent-session-subscribers'
 import { StructuredAgentSessionTaskQueue } from './structured-agent-session-task-queue'
-import { restoreStructuredAgentSessionsOnRestart } from './structured-agent-session-restart-restore'
 import type { StructuredAgentSessionHandoffTransport } from './structured-agent-session-handoff-types'
 import {
  createStructuredAgentSessionHostHandoff,
@@ -56,6 +55,7 @@ import {
 } from './structured-agent-session-host-handoff'
 import { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state'
 import { listStructuredAgentSessionTabs } from './structured-agent-session-host-tabs'
+import { StructuredAgentSessionReadableRestorer } from './structured-agent-session-readable-restorer'
 import type {
  StructuredAgentSessionCaller,
  StructuredAgentSessionHostSession
@@ -80,6 +80,7 @@ export class StructuredAgentSessionHost {
  private readonly runtimeState: StructuredAgentSessionHostRuntimeState
  private readonly reconcileLeases: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
  private readonly handoffs: StructuredAgentSessionHostHandoff
+  private readonly readableRestorer: StructuredAgentSessionReadableRestorer

  constructor(readonly deps: StructuredAgentSessionHostDeps) {
    this.runtimeState = new StructuredAgentSessionHostRuntimeState(deps, (record) =>
@@ -98,6 +99,20 @@ export class StructuredAgentSessionHost {
      subscribers: this.subscribers,
      now: this.now
    })
+    this.readableRestorer = new StructuredAgentSessionReadableRestorer({
+      store: deps.store,
+      journalRoot: deps.journalRoot,
+      reconcile: this.reconcileLeases,
+      resume: (params) =>
+        this.attach({ callerKey: 'trusted-local:host-restart' }, params).then(
+          (result) => result.ok
+        ),
+      serialize: (sessionId, task) => this.serialize(sessionId, task),
+      hasSession: (sessionId) => this.sessions.has(sessionId),
+      onReadable: (sessionId, restored) => this.sessions.set(sessionId, restored),
+      restoreHandoff: (sessionId) => this.handoffs.restore(sessionId),
+      now: this.now
+    })
    this.runtimeState.startLeaseRenewal()
  }

@@ -113,23 +128,8 @@ export class StructuredAgentSessionHost {
    return listStructuredAgentSessionTabs(this.sessions)
  }

-  async restoreReadableSessions(): Promise<void> {
-    await restoreStructuredAgentSessionsOnRestart({
-      store: this.deps.store,
-      journalRoot: this.deps.journalRoot,
-      records: this.deps.store.listRecords().filter((record) => record.provider === 'codex'),
-      reconcile: this.reconcileLeases,
-      operationId: () =>
-        `${Math.trunc(this.now()).toString().padStart(13, '0')}-${randomUUID().replaceAll('-', '')}`,
-      resume: (params) =>
-        this.attach({ callerKey: 'trusted-local:host-restart' }, params).then(
-          (result) => result.ok
-        ),
-      serialize: (sessionId, task) => this.serialize(sessionId, task),
-      hasSession: (sessionId) => this.sessions.has(sessionId),
-      onReadable: (sessionId, restored) => this.sessions.set(sessionId, restored),
-      restoreHandoff: (sessionId) => this.handoffs.restore(sessionId)
-    })
+  restoreReadableSessions(): Promise<void> {
+    return this.readableRestorer.restore()
  }

  private serialize<T>(sessionId: string, task: () => Promise<T>): Promise<T> {
@@ -207,6 +207,7 @@ export class StructuredAgentSessionHost {

  async flushAllStreamedEvents(): Promise<void> {
    this.runtimeState.stopLeaseRenewal()
+    this.handoffs.stopTuiHistoryCatchup()
    await this.tasks.drainAttaches()
    await this.runtimeState.flushAllEventSinks()
  }
@@ -0,0 +1,37 @@
+import { randomUUID } from 'node:crypto'
+import type { AgentSessionWireRefusal } from '../../../shared/agent-session-wire'
+import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import type { AgentSessionAttachParams } from './structured-agent-session-attach'
+import type { RestoredStructuredAgentSessionRead } from './structured-agent-session-read-restore'
+import { restoreStructuredAgentSessionsOnRestart } from './structured-agent-session-restart-restore'
+
+export class StructuredAgentSessionReadableRestorer {
+  private restorePromise: Promise<void> | null = null
+
+  constructor(
+    private readonly input: {
+      store: AgentSessionRecordStore
+      journalRoot: string
+      reconcile: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
+      resume: (params: AgentSessionAttachParams) => Promise<boolean>
+      serialize: <T>(sessionId: string, task: () => Promise<T>) => Promise<T>
+      hasSession: (sessionId: string) => boolean
+      onReadable: (sessionId: string, restored: RestoredStructuredAgentSessionRead) => void
+      restoreHandoff: (sessionId: string) => Promise<void>
+      now: () => number
+    }
+  ) {}
+
+  restore(): Promise<void> {
+    this.restorePromise ??= restoreStructuredAgentSessionsOnRestart({
+      ...this.input,
+      records: this.input.store.listRecords().filter((record) => record.provider === 'codex'),
+      operationId: () =>
+        `${Math.trunc(this.input.now()).toString().padStart(13, '0')}-${randomUUID().replaceAll('-', '')}`
+    }).catch((error) => {
+      this.restorePromise = null
+      throw error
+    })
+    return this.restorePromise
+  }
+}
@@ -0,0 +1,124 @@
+import { appendFile, mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import { openAgentSessionJournal } from '../agent-session-journal/journal-store'
+import { StructuredTuiTranscriptCatchup } from './structured-tui-transcript-catchup'
+
+const NOW = 1_800_000_000_000
+const SESSION = 'session-catchup'
+const THREAD = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60'
+
+let root: string
+let store: AgentSessionRecordStore
+
+function rolloutLine(message: string): string {
+  return `${JSON.stringify({
+    type: 'event_msg',
+    timestamp: '2026-08-11T10:00:00.000Z',
+    payload: { type: 'agent_message', message }
+  })}\n`
+}
+
+beforeEach(async () => {
+  root = await mkdtemp(join(tmpdir(), 'orca-tui-catchup-'))
+  store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
+})
+
+afterEach(async () => {
+  await rm(root, { recursive: true, force: true })
+})
+
+describe('StructuredTuiTranscriptCatchup', () => {
+  it('tails only TUI-era appends from the durable account home', async () => {
+    const accountHome = join(root, 'isolated-codex-home')
+    const sessionsDir = join(accountHome, 'sessions', '2026', '08', '11')
+    const rollout = join(sessionsDir, `rollout-2026-08-11T10-00-00-${THREAD}.jsonl`)
+    await mkdir(sessionsDir, { recursive: true })
+    await writeFile(rollout, rolloutLine('before handoff'), 'utf8')
+    const reserved = await store.reserveOwner({
+      sessionId: SESSION,
+      location: {
+        executionHostId: 'local',
+        wslDistro: null,
+        workspaceId: 'workspace-1',
+        workspaceKind: 'folder'
+      },
+      provider: 'codex',
+      accountHome: { variable: 'CODEX_HOME', path: accountHome },
+      runtimeKind: 'tui',
+      expectedFence: null,
+      spawnToken: 'tui-token',
+      claimKeyId: 'key-1',
+      handoffOperationId: null,
+      probe: { outcome: 'reservation-unused' },
+      operation: {
+        callerKey: 'test',
+        operationId: `${NOW}-${'1'.padStart(32, '0')}`,
+        fingerprint: 'initial'
+      },
+      now: NOW
+    })
+    const fence = reserved.record.lease.runtimeFence
+    await store.commitProcessIdentity({
+      sessionId: SESSION,
+      fence,
+      process: {
+        hostId: 'local',
+        pid: 4200,
+        processStartTimeMs: NOW - 1_000,
+        spawnToken: 'tui-token'
+      },
+      now: NOW
+    })
+    await store.proveOwner({
+      sessionId: SESSION,
+      fence,
+      link: {
+        linkId: 'tui-link',
+        handle: { provider: 'codex', threadId: THREAD },
+        origin: 'created',
+        mintedAtFence: fence,
+        observedAt: NOW
+      },
+      now: NOW
+    })
+    const journal = await openAgentSessionJournal({
+      identity: {
+        sessionId: SESSION,
+        workspaceId: 'workspace-1',
+        hostId: 'local',
+        agent: 'codex',
+        providerHandle: { kind: 'codex', threadId: THREAD }
+      },
+      journalDir: join(root, 'journal')
+    })
+    const publish = vi.fn()
+    const catchup = new StructuredTuiTranscriptCatchup({
+      store,
+      session: () => ({ journal, params: {} as never, fence }),
+      schedule: async (_sessionId, task) => task(),
+      publish
+    })
+
+    await catchup.prepare(SESSION, fence)
+    await catchup.activate(SESSION)
+    expect(journal.snapshot().items).toEqual([])
+
+    await appendFile(rollout, rolloutLine('during TUI'), 'utf8')
+    await vi.waitFor(() =>
+      expect(journal.snapshot().items.map((item) => item.body)).toContainEqual({
+        kind: 'message',
+        role: 'assistant',
+        blocks: [{ type: 'text', text: 'during TUI' }]
+      })
+    )
+    expect(publish).toHaveBeenCalledOnce()
+
+    catchup.stop(SESSION)
+    await appendFile(rollout, rolloutLine('after stop'), 'utf8')
+    await new Promise((resolve) => setTimeout(resolve, 100))
+    expect(journal.snapshot().items).toHaveLength(1)
+  })
+})
@@ -0,0 +1,153 @@
+import { join } from 'node:path'
+import type { NativeChatMessage } from '../../../shared/native-chat-types'
+import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import { appendLegacyTranscriptMessages } from '../agent-session-journal/journal-legacy-import'
+import { resolveSessionFilePath } from '../session-file-resolver'
+import {
+  subscribeNativeChatTranscript,
+  type NativeChatTranscriptSubscription
+} from '../transcript-watch'
+import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types'
+
+type CatchupState = {
+  active: boolean
+  fence: number
+  providerSessionId: string
+  pending: NativeChatMessage[]
+  seen: Set<string>
+  subscription: NativeChatTranscriptSubscription | null
+}
+
+export class StructuredTuiTranscriptCatchup {
+  private readonly states = new Map<string, CatchupState>()
+
+  constructor(
+    private readonly input: {
+      store: AgentSessionRecordStore
+      session: (sessionId: string) => StructuredAgentSessionHostSession
+      schedule: (sessionId: string, task: () => Promise<void>) => Promise<void>
+      publish: (sessionId: string) => void
+      onError?: (input: { sessionId: string; error: unknown }) => void
+    }
+  ) {}
+
+  async prepare(sessionId: string, fence: number): Promise<void> {
+    this.stop(sessionId)
+    const record = this.input.store.getRecord(sessionId)
+    const head = record?.providerHandleChain.at(-1)
+    if (!record || head?.handle.provider !== 'codex') {
+      return
+    }
+    const providerSessionId = head.handle.threadId
+    const codexSessionsDirs = [join(record.accountHome.path, 'sessions')]
+    const filePath = await resolveSessionFilePath('codex', providerSessionId, {
+      codexSessionsDirs
+    })
+    let initialReady: (() => void) | null = null
+    const ready = filePath ? new Promise<void>((resolve) => (initialReady = resolve)) : null
+    const state: CatchupState = {
+      active: false,
+      fence,
+      providerSessionId,
+      pending: [],
+      seen: new Set(),
+      subscription: null
+    }
+    const receive = (messages: NativeChatMessage[]) => this.receive(sessionId, state, messages)
+    this.states.set(sessionId, state)
+    try {
+      state.subscription = await subscribeNativeChatTranscript({
+        agent: 'codex',
+        sessionId: providerSessionId,
+        codexSessionsDirs,
+        ...(filePath ? { filePath, initialLimit: 0 } : {}),
+        onInitialSnapshot: (messages) => {
+          receive(messages)
+          initialReady?.()
+          initialReady = null
+        },
+        onAppend: receive
+      })
+      await ready
+    } catch (error) {
+      if (this.states.get(sessionId) === state) {
+        this.states.delete(sessionId)
+      }
+      state.subscription?.unsubscribe()
+      throw error
+    }
+  }
+
+  async activate(sessionId: string): Promise<void> {
+    const state = this.states.get(sessionId)
+    if (!state) {
+      return
+    }
+    state.active = true
+    const pending = state.pending.splice(0)
+    await this.append(sessionId, state, pending)
+  }
+
+  stop(sessionId: string): void {
+    const state = this.states.get(sessionId)
+    this.states.delete(sessionId)
+    state?.subscription?.unsubscribe()
+  }
+
+  stopAll(): void {
+    for (const sessionId of this.states.keys()) {
+      this.stop(sessionId)
+    }
+  }
+
+  private receive(sessionId: string, state: CatchupState, messages: NativeChatMessage[]): void {
+    if (this.states.get(sessionId) !== state) {
+      return
+    }
+    if (!state.active) {
+      state.pending.push(...messages)
+      return
+    }
+    void this.input
+      .schedule(sessionId, () => this.append(sessionId, state, messages))
+      .catch((error) => this.input.onError?.({ sessionId, error }))
+  }
+
+  private async append(
+    sessionId: string,
+    state: CatchupState,
+    messages: NativeChatMessage[]
+  ): Promise<void> {
+    if (this.states.get(sessionId) !== state) {
+      return
+    }
+    const record = this.input.store.getRecord(sessionId)
+    if (
+      !record ||
+      record.lease.runtimeKind !== 'tui' ||
+      record.lease.claimStatus !== 'live' ||
+      record.lease.runtimeFence !== state.fence
+    ) {
+      return
+    }
+    const fresh = messages.filter((message) => !state.seen.has(message.id))
+    if (fresh.length === 0) {
+      return
+    }
+    try {
+      await appendLegacyTranscriptMessages({
+        journal: this.input.session(sessionId).journal,
+        agent: 'codex',
+        sessionId: state.providerSessionId,
+        fence: state.fence,
+        messages: fresh
+      })
+      for (const message of fresh) {
+        state.seen.add(message.id)
+      }
+      this.input.publish(sessionId)
+    } catch (error) {
+      this.input.onError?.({ sessionId, error })
+    }
+  }
+}