Skip to content

fix(storage): import current Codex transcript messages #3626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
liugddx wants to merge 1 commit into from
+226 −3
Closed

fix(storage): import current Codex transcript messages #3626

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 120 additions & 0 deletions packages/storage/src/__tests__/codex-session-adapter.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,6 +241,51 @@ describe('CodexSessionAdapter', () => {
});
});

test('imports current item-completed presentation messages without provider mirrors', async () => {
await withCodexHome(async (codexHome) => {
const sessionId = 'codex-item-completed';
await seedRawRollout(codexHome, sessionId, itemCompletedRollout(sessionId));
const adapter = new CodexSessionAdapter({ codexHome });

assert.equal((await adapter.listSessions())[0]?.name, 'Review the current PR');
const session = await adapter.readSession(sessionId);
assert.deepEqual(
session.messages.map((message) =>
message.type === 'user' || message.type === 'assistant'
? { type: message.type, id: message.id, turnId: message.turnId, text: message.text }
: { type: message.type },
),
[
{
type: 'user',
id: 'current-user-item',
turnId: 'current-turn',
text: 'Review the current PR',
},
{
type: 'assistant',
id: 'current-commentary-item',
turnId: 'current-turn',
text: 'I am checking the changed paths.',
},
{
type: 'assistant',
id: 'current-final-item',
turnId: 'current-turn',
text: 'The PR is ready.',
},
{ type: 'turn_state' },
],
);
assert.equal(
session.messages.some(
(message) => message.type === 'user' && message.text.includes('environment_context'),
),
false,
);
});
});

test('imports terminal errors as failed without failing turns on non-terminal errors', async () => {
await withCodexHome(async (codexHome) => {
const sessionId = 'codex-error-semantics';
Expand Down Expand Up @@ -427,6 +472,81 @@ function minimalRollout(
].join('\n');
}

function itemCompletedRollout(sessionId: string): string {
const event = (second: number, payload: Record<string, unknown>): string =>
JSON.stringify({
timestamp: `2026-08-08T00:00:${String(second).padStart(2, '0')}.000Z`,
type: 'event_msg',
payload,
});
const response = (second: number, payload: Record<string, unknown>): string =>
JSON.stringify({
timestamp: `2026-08-08T00:00:${String(second).padStart(2, '0')}.100Z`,
type: 'response_item',
payload,
});
return [
JSON.stringify({
timestamp: '2026-08-08T00:00:00.000Z',
type: 'session_meta',
payload: {
session_id: sessionId,
id: sessionId,
cwd: '/workspace/project',
source: 'cli',
},
}),
event(1, { type: 'task_started', turn_id: 'current-turn' }),
response(1, {
type: 'message',
id: 'injected-context',
role: 'user',
content: [{ type: 'input_text', text: '<environment_context>private</environment_context>' }],
}),
event(2, {
type: 'item_completed',
thread_id: sessionId,
turn_id: 'current-turn',
item: {
type: 'UserMessage',
id: 'current-user-item',
content: [{ type: 'text', text: 'Review the current PR', text_elements: [] }],
},
}),
event(3, {
type: 'item_completed',
thread_id: sessionId,
turn_id: 'current-turn',
item: {
type: 'AgentMessage',
id: 'current-commentary-item',
content: [{ type: 'Text', text: 'I am checking the changed paths.' }],
phase: 'commentary',
},
}),
response(3, {
type: 'message',
id: 'provider-commentary-mirror',
role: 'assistant',
content: [{ type: 'output_text', text: 'I am checking the changed paths.' }],
phase: 'commentary',
}),
event(4, {
type: 'item_completed',
thread_id: sessionId,
turn_id: 'current-turn',
item: {
type: 'AgentMessage',
id: 'current-final-item',
content: [{ type: 'Text', text: 'The PR is ready.' }],
phase: 'final_answer',
},
}),
event(5, { type: 'task_complete', turn_id: 'current-turn' }),
'',
].join('\n');
}

function errorSemanticsRollout(sessionId: string): string {
const event = (second: number, payload: Record<string, unknown>): string =>
JSON.stringify({
Expand Down
109 changes: 106 additions & 3 deletions packages/storage/src/codex-session-adapter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,11 @@ type JsonRecord = Record<string, unknown>;
*
* Codex persists presentation history as `event_msg` records and provider
* protocol facts as `response_item` records. User, assistant, and reasoning
* messages come from `event_msg` to avoid importing their response-item
* mirrors twice. Tool calls/results come from response items because they own
* the stable call identity and raw arguments/output.
* messages come from the presentation events to avoid importing their
* response-item mirrors twice. Codex <=0.144 wrote one event per message;
* newer builds wrap the presentation item in `item_completed`. Tool
* calls/results come from response items because they own the stable call
* identity and raw arguments/output.
*/
export class CodexSessionAdapter implements ExternalSessionAdapter {
readonly id = CODEX_SESSION_ADAPTER_ID;
Expand Down Expand Up @@ -261,6 +263,18 @@ function convertCodexRollout(
let lastTimestamp = normalizeEpochMs(sessionMeta?.timestamp) ?? 0;
let firstUserText: string | undefined;
const failedTurnIds = new Set<string>();
// The two presentation schemas are mirrors, not additive streams. Prefer
// the legacy events if a transitional rollout contains both so one visible
// message cannot be imported twice.
const usesCompletedPresentationItems = !records.some((record) => {
if (record.value.type !== 'event_msg') return false;
const payload = asRecord(record.value.payload);
return (
payload?.type === 'user_message' ||
payload?.type === 'agent_message' ||
payload?.type === 'agent_reasoning'
);
});

const timestampFor = (record: ParsedRolloutRecord): number => {
const parsed = normalizeEpochMs(record.value.timestamp);
Expand Down Expand Up @@ -295,6 +309,51 @@ function convertCodexRollout(
continue;
}

if (eventType === 'item_completed' && usesCompletedPresentationItems) {
const item = asRecord(payload.item);
if (!item) continue;
const itemType = stringField(item, 'type');
const eventTurnId = stringField(payload, 'turn_id');
if (eventTurnId) {
activeTurnId = eventTurnId;
activeTurnIsExplicit = true;
}

if (itemType === 'UserMessage') {
if (!activeTurnIsExplicit) {
activeTurnId = generatedCodexId(expectedSessionId, 'turn', record.line);
}
const text = completedUserMessageText(item);
if (text.length === 0) continue;
firstUserText ??= text;
messages.push({
type: 'user',
id: stringField(item, 'id') ?? generatedCodexId(expectedSessionId, 'user', record.line),
turnId: ensureTurnId(record.line),
ts: timestampFor(record),
text,
});
continue;
}

if (itemType === 'AgentMessage') {
const text = completedAgentMessageText(item);
if (text.length === 0) continue;
messages.push({
type: 'assistant',
id:
stringField(item, 'id') ??
generatedCodexId(expectedSessionId, 'assistant', record.line),
turnId: ensureTurnId(record.line),
ts: timestampFor(record),
text,
modelId: activeModel,
contentOrder: ['text'],
});
continue;
}
}

if (eventType === 'user_message') {
if (!activeTurnIsExplicit) {
activeTurnId = generatedCodexId(expectedSessionId, 'turn', record.line);
Expand Down Expand Up @@ -520,6 +579,16 @@ function catalogEntryFromRolloutHead(
firstUserText === undefined
) {
firstUserText = stringField(payload, 'message');
} else if (
record.type === 'event_msg' &&
payload.type === 'item_completed' &&
firstUserText === undefined
) {
const item = asRecord(payload.item);
if (item?.type === 'UserMessage') {
const text = completedUserMessageText(item);
if (text.length > 0) firstUserText = text;
}
}
if (id && firstUserText !== undefined) break;
}
Expand Down Expand Up @@ -823,6 +892,40 @@ function mediaOnlyUserText(payload: JsonRecord): string {
return audio.length > 0 || localAudio.length > 0 ? '[Audio]' : '';
}

function completedUserMessageText(item: JsonRecord): string {
const content = Array.isArray(item.content) ? item.content : [];
const text = presentationItemText(content, 'text');
if (text.length > 0) return text;
if (
content.some((part) => {
const type = stringField(asRecord(part), 'type');
return type === 'image' || type === 'local_image';
})
) {
return '[Image]';
}
return content.some((part) => {
const type = stringField(asRecord(part), 'type');
return type === 'audio' || type === 'local_audio';
})
? '[Audio]'
: '';
}

function completedAgentMessageText(item: JsonRecord): string {
const content = Array.isArray(item.content) ? item.content : [];
return presentationItemText(content, 'Text');
}

function presentationItemText(content: unknown[], expectedType: string): string {
return content
.flatMap((part) => {
const record = asRecord(part);
return record?.type === expectedType && typeof record.text === 'string' ? [record.text] : [];
})
.join('\n');
}

async function isDirectory(path: string): Promise<boolean> {
try {
return (await stat(path)).isDirectory();
Expand Down