Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion apps/cli/src/backends/codex/appServer/runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2327,7 +2327,7 @@ export function createCodexAppServerRuntime(params: Readonly<{
await flushItemTranscriptBoundary(null);
await params.session.sendUserTextMessageCommitted(update.text, {
localId: `codex-app-server-user:${context.streamScopeId}:${update.itemId}`,
meta: { importedFrom: 'codex-app-server' },
meta: { importedFrom: 'codex-app-server', ...(update.nativeContent ? { codexNativeContent: update.nativeContent } : {}) },
});
return;
}
Expand Down
Original file line number Diff line number Diff line change
@@ -1,32 +1,28 @@
import { createCodexAppServerStreamEventBridge } from '../streamEventBridge';
import type { CodexRolloutAction } from '../../localControl/rolloutMapper';
import { codexNativeProjectionError } from '../../utils/codexNativeProjectionError';

/** The native history API returns the same completed items as live app-server notifications. */
export function projectCodexNativeHistoryItem(item: Readonly<Record<string, unknown>>): CodexRolloutAction[] {
if (item.type === 'userMessage' && (!Array.isArray(item.content) || !item.content.every((part: unknown) =>
part !== null && typeof part === 'object' && 'type' in part && part.type === 'text'
&& 'text' in part && typeof part.text === 'string'))) {
throw new Error('Codex native history contains unsupported user content');
}
const bridge = createCodexAppServerStreamEventBridge();
const updates = bridge.onNotification({
method: item.type === 'userMessage' ? 'item/started' : 'item/completed', params: { item },
});
const actions: CodexRolloutAction[] = [];
for (const update of updates) {
switch (update.type) {
case 'provider-user-text': actions.push({ type: 'user-text', text: update.text, ...(update.clientId ? { clientId: update.clientId } : {}) }); break;
case 'provider-user-text': actions.push({ type: 'user-text', text: update.text, ...(update.clientId ? { clientId: update.clientId } : {}), ...(update.nativeContent ? { nativeContent: update.nativeContent } : {}) }); break;
case 'assistant-text-final':
case 'async-user-input-request': actions.push({ type: 'assistant-text', text: update.text }); break;
case 'tool-call': actions.push({ type: 'tool-call', callId: update.callId, name: update.name, input: update.input }); break;
case 'tool-result': actions.push({ type: 'tool-result', callId: update.callId, output: update.output }); break;
case 'context-compaction': actions.push({ type: 'context-compaction', phase: 'completed', lifecycleId: `codex:context-compaction:${update.itemId}`, source: 'provider-event', providerEventId: update.itemId }); break;
case 'reasoning-final': actions.push({ type: 'thinking', text: update.text }); break;
default: throw new Error('Codex native history contains an unsupported transcript item');
default: throw codexNativeProjectionError('transcript item', item.type);
}
}
if (updates.length === 0 && !['reasoning', 'hookPrompt'].includes(String(item.type))) {
throw new Error('Codex native history contains an unsupported transcript item');
throw codexNativeProjectionError('transcript item', item.type);
}
return actions;
}
35 changes: 34 additions & 1 deletion apps/cli/src/backends/codex/appServer/streamEventBridge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,34 @@ import { describe, expect, it } from 'vitest';
import { createCodexAppServerStreamEventBridge } from './streamEventBridge';

describe('createCodexAppServerStreamEventBridge', () => {
it('retains collaboration identities that arrive only when a live tool completes', () => {
const bridge = createCodexAppServerStreamEventBridge();
bridge.onNotification({ method: 'item/started', params: { item: {
id: 'spawn', type: 'collabAgentToolCall', tool: 'spawnAgent', receiverThreadIds: [],
} } });
const result = bridge.onNotification({ method: 'item/completed', params: { item: {
id: 'spawn', type: 'collabAgentToolCall', tool: 'spawnAgent', receiverThreadIds: ['child'], status: 'completed',
} } });
expect(result).toEqual([expect.objectContaining({ type: 'tool-result', output: expect.objectContaining({ receiverThreadIds: ['child'] }) })]);
});
it.each([
{ type: 'subAgentActivity', name: 'subAgentActivity', input: { kind: 'interacted', agentThreadId: 'child-1', agentPath: '/root/child', model: 'model', reasoningEffort: 'high' }, result: { completedAtMs: 20 } },
{ type: 'collabAgentToolCall', name: 'spawnAgent', input: { tool: 'spawnAgent', senderThreadId: 'parent', receiverThreadIds: ['child-1'], prompt: 'Inspect fixture', model: 'model' }, result: { status: 'completed', agentsStates: { 'child-1': { status: 'running' } } } },
{ type: 'dynamicToolCall', name: 'lookup', input: { namespace: 'catalog', tool: 'lookup', arguments: { query: 'fixture' } }, result: { status: 'completed', success: true, contentItems: [{ type: 'inputText', text: 'result' }], durationMs: 10 } },
])('preserves native $type inputs and results in both history and live projection', ({ type, name, input, result }) => {
const completed = { id: 'call-1', type, ...input, ...result };
const history = createCodexAppServerStreamEventBridge().onNotification({ method: 'item/completed', params: { item: completed } });
expect(history).toEqual([
expect.objectContaining({ type: 'tool-call', callId: 'call-1', name, input }),
expect.objectContaining({ type: 'tool-result', callId: 'call-1', output: result }),
]);
const live = createCodexAppServerStreamEventBridge();
expect(live.onNotification({ method: 'item/started', params: { item: { id: 'call-1', type, ...input } } }))
.toEqual([expect.objectContaining({ type: 'tool-call', name, input })]);
const settled = live.onNotification({ method: 'item/completed', params: { item: completed } });
expect(settled).toHaveLength(1);
expect(settled[0]).toMatchObject({ type: 'tool-result', callId: 'call-1', output: result });
});
it('keeps native completed command output in the tool result when no started event was observed', () => {
const updates = createCodexAppServerStreamEventBridge().onNotification({ method: 'item/completed', params: {
item: { id: 'native-command', type: 'commandExecution', command: 'pwd', cwd: '/project',
Expand Down Expand Up @@ -39,7 +67,12 @@ describe('createCodexAppServerStreamEventBridge', () => {
type: 'provider-user-text',
itemId: 'native_user_1',
clientId: null,
text: 'hello from Codex TUI\nsecond paragraph',
text: 'hello from Codex TUI\n[Codex mention: README.md (/repo/README.md)]\nsecond paragraph',
nativeContent: [
{ type: 'text', text: 'hello from Codex TUI' },
{ type: 'mention', name: 'README.md', path: '/repo/README.md' },
{ type: 'text', text: 'second paragraph' },
],
}]);

expect(
Expand Down
22 changes: 17 additions & 5 deletions apps/cli/src/backends/codex/appServer/streamEventBridge.ts
Original file line number Diff line number Diff line change
@@ -1,13 +1,14 @@
import { readCodexReasoningText } from '../utils/readCodexReasoningText';
import { extractMcpToolCallResultOutput } from '../runtime/sessionTurnLifecycle';
import { readCodexMessageContentText } from '../utils/readCodexMessageContentText';
import { readCodexUserMessageContent } from '../utils/readCodexUserMessageContent';
import { canonicalizeCodexMcpToolName } from '../utils/canonicalizeCodexMcpToolName';
import { extractCodexGeneratedMedia } from '../media/extractCodexGeneratedMedia';
import type { SessionMediaSource } from '@/agent/core/AgentMessage';

type RecordLike = Record<string, unknown>;

type ToolKind = 'command' | 'mcp' | 'file-change' | 'web-search' | 'image-view' | 'sleep';
type ToolKind = 'command' | 'mcp' | 'file-change' | 'web-search' | 'image-view' | 'sleep' | 'native-agent' | 'dynamic';
type PermissionRequestToolName = 'request_permissions';

type NotificationEnvelope = Readonly<{
Expand All @@ -27,7 +28,7 @@ type ToolContext = Readonly<{
}>;

export type CodexAppServerStreamUpdate =
| Readonly<{ type: 'provider-user-text'; itemId: string; clientId: string | null; text: string }>
| Readonly<{ type: 'provider-user-text'; itemId: string; clientId: string | null; text: string; nativeContent?: readonly Record<string, unknown>[] }>
| Readonly<{ type: 'assistant-text-delta'; itemId: string; text: string }>
| Readonly<{ type: 'assistant-text-final'; itemId: string; text: string }>
| Readonly<{ type: 'async-user-input-request'; itemId: string; text: string; questions: unknown[] }>
Expand Down Expand Up @@ -140,6 +141,17 @@ function omitKeys(value: RecordLike, keys: readonly string[]): RecordLike {

function readToolContextFromItem(item: RecordLike): ToolContext | null {
const itemType = readItemType(item);
// These native records describe an activity/tool occurrence, not completion
// of the child agent itself. Keep its identity and reported state intact.
if (itemType === 'subagentactivity' || itemType === 'collabagenttoolcall' || itemType === 'dynamictoolcall') {
const name = itemType === 'subagentactivity' ? 'subAgentActivity' : readString(item.tool);
if (!name) return null;
return {
toolKind: itemType === 'dynamictoolcall' ? 'dynamic' : 'native-agent',
name,
input: omitKeys(item, ['id', 'itemId', 'type', 'itemType', 'status', 'agentsStates', 'contentItems', 'success', 'durationMs', 'completedAtMs', 'error']),
};
}
if (itemType === 'commandexecution') {
const command = readString(item.command);
const cwd = readString(item.cwd);
Expand Down Expand Up @@ -269,13 +281,13 @@ export function createCodexAppServerStreamEventBridge(): Readonly<{
if (item) {
if (readItemType(item) === 'usermessage') {
const itemId = readItemId(item);
const text = readCodexMessageContentText(item.content);
if (itemId && text) {
const content = readCodexUserMessageContent(item.content);
if (itemId && content) {
return [{
type: 'provider-user-text',
itemId,
clientId: readString(item.clientId) ?? readString(item.client_id),
text,
...content,
}];
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,7 @@ export function mapCodexProjectedEventsToDirectMessages(params: Readonly<{
raw: {
role: 'user',
content: { type: 'text', text: action.text },
...(action.nativeContent ? { meta: { codexNativeContent: action.nativeContent } } : {}),
},
});
continue;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -108,8 +108,9 @@ export async function pageCodexNativeTranscript(params: Readonly<{
for (; offset < messages.length; offset += 1) {
const item = messages[offset]!;
const size = measureDirectTranscriptItemBytes(item);
if (items.length >= params.maxItems || bytes + size > params.maxBytes) {
if (items.length === 0) throw new Error('Codex native transcript item exceeds the requested byte budget');
// Match rollout stream paging: the byte budget bounds aggregation,
// not a valid indivisible record. Return a larger record on its own.
if (items.length > 0 && (items.length >= params.maxItems || bytes + size > params.maxBytes)) {
const nextCursor = encode({ ...state, skip: index, offset, pendingId: key,
anchor: index ? identity(page.data[index - 1]!) : null,
fingerprint: index ? fingerprint(page.data[index - 1]!) : null });
Expand Down
Loading
Loading