diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts index 87a64a08e7a..044315e81aa 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts @@ -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) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts index a95d0ace7b8..df61ee47861 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts @@ -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 { + 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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-forward.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-forward.ts index ede43215076..74cf24f913b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-forward.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-forward.ts @@ -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, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts index ab21fe630ab..abbd15bde10 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-restart.ts @@ -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 { 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 { + await input.deps.prepareTuiHistoryCatchup?.(record.sessionId, record.lease.runtimeFence) + await input.deps.activateTuiHistoryCatchup?.(record.sessionId) +} + async function continueHandoff(input: RestartAccess, record: AgentSessionRecord): Promise { const direction = record.lease.runtimeKind === 'native' ? 'to-tui' : 'to-native' const operationId = record.lease.handoffOperationId! diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-reverse.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-reverse.ts index 8596813d041..3a182d0c0dc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-reverse.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-reverse.ts @@ -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, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-types.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-types.ts index bc554c2b0ca..9433089b26b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-types.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff-types.ts @@ -53,6 +53,9 @@ export type StructuredAgentSessionHandoffDeps = { fence: number transcriptPath?: string }) => Promise + prepareTuiHistoryCatchup?: (sessionId: string, fence: number) => Promise + activateTuiHistoryCatchup?: (sessionId: string) => Promise + stopTuiHistoryCatchup?: (sessionId: string) => void publish: (sessionId: string, status: AgentSessionHandoffStatus) => void schedule: (sessionId: string, task: () => Promise) => Promise now: () => number diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts index 46f8785b027..c6b8209c684 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts @@ -55,6 +55,11 @@ let stopRecoveredOwner: ReturnType< typeof vi.fn > let operations: number +let prepareTuiHistoryCatchup: ReturnType< + typeof vi.fn<(sessionId: string, fence: number) => Promise> +> +let activateTuiHistoryCatchup: ReturnType Promise>> +let stopTuiHistoryCatchup: ReturnType 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 () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts index de77ba98b48..642836a7f2f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-handoff.ts @@ -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) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts index 39206eeace0..9f57dda1d73 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -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 () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index cbf8408f33e..995ee4a2fa5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -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 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 { - 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 { + return this.readableRestorer.restore() } private serialize(sessionId: string, task: () => Promise): Promise { @@ -207,6 +207,7 @@ export class StructuredAgentSessionHost { async flushAllStreamedEvents(): Promise { this.runtimeState.stopLeaseRenewal() + this.handoffs.stopTuiHistoryCatchup() await this.tasks.drainAttaches() await this.runtimeState.flushAllEventSinks() } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts new file mode 100644 index 00000000000..3a2b5097848 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts @@ -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 | null = null + + constructor( + private readonly input: { + store: AgentSessionRecordStore + journalRoot: string + reconcile: (sessionId: string) => Promise + resume: (params: AgentSessionAttachParams) => Promise + serialize: (sessionId: string, task: () => Promise) => Promise + hasSession: (sessionId: string) => boolean + onReadable: (sessionId: string, restored: RestoredStructuredAgentSessionRead) => void + restoreHandoff: (sessionId: string) => Promise + now: () => number + } + ) {} + + restore(): Promise { + 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 + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.test.ts b/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.test.ts new file mode 100644 index 00000000000..016890b5802 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.test.ts @@ -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) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.ts b/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.ts new file mode 100644 index 00000000000..68d52de17ef --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-tui-transcript-catchup.ts @@ -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 + subscription: NativeChatTranscriptSubscription | null +} + +export class StructuredTuiTranscriptCatchup { + private readonly states = new Map() + + constructor( + private readonly input: { + store: AgentSessionRecordStore + session: (sessionId: string) => StructuredAgentSessionHostSession + schedule: (sessionId: string, task: () => Promise) => Promise + publish: (sessionId: string) => void + onError?: (input: { sessionId: string; error: unknown }) => void + } + ) {} + + async prepare(sessionId: string, fence: number): Promise { + 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((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 { + 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 { + 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 }) + } + } +}