fix(native-chat): recover TUI transcript gaps

This commit is contained in:
Brennan Benson
2026-08-11 21:56:01 -07:00
parent ed78cfd598
commit ff35abf7a8
10 changed files with 373 additions and 85 deletions
@@ -41,6 +41,7 @@ export type LegacyImportOptions = ResolveSessionFileOptions & {
/** Resolve directly to this file, skipping path discovery. */
filePath?: string
limits?: JournalPayloadLimits
decodedMessageIdentities?: true
}
export type LegacyImportResult =
@@ -100,7 +101,8 @@ export async function importLegacyTranscriptIntoJournal(input: {
filePath,
transcriptAgent,
agent: input.agent,
sessionId: input.sessionId
sessionId: input.sessionId,
decodedMessageIdentities: options.decodedMessageIdentities
})
} catch (err) {
return { ok: false, error: err instanceof Error ? err.message : String(err) }
@@ -140,6 +142,7 @@ async function decodeWithIdentities(input: {
transcriptAgent: keyof typeof TRANSCRIPT_DECODERS
agent: AgentType
sessionId: string
decodedMessageIdentities?: true
}): Promise<{ messages: NativeChatMessage[]; identities: AgentJournalItemIdentity[] }> {
const tracker = createLegacyIdentityTracker({
transcriptAgent: input.transcriptAgent,
@@ -156,11 +159,20 @@ async function decodeWithIdentities(input: {
input.filePath,
0,
(line, fallbackId) => {
const identity = tracker.identify(line, lineIndex)
const trackedIdentity = tracker.identify(line, lineIndex)
lineIndex += 1
const message = decode(line, fallbackId)
if (message) {
identities.push(identity)
identities.push(
input.decodedMessageIdentities
? {
provider: 'legacy',
agent: input.agent,
sessionId: input.sessionId,
recordId: message.id
}
: trackedIdentity
)
}
return message
},
@@ -268,7 +268,7 @@ async function startRecoveredTuiCatchup(
input: RestartAccess,
record: AgentSessionRecord
): Promise<void> {
await input.deps.prepareTuiHistoryCatchup?.(record.sessionId, record.lease.runtimeFence)
await input.deps.recoverTuiHistoryCatchup?.(record.sessionId, record.lease.runtimeFence)
await input.deps.activateTuiHistoryCatchup?.(record.sessionId)
}
@@ -54,6 +54,7 @@ export type StructuredAgentSessionHandoffDeps = {
transcriptPath?: string
}) => Promise<void>
prepareTuiHistoryCatchup?: (sessionId: string, fence: number) => Promise<void>
recoverTuiHistoryCatchup?: (sessionId: string, fence: number) => Promise<void>
activateTuiHistoryCatchup?: (sessionId: string) => Promise<void>
stopTuiHistoryCatchup?: (sessionId: string) => void
publish: (sessionId: string, status: AgentSessionHandoffStatus) => void
@@ -58,6 +58,9 @@ let operations: number
let prepareTuiHistoryCatchup: ReturnType<
typeof vi.fn<(sessionId: string, fence: number) => Promise<void>>
>
let recoverTuiHistoryCatchup: 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>>
@@ -226,6 +229,7 @@ function createCoordinator(): StructuredAgentSessionHandoffCoordinator {
)
},
prepareTuiHistoryCatchup,
recoverTuiHistoryCatchup,
activateTuiHistoryCatchup,
stopTuiHistoryCatchup,
publish: (_sessionId, status) => statuses.push(status),
@@ -261,6 +265,7 @@ beforeEach(async () => {
nativeAcquireFailure = null
acquireNativeCalls = 0
prepareTuiHistoryCatchup = vi.fn(async () => undefined)
recoverTuiHistoryCatchup = vi.fn(async () => undefined)
activateTuiHistoryCatchup = vi.fn(async () => undefined)
stopTuiHistoryCatchup = vi.fn()
stopRecoveredOwner = vi.fn(async () => undefined)
@@ -495,6 +500,10 @@ describe('structured session ownership handoff', () => {
handoffStage: null
})
expect(coordinator.status(SESSION)).toMatchObject({ owner: 'tui', phase: 'idle' })
expect(recoverTuiHistoryCatchup).toHaveBeenCalledWith(
SESSION,
store.getRecord(SESSION)?.lease.runtimeFence
)
})
it('clears stale recovery after re-proving the live TUI owner', async () => {
@@ -508,6 +517,7 @@ describe('structured session ownership handoff', () => {
now: NOW
})
coordinator = createCoordinator()
recoverTuiHistoryCatchup.mockClear()
await coordinator.restore(SESSION)
expect(coordinator.status(SESSION)).toMatchObject({
@@ -517,6 +527,7 @@ describe('structured session ownership handoff', () => {
terminal: { handle: 'term-tui' }
})
expect(reproveTuiOwner).toHaveBeenCalledOnce()
expect(recoverTuiHistoryCatchup).toHaveBeenCalledWith(SESSION, record.lease.runtimeFence)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
runtimeKind: 'tui',
claimStatus: 'live',
@@ -47,6 +47,10 @@ export function createStructuredAgentSessionHostHandoff(
const session = host.session(sessionId)
host.subscribers.publish(sessionId, session.journal)
},
reset: (sessionId, fence) => {
const session = host.session(sessionId)
host.subscribers.reset(sessionId, session.journal, 'epoch_changed', fence)
},
...(deps.onEventSinkError ? { onError: deps.onEventSinkError } : {})
})
const coordinator = new StructuredAgentSessionHandoffCoordinator({
@@ -64,6 +68,7 @@ export function createStructuredAgentSessionHostHandoff(
(await deps.adapter.cancelTurn({ sessionId, turnId, fence })).cancelled,
importTuiHistory: (input) => importTuiHistory(deps, host, input),
prepareTuiHistoryCatchup: (sessionId, fence) => tuiHistoryCatchup.prepare(sessionId, fence),
recoverTuiHistoryCatchup: (sessionId, fence) => tuiHistoryCatchup.recover(sessionId, fence),
activateTuiHistoryCatchup: (sessionId) => tuiHistoryCatchup.activate(sessionId),
stopTuiHistoryCatchup: (sessionId) => tuiHistoryCatchup.stop(sessionId),
publish: (sessionId, status) => {
@@ -0,0 +1,60 @@
import { readFile } from 'node:fs/promises'
import { join } from 'node:path'
import { durableWriteTempPath, writeFileDurable } from '../../durable-file-write'
const BOUNDARY_FILE = 'structured-tui-transcript-boundary.json'
const BOUNDARY_SCHEMA_VERSION = 1
export type StructuredTuiTranscriptBoundary = {
providerSessionId: string
runtimeFence: number
filePath: string | null
offset: number
}
function boundaryPath(journalDirectory: string): string {
return join(journalDirectory, BOUNDARY_FILE)
}
export async function readStructuredTuiTranscriptBoundary(
journalDirectory: string
): Promise<StructuredTuiTranscriptBoundary | null> {
let parsed: unknown
try {
parsed = JSON.parse(await readFile(boundaryPath(journalDirectory), 'utf8'))
} catch {
return null
}
if (!parsed || typeof parsed !== 'object') {
return null
}
const value = parsed as Record<string, unknown>
if (
value.schemaVersion !== BOUNDARY_SCHEMA_VERSION ||
typeof value.providerSessionId !== 'string' ||
!Number.isSafeInteger(value.runtimeFence) ||
(value.filePath !== null && typeof value.filePath !== 'string') ||
!Number.isSafeInteger(value.offset) ||
(value.offset as number) < 0
) {
return null
}
return {
providerSessionId: value.providerSessionId,
runtimeFence: value.runtimeFence as number,
filePath: value.filePath as string | null,
offset: value.offset as number
}
}
export async function writeStructuredTuiTranscriptBoundary(
journalDirectory: string,
boundary: StructuredTuiTranscriptBoundary
): Promise<void> {
const filePath = boundaryPath(journalDirectory)
await writeFileDurable(
durableWriteTempPath(filePath),
filePath,
JSON.stringify({ schemaVersion: BOUNDARY_SCHEMA_VERSION, ...boundary })
)
}
@@ -30,95 +30,128 @@ afterEach(async () => {
await rm(root, { recursive: true, force: true })
})
async function createCatchupFixture() {
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')
})
return { fence, journal, rollout }
}
function createCatchup(input: Awaited<ReturnType<typeof createCatchupFixture>>) {
return new StructuredTuiTranscriptCatchup({
store,
session: () => ({ journal: input.journal, params: {} as never, fence: input.fence }),
schedule: async (_sessionId, task) => task(),
publish: vi.fn(),
reset: vi.fn()
})
}
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
})
const fixture = await createCatchupFixture()
const catchup = createCatchup(fixture)
await catchup.prepare(SESSION, fence)
await catchup.prepare(SESSION, fixture.fence)
await catchup.activate(SESSION)
expect(journal.snapshot().items).toEqual([])
expect(fixture.journal.snapshot().items).toEqual([])
await appendFile(rollout, rolloutLine('during TUI'), 'utf8')
await appendFile(fixture.rollout, rolloutLine('during TUI'), 'utf8')
await vi.waitFor(() =>
expect(journal.snapshot().items.map((item) => item.body)).toContainEqual({
expect(fixture.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 appendFile(fixture.rollout, rolloutLine('after stop'), 'utf8')
await new Promise((resolve) => setTimeout(resolve, 100))
expect(journal.snapshot().items).toHaveLength(1)
expect(fixture.journal.snapshot().items).toHaveLength(1)
})
it('replays transcript writes made while the host watcher was down', async () => {
const fixture = await createCatchupFixture()
const beforeCrash = createCatchup(fixture)
await beforeCrash.prepare(SESSION, fixture.fence)
await beforeCrash.activate(SESSION)
await appendFile(fixture.rollout, rolloutLine('before host crash'), 'utf8')
await vi.waitFor(() => expect(fixture.journal.snapshot().items).toHaveLength(1))
beforeCrash.stopAll()
await appendFile(fixture.rollout, rolloutLine('while host was down'), 'utf8')
const recovered = createCatchup(fixture)
await recovered.recover(SESSION, fixture.fence)
await recovered.activate(SESSION)
const text = fixture.journal
.snapshot()
.items.flatMap((item) =>
item.body.kind === 'message'
? item.body.blocks.flatMap((block) => (block.type === 'text' ? [block.text] : []))
: []
)
expect(text).toEqual(['before host crash', 'while host was down'])
recovered.stopAll()
})
})
@@ -1,13 +1,26 @@
import { stat } from 'node:fs/promises'
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 {
appendLegacyTranscriptMessages,
importLegacyTranscriptIntoJournal
} from '../agent-session-journal/journal-legacy-import'
import { resolveSessionFilePath } from '../session-file-resolver'
import {
readIncrementalTranscriptMessages,
type IncrementalTranscriptState
} from '../transcript-incremental-reader'
import { nativeChatLineDecoderForAgent } from '../transcript-tail-reader'
import {
subscribeNativeChatTranscript,
type NativeChatTranscriptSubscription
} from '../transcript-watch'
import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types'
import {
readStructuredTuiTranscriptBoundary,
writeStructuredTuiTranscriptBoundary
} from './structured-tui-transcript-boundary'
type CatchupState = {
active: boolean
@@ -27,11 +40,20 @@ export class StructuredTuiTranscriptCatchup {
session: (sessionId: string) => StructuredAgentSessionHostSession
schedule: (sessionId: string, task: () => Promise<void>) => Promise<void>
publish: (sessionId: string) => void
reset: (sessionId: string, fence: number) => void
onError?: (input: { sessionId: string; error: unknown }) => void
}
) {}
async prepare(sessionId: string, fence: number): Promise<void> {
await this.start(sessionId, fence, false)
}
async recover(sessionId: string, fence: number): Promise<void> {
await this.start(sessionId, fence, true)
}
private async start(sessionId: string, fence: number, recovering: boolean): Promise<void> {
this.stop(sessionId)
const record = this.input.store.getRecord(sessionId)
const head = record?.providerHandleChain.at(-1)
@@ -39,11 +61,17 @@ export class StructuredTuiTranscriptCatchup {
return
}
const providerSessionId = head.handle.threadId
const journal = this.input.session(sessionId).journal
const codexSessionsDirs = [join(record.accountHome.path, 'sessions')]
const boundary = recovering
? await readStructuredTuiTranscriptBoundary(journal.directory)
: null
const filePath = await resolveSessionFilePath('codex', providerSessionId, {
codexSessionsDirs
codexSessionsDirs,
...(boundary?.filePath ? { transcriptPath: boundary.filePath } : {})
})
let initialReady: (() => void) | null = null
let baselineOffset = 0
const ready = filePath ? new Promise<void>((resolve) => (initialReady = resolve)) : null
const state: CatchupState = {
active: false,
@@ -61,7 +89,8 @@ export class StructuredTuiTranscriptCatchup {
sessionId: providerSessionId,
codexSessionsDirs,
...(filePath ? { filePath, initialLimit: 0 } : {}),
onInitialSnapshot: (messages) => {
onInitialSnapshot: (messages, _hasMore, beforeOffset) => {
baselineOffset = beforeOffset
receive(messages)
initialReady?.()
initialReady = null
@@ -69,6 +98,33 @@ export class StructuredTuiTranscriptCatchup {
onAppend: receive
})
await ready
if (!recovering) {
await writeStructuredTuiTranscriptBoundary(journal.directory, {
providerSessionId,
runtimeFence: fence,
filePath,
offset: baselineOffset
})
} else if (
filePath &&
boundary?.providerSessionId === providerSessionId &&
boundary.runtimeFence === fence &&
boundary.filePath === filePath
) {
await this.readRecoveryGap(sessionId, state, filePath, boundary.offset)
} else if (filePath) {
const imported = await importLegacyTranscriptIntoJournal({
journal,
agent: 'codex',
sessionId: providerSessionId,
fence,
options: { filePath, decodedMessageIdentities: true }
})
if (!imported.ok) {
throw new Error(imported.error)
}
this.input.reset(sessionId, fence)
}
} catch (error) {
if (this.states.get(sessionId) === state) {
this.states.delete(sessionId)
@@ -78,6 +134,45 @@ export class StructuredTuiTranscriptCatchup {
}
}
private async readRecoveryGap(
sessionId: string,
state: CatchupState,
filePath: string,
offset: number
): Promise<void> {
let size: number
try {
size = (await stat(filePath)).size
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
return
}
throw error
}
const start = offset <= size ? offset : 0
const incremental: IncrementalTranscriptState = {
offset: start,
pendingChunks: [],
pendingStart: start,
pendingBytes: 0,
droppingOversizedRecord: false
}
const decode = nativeChatLineDecoderForAgent('codex')
if (!decode) {
throw new Error('Transcript unavailable')
}
let messages: NativeChatMessage[]
try {
messages = await readIncrementalTranscriptMessages(filePath, incremental, decode)
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
return
}
throw error
}
this.receive(sessionId, state, messages)
}
async activate(sessionId: string): Promise<void> {
const state = this.states.get(sessionId)
if (!state) {
@@ -130,7 +225,14 @@ export class StructuredTuiTranscriptCatchup {
) {
return
}
const fresh = messages.filter((message) => !state.seen.has(message.id))
const ids = new Set(state.seen)
const fresh = messages.filter((message) => {
if (ids.has(message.id)) {
return false
}
ids.add(message.id)
return true
})
if (fresh.length === 0) {
return
}
@@ -0,0 +1,46 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { setStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry'
import { OrcaRuntimeService } from './orca-runtime'
afterEach(() => setStructuredAgentSessionHost(null))
describe('structured session cold restoration', () => {
it('hydrates PTY inventory before recovering durable TUI owners and only runs once', async () => {
const runtime = new OrcaRuntimeService()
const hydrate = vi.fn()
const refresh = vi.fn(async () => new Set<string>())
const ensureHost = vi.fn(async () => undefined)
const restoreReadableSessions = vi.fn(async () => undefined)
const internal = runtime as unknown as {
getKnownWorkspaceSessionWorktreeIds(): Set<string>
hydrateHeadlessMobileSessionTabsFromWorkspaceSession(
worktreeId?: string,
options?: { allowAttachedWindow?: boolean; onlyRuntimeOwnedTerminals?: boolean }
): Set<string>
refreshMobileSessionPtyRecords(): Promise<Set<string> | null>
ensureStructuredAgentSessionHost(): Promise<void>
}
internal.getKnownWorkspaceSessionWorktreeIds = () => new Set(['workspace-1'])
internal.hydrateHeadlessMobileSessionTabsFromWorkspaceSession = hydrate
internal.refreshMobileSessionPtyRecords = refresh
internal.ensureStructuredAgentSessionHost = ensureHost
setStructuredAgentSessionHost({
restoreReadableSessions,
listSessionTabs: () => []
} as never)
await runtime.restoreStructuredAgentSessionTabs()
await runtime.restoreStructuredAgentSessionTabs()
expect(hydrate).toHaveBeenCalledWith('workspace-1', {
allowAttachedWindow: true,
onlyRuntimeOwnedTerminals: true
})
expect(hydrate).toHaveBeenCalledWith()
expect(refresh).toHaveBeenCalledOnce()
expect(restoreReadableSessions).toHaveBeenCalledOnce()
expect(refresh.mock.invocationCallOrder[0]).toBeLessThan(
restoreReadableSessions.mock.invocationCallOrder[0] ?? Infinity
)
})
})
+19 -1
View File
@@ -2806,6 +2806,7 @@ export class OrcaRuntimeService {
private readonly terminalFocusNavigationCoalescer =
new TerminalFocusNavigationCoalescer<RuntimeTerminalFocus>()
private pendingMobileSessionPtyInventoryRefresh: Promise<Set<string> | null> | null = null
private structuredAgentSessionTabRestorePromise: Promise<void> | null = null
private leaves = new Map<string, RuntimeLeafRecord>()
// Why: PTY output is a per-keystroke hot path. Looking up affected leaves by
// ptyId keeps active TUI redraws independent of the total open terminal count.
@@ -9509,7 +9510,24 @@ export class OrcaRuntimeService {
}
}
async restoreStructuredAgentSessionTabs(): Promise<void> {
restoreStructuredAgentSessionTabs(): Promise<void> {
this.structuredAgentSessionTabRestorePromise ??=
this.restoreStructuredAgentSessionTabsOnce().catch((error) => {
this.structuredAgentSessionTabRestorePromise = null
throw error
})
return this.structuredAgentSessionTabRestorePromise
}
private async restoreStructuredAgentSessionTabsOnce(): Promise<void> {
for (const worktreeId of this.getKnownWorkspaceSessionWorktreeIds()) {
this.hydrateHeadlessMobileSessionTabsFromWorkspaceSession(worktreeId, {
allowAttachedWindow: true,
onlyRuntimeOwnedTerminals: true
})
}
this.hydrateHeadlessMobileSessionTabsFromWorkspaceSession()
await this.refreshMobileSessionPtyRecords()
await this.ensureStructuredAgentSessionHost()
const host = getStructuredAgentSessionHost()
await host?.restoreReadableSessions()