wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-08-06 06:41:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: new codex remote mode

Browse Source
This commit is contained in:
weishu
2026-01-26 22:22:32 +08:00
parent 6acde5a9b9
commit 3a1379dc01
8 changed files with 1644 additions and 356 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/src/codex/appServerTypes.ts
+135
View File
@@ -0,0 +1,135 @@
export type ApprovalPolicy = 'untrusted' | 'on-failure' | 'on-request' | 'never';
export type SandboxMode = 'read-only' | 'workspace-write' | 'danger-full-access';
export interface InitializeParams {
clientInfo: {
name: string;
title?: string;
version: string;
};
}
export interface InitializeResponse {
userAgent?: string;
[key: string]: unknown;
}
export interface ThreadStartParams {
model?: string;
modelProvider?: string;
cwd?: string;
approvalPolicy?: ApprovalPolicy;
sandbox?: SandboxMode;
config?: Record<string, unknown>;
baseInstructions?: string;
developerInstructions?: string;
personality?: string;
ephemeral?: boolean;
experimentalRawEvents?: boolean;
}
export interface ThreadStartResponse {
thread: {
id: string;
};
[key: string]: unknown;
}
export type ResponseItem = Record<string, unknown>;
export interface ThreadResumeParams {
threadId: string;
history?: ResponseItem[];
path?: string;
model?: string;
modelProvider?: string;
cwd?: string;
approvalPolicy?: ApprovalPolicy;
sandbox?: SandboxMode;
config?: Record<string, unknown>;
baseInstructions?: string;
developerInstructions?: string;
personality?: string;
}
export interface ThreadResumeResponse {
thread: {
id: string;
};
[key: string]: unknown;
}
export type UserInput =
| {
type: 'text';
text: string;
textElements?: Array<{
byteRange: { start: number; end: number };
placeholder?: string;
}>;
}
| {
type: 'image';
url: string;
}
| {
type: 'localImage';
path: string;
}
| {
type: 'skill';
name: string;
path: string;
};
export type SandboxPolicy =
| { type: 'dangerFullAccess' }
| { type: 'readOnly' }
| { type: 'externalSandbox'; networkAccess?: 'restricted' | 'enabled' }
| {
type: 'workspaceWrite';
writableRoots?: string[];
networkAccess?: boolean;
excludeTmpdirEnvVar?: boolean;
excludeSlashTmp?: boolean;
};
export type ReasoningEffort = 'low' | 'medium' | 'high' | 'auto';
export type ReasoningSummary = 'auto' | 'none' | 'brief' | 'detailed';
export type CollaborationMode = {
mode: 'plan' | 'code' | 'pair_programming' | 'execute' | 'custom' | (string & {});
settings?: Record<string, unknown>;
};
export interface TurnStartParams {
threadId: string;
input: UserInput[];
cwd?: string;
approvalPolicy?: ApprovalPolicy;
sandboxPolicy?: SandboxPolicy;
model?: string;
effort?: ReasoningEffort;
summary?: ReasoningSummary;
personality?: string;
outputSchema?: unknown;
collaborationMode?: CollaborationMode;
}
export interface TurnStartResponse {
turn: {
id: string;
status?: string;
};
[key: string]: unknown;
}
export interface TurnInterruptParams {
threadId: string;
turnId: string;
}
export interface TurnInterruptResponse {
ok: boolean;
[key: string]: unknown;
}
cli/src/codex/codexAppServerClient.ts
+409
View File
@@ -0,0 +1,409 @@
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process';
import { logger } from '@/ui/logger';
import { killProcessByChildProcess } from '@/utils/process';
import type {
InitializeParams,
InitializeResponse,
ThreadStartParams,
ThreadStartResponse,
ThreadResumeParams,
ThreadResumeResponse,
TurnStartParams,
TurnStartResponse,
TurnInterruptParams,
TurnInterruptResponse
} from './appServerTypes';
type JsonRpcLiteRequest = {
id: number;
method: string;
params?: unknown;
};
type JsonRpcLiteNotification = {
method: string;
params?: unknown;
};
type JsonRpcLiteResponse = {
id: number | string | null;
result?: unknown;
error?: {
code?: number;
message: string;
data?: unknown;
};
};
type RequestHandler = (params: unknown) => Promise<unknown> | unknown;
type PendingRequest = {
resolve: (value: unknown) => void;
reject: (error: Error) => void;
cleanup: () => void;
};
function asRecord(value: unknown): Record<string, unknown> | null {
if (!value || typeof value !== 'object') {
return null;
}
return value as Record<string, unknown>;
}
function createAbortError(): Error {
const error = new Error('Request aborted');
error.name = 'AbortError';
return error;
}
export class CodexAppServerClient {
private process: ChildProcessWithoutNullStreams | null = null;
private connected = false;
private buffer = '';
private nextId = 1;
private readonly pending = new Map<number, PendingRequest>();
private readonly requestHandlers = new Map<string, RequestHandler>();
private notificationHandler: ((method: string, params: unknown) => void) | null = null;
private protocolError: Error | null = null;
static readonly DEFAULT_TIMEOUT_MS = 14 * 24 * 60 * 60 * 1000;
async connect(): Promise<void> {
if (this.connected) {
return;
}
this.process = spawn('codex', ['app-server'], {
env: Object.keys(process.env).reduce((acc, key) => {
const value = process.env[key];
if (typeof value === 'string') acc[key] = value;
return acc;
}, {} as Record<string, string>),
stdio: ['pipe', 'pipe', 'pipe'],
shell: process.platform === 'win32'
});
this.process.stdout.setEncoding('utf8');
this.process.stdout.on('data', (chunk) => this.handleStdout(chunk));
this.process.stderr.setEncoding('utf8');
this.process.stderr.on('data', (chunk) => {
const text = chunk.toString().trim();
if (text.length > 0) {
logger.debug(`[CodexAppServer][stderr] ${text}`);
}
});
this.process.on('exit', (code, signal) => {
const message = `Codex app-server exited (code=${code ?? 'null'}, signal=${signal ?? 'null'})`;
logger.debug(message);
this.rejectAllPending(new Error(message));
this.connected = false;
this.resetParserState();
this.process = null;
});
this.process.on('error', (error) => {
logger.debug('[CodexAppServer] Process error', error);
const message = error instanceof Error ? error.message : String(error);
this.rejectAllPending(new Error(
`Failed to spawn codex app-server: ${message}. Is it installed and on PATH?`,
{ cause: error }
));
this.connected = false;
this.resetParserState();
this.process = null;
});
this.connected = true;
logger.debug('[CodexAppServer] Connected');
}
setNotificationHandler(handler: ((method: string, params: unknown) => void) | null): void {
this.notificationHandler = handler;
}
registerRequestHandler(method: string, handler: RequestHandler): void {
this.requestHandlers.set(method, handler);
}
async initialize(params: InitializeParams): Promise<InitializeResponse> {
const response = await this.sendRequest('initialize', params, { timeoutMs: 30_000 });
this.sendNotification('initialized');
return response as InitializeResponse;
}
async startThread(params: ThreadStartParams, options?: { signal?: AbortSignal }): Promise<ThreadStartResponse> {
const response = await this.sendRequest('thread/start', params, {
signal: options?.signal,
timeoutMs: CodexAppServerClient.DEFAULT_TIMEOUT_MS
});
return response as ThreadStartResponse;
}
async resumeThread(params: ThreadResumeParams, options?: { signal?: AbortSignal }): Promise<ThreadResumeResponse> {
const response = await this.sendRequest('thread/resume', params, {
signal: options?.signal,
timeoutMs: CodexAppServerClient.DEFAULT_TIMEOUT_MS
});
return response as ThreadResumeResponse;
}
async startTurn(params: TurnStartParams, options?: { signal?: AbortSignal }): Promise<TurnStartResponse> {
const response = await this.sendRequest('turn/start', params, {
signal: options?.signal,
timeoutMs: CodexAppServerClient.DEFAULT_TIMEOUT_MS
});
return response as TurnStartResponse;
}
async interruptTurn(params: TurnInterruptParams): Promise<TurnInterruptResponse> {
const response = await this.sendRequest('turn/interrupt', params, {
timeoutMs: 30_000
});
return response as TurnInterruptResponse;
}
async disconnect(): Promise<void> {
if (!this.connected) {
return;
}
const child = this.process;
this.process = null;
try {
child?.stdin.end();
if (child) {
await killProcessByChildProcess(child);
}
} catch (error) {
logger.debug('[CodexAppServer] Error while stopping process', error);
} finally {
this.rejectAllPending(new Error('Codex app-server disconnected'));
this.connected = false;
this.resetParserState();
}
logger.debug('[CodexAppServer] Disconnected');
}
private async sendRequest(
method: string,
params?: unknown,
options?: { signal?: AbortSignal; timeoutMs?: number }
): Promise<unknown> {
if (!this.connected) {
await this.connect();
}
const id = this.nextId++;
const payload: JsonRpcLiteRequest = {
id,
method,
params
};
const timeoutMs = options?.timeoutMs ?? CodexAppServerClient.DEFAULT_TIMEOUT_MS;
return new Promise((resolve, reject) => {
let timeout: ReturnType<typeof setTimeout> | null = null;
let aborted = false;
const cleanup = () => {
if (timeout) {
clearTimeout(timeout);
}
if (options?.signal) {
options.signal.removeEventListener('abort', onAbort);
}
};
const onAbort = () => {
if (aborted) return;
aborted = true;
this.pending.delete(id);
cleanup();
reject(createAbortError());
};
if (options?.signal) {
if (options.signal.aborted) {
onAbort();
return;
}
options.signal.addEventListener('abort', onAbort, { once: true });
}
if (Number.isFinite(timeoutMs)) {
timeout = setTimeout(() => {
if (this.pending.has(id)) {
this.pending.delete(id);
cleanup();
reject(new Error(`Codex app-server request '${method}' timed out after ${timeoutMs}ms`));
}
}, timeoutMs);
timeout.unref();
}
this.pending.set(id, {
resolve: (value) => {
cleanup();
resolve(value);
},
reject: (error) => {
cleanup();
reject(error);
},
cleanup
});
this.writePayload(payload);
});
}
private sendNotification(method: string, params?: unknown): void {
const payload: JsonRpcLiteNotification = { method, params };
this.writePayload(payload);
}
private handleStdout(chunk: string): void {
this.buffer += chunk;
let newlineIndex = this.buffer.indexOf('\n');
while (newlineIndex >= 0) {
const line = this.buffer.slice(0, newlineIndex).trim();
this.buffer = this.buffer.slice(newlineIndex + 1);
if (line.length > 0) {
this.handleLine(line);
}
newlineIndex = this.buffer.indexOf('\n');
}
}
private handleLine(line: string): void {
if (this.protocolError) {
return;
}
let message: Record<string, unknown> | null = null;
try {
const parsed = JSON.parse(line);
message = asRecord(parsed);
if (!message) {
logger.debug('[CodexAppServer] Ignoring non-object JSON from stdout', { line });
return;
}
} catch (error) {
const protocolError = new Error('Failed to parse JSON from codex app-server');
this.protocolError = protocolError;
logger.debug('[CodexAppServer] Failed to parse JSON line', { line, error });
this.rejectAllPending(protocolError);
this.process?.stdin.end();
return;
}
if (typeof message.method === 'string') {
const method = message.method;
const params = 'params' in message ? message.params : null;
if ('id' in message && message.id !== undefined) {
const requestId = message.id;
void this.handleIncomingRequest({
id: requestId,
method,
params
});
return;
}
this.notificationHandler?.(method, params ?? null);
return;
}
if ('id' in message) {
this.handleResponse(message as JsonRpcLiteResponse);
}
}
private async handleIncomingRequest(request: { id: unknown; method: string; params?: unknown }): Promise<void> {
const responseId = typeof request.id === 'number' || typeof request.id === 'string'
? request.id
: null;
const handler = this.requestHandlers.get(request.method);
if (!handler) {
this.writePayload({
id: responseId,
error: {
code: -32601,
message: `Method not found: ${request.method}`
}
} satisfies JsonRpcLiteResponse);
return;
}
try {
const result = await handler(request.params ?? null);
this.writePayload({
id: responseId,
result
} satisfies JsonRpcLiteResponse);
} catch (error) {
this.writePayload({
id: responseId,
error: {
code: -32603,
message: error instanceof Error ? error.message : 'Internal error'
}
} satisfies JsonRpcLiteResponse);
}
}
private handleResponse(response: JsonRpcLiteResponse): void {
if (response.id === null || response.id === undefined) {
logger.debug('[CodexAppServer] Received response without id');
return;
}
if (typeof response.id !== 'number') {
logger.debug('[CodexAppServer] Received response with non-numeric id', response.id);
return;
}
const pending = this.pending.get(response.id);
if (!pending) {
logger.debug('[CodexAppServer] Received response with no pending request', response.id);
return;
}
this.pending.delete(response.id);
if (response.error) {
pending.reject(new Error(response.error.message));
return;
}
pending.resolve(response.result);
}
private writePayload(payload: JsonRpcLiteRequest | JsonRpcLiteNotification | JsonRpcLiteResponse): void {
const serialized = JSON.stringify(payload);
this.process?.stdin.write(`${serialized}\n`);
}
private resetParserState(): void {
this.buffer = '';
this.protocolError = null;
}
private rejectAllPending(error: Error): void {
for (const { reject, cleanup } of this.pending.values()) {
cleanup();
reject(error);
}
this.pending.clear();
}
}
cli/src/codex/codexRemoteLauncher.ts
+377 -356
View File
@@ -1,10 +1,8 @@
import React from 'react';
import { randomUUID } from 'node:crypto';
import os from 'node:os';
import fs from 'node:fs';
import { join } from 'node:path';
import { CodexMcpClient } from './codexMcpClient';
import { CodexAppServerClient } from './codexAppServerClient';
import { CodexPermissionHandler } from './utils/permissionHandler';
import { ReasoningProcessor } from './utils/reasoningProcessor';
import { DiffProcessor } from './utils/diffProcessor';
@@ -17,7 +15,9 @@ import type { CodexSession } from './session';
import type { EnhancedMode } from './loop';
import { hasCodexCliOverrides } from './utils/codexCliOverrides';
import { buildCodexStartConfig } from './utils/codexStartConfig';
import { convertCodexEvent } from './utils/codexEventConverter';
import { AppServerEventConverter } from './utils/appServerEventConverter';
import { registerAppServerPermissionHandlers } from './utils/appServerPermissionAdapter';
import { buildThreadStartParams, buildTurnStartParams } from './utils/appServerConfig';
import {
RemoteLauncherBase,
type RemoteLauncherDisplayContext,
@@ -26,20 +26,30 @@ import {
type HappyServer = Awaited<ReturnType<typeof buildHapiMcpBridge>>['server'];
function shouldUseAppServer(): boolean {
const useMcpServer = process.env.CODEX_USE_MCP_SERVER === '1';
return !useMcpServer;
}
class CodexRemoteLauncher extends RemoteLauncherBase {
private readonly session: CodexSession;
private readonly client: CodexMcpClient;
private readonly useAppServer: boolean;
private readonly mcpClient: CodexMcpClient | null;
private readonly appServerClient: CodexAppServerClient | null;
private permissionHandler: CodexPermissionHandler | null = null;
private reasoningProcessor: ReasoningProcessor | null = null;
private diffProcessor: DiffProcessor | null = null;
private happyServer: HappyServer | null = null;
private abortController: AbortController = new AbortController();
private storedSessionIdForResume: string | null = null;
private currentThreadId: string | null = null;
private currentTurnId: string | null = null;
constructor(session: CodexSession) {
super(process.env.DEBUG ? session.logPath : undefined);
this.session = session;
this.client = new CodexMcpClient();
this.useAppServer = shouldUseAppServer();
this.mcpClient = this.useAppServer ? null : new CodexMcpClient();
this.appServerClient = this.useAppServer ? new CodexAppServerClient() : null;
}
protected createDisplay(context: RemoteLauncherDisplayContext): React.ReactElement {
@@ -49,9 +59,19 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
private async handleAbort(): Promise<void> {
logger.debug('[Codex] Abort requested - stopping current task');
try {
if (this.client.hasActiveSession()) {
this.storedSessionIdForResume = this.client.storeSessionForResume();
logger.debug('[Codex] Stored session for resume:', this.storedSessionIdForResume);
if (this.useAppServer && this.appServerClient) {
if (this.currentThreadId && this.currentTurnId) {
try {
await this.appServerClient.interruptTurn({
threadId: this.currentThreadId,
turnId: this.currentTurnId
});
} catch (error) {
logger.debug('[Codex] Error interrupting app-server turn:', error);
}
}
this.currentTurnId = null;
}
this.abortController.abort();
@@ -107,52 +127,10 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
protected async runMainLoop(): Promise<void> {
const session = this.session;
const messageBuffer = this.messageBuffer;
const client = this.client;
function findCodexResumeFile(sessionId: string | null): string | null {
if (!sessionId) return null;
try {
const codexHomeDir = process.env.CODEX_HOME || join(os.homedir(), '.codex');
const rootDir = join(codexHomeDir, 'sessions');
function collectFilesRecursive(dir: string, acc: string[] = []): string[] {
let entries: fs.Dirent[];
try {
entries = fs.readdirSync(dir, { withFileTypes: true });
} catch {
return acc;
}
for (const entry of entries) {
const full = join(dir, entry.name);
if (entry.isDirectory()) {
collectFilesRecursive(full, acc);
} else if (entry.isFile()) {
acc.push(full);
}
}
return acc;
}
const candidates = collectFilesRecursive(rootDir)
.filter((full) => full.endsWith(`-${sessionId}.jsonl`))
.filter((full) => {
try { return fs.statSync(full).isFile(); } catch { return false; }
})
.sort((a, b) => {
const sa = fs.statSync(a).mtimeMs;
const sb = fs.statSync(b).mtimeMs;
return sb - sa;
});
return candidates[0] || null;
} catch {
return null;
}
}
const RESUME_CONTEXT_MAX_ITEMS = 40;
const RESUME_CONTEXT_MAX_CHARS = 16000;
const RESUME_CONTEXT_TOOL_MAX_CHARS = 2000;
const RESUME_CONTEXT_REASONING_MAX_CHARS = 2000;
const useAppServer = this.useAppServer;
const mcpClient = this.mcpClient;
const appServerClient = this.appServerClient;
const appServerEventConverter = useAppServer ? new AppServerEventConverter() : null;
const normalizeCommand = (value: unknown): string | undefined => {
if (typeof value === 'string') {
@@ -166,150 +144,27 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
return undefined;
};
const normalizeCwd = (value: unknown): string | undefined => {
if (typeof value !== 'string') return undefined;
const trimmed = value.trim();
return trimmed.length > 0 ? trimmed : undefined;
};
const permissionDetails = new Map<string, { command?: string; cwd?: string }>();
const recordPermissionDetails = (id: string, command?: string, cwd?: string) => {
const existing = permissionDetails.get(id) ?? {};
const next = {
command: command ?? existing.command,
cwd: cwd ?? existing.cwd
};
permissionDetails.set(id, next);
return next;
};
const getPermissionDetails = (id: string) => permissionDetails.get(id) ?? {};
function readResumeFileContent(resumeFile: string): { content: string; truncated: boolean } | null {
try {
const stat = fs.statSync(resumeFile);
if (!stat.isFile()) {
return null;
}
return { content: fs.readFileSync(resumeFile, 'utf8'), truncated: false };
} catch (error) {
logger.debug('[Codex] Failed to read resume file:', error);
const asRecord = (value: unknown): Record<string, unknown> | null => {
if (!value || typeof value !== 'object') {
return null;
}
}
return value as Record<string, unknown>;
};
function safeStringify(value: unknown): string | null {
if (value === null || value === undefined) {
return null;
}
if (typeof value === 'string') {
return value;
}
const asString = (value: unknown): string | null => {
return typeof value === 'string' && value.length > 0 ? value : null;
};
const formatOutputPreview = (value: unknown): string => {
if (typeof value === 'string') return value;
if (typeof value === 'number' || typeof value === 'boolean') return String(value);
if (value === null || value === undefined) return '';
try {
return JSON.stringify(value);
} catch {
return null;
return String(value);
}
}
function formatResumeValue(value: unknown, maxChars: number, singleLine = false): string | null {
const raw = safeStringify(value);
if (!raw) {
return null;
}
const normalized = singleLine ? raw.replace(/\s+/g, ' ').trim() : raw;
if (!normalized) {
return null;
}
if (normalized.length <= maxChars) {
return normalized;
}
return `${normalized.slice(0, maxChars)}...`;
}
function buildResumeInstructionsFromFile(resumeFile: string): string | undefined {
const result = readResumeFileContent(resumeFile);
if (!result) {
return undefined;
}
const items: { role: 'user' | 'assistant' | 'tool'; text: string }[] = [];
let truncated = result.truncated;
const lines = result.content.split('\n');
for (const line of lines) {
const trimmed = line.trim();
if (!trimmed) {
continue;
}
try {
const parsed = JSON.parse(trimmed);
const converted = convertCodexEvent(parsed);
if (converted?.userMessage) {
items.push({ role: 'user', text: converted.userMessage });
}
if (converted?.message?.type === 'message') {
items.push({ role: 'assistant', text: converted.message.message });
}
if (converted?.message?.type === 'reasoning') {
const reasoning = formatResumeValue(converted.message.message, RESUME_CONTEXT_REASONING_MAX_CHARS);
if (reasoning) {
items.push({ role: 'assistant', text: `Reasoning: ${reasoning}` });
}
}
if (converted?.message?.type === 'tool-call') {
const input = formatResumeValue(converted.message.input, RESUME_CONTEXT_TOOL_MAX_CHARS, true);
const text = input
? `Call ${converted.message.name} ${input}`
: `Call ${converted.message.name}`;
items.push({ role: 'tool', text });
}
if (converted?.message?.type === 'tool-call-result') {
const output = formatResumeValue(converted.message.output, RESUME_CONTEXT_TOOL_MAX_CHARS, true);
if (output) {
items.push({ role: 'tool', text: `Result ${output}` });
}
}
} catch {
continue;
}
}
if (items.length === 0) {
return undefined;
}
if (items.length > RESUME_CONTEXT_MAX_ITEMS) {
items.splice(0, items.length - RESUME_CONTEXT_MAX_ITEMS);
truncated = true;
}
const rendered = items.map((item) => {
if (item.role === 'user') {
return `User: ${item.text}`;
}
if (item.role === 'tool') {
return `Tool: ${item.text}`;
}
return `Assistant: ${item.text}`;
});
let totalChars = rendered.reduce((sum, line) => sum + line.length + 1, 0);
while (rendered.length > 1 && totalChars > RESUME_CONTEXT_MAX_CHARS) {
const removed = rendered.shift();
totalChars -= (removed?.length ?? 0) + 1;
truncated = true;
}
if (rendered.length === 0) {
return undefined;
}
const header = truncated
? 'Continue from the prior session context below (transcript truncated):'
: 'Continue from the prior session context below:';
return `${header}\n${rendered.join('\n')}`;
}
};
const permissionHandler = new CodexPermissionHandler(session.client, {
onRequest: ({ id, toolName, input }) => {
@@ -360,140 +215,234 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
this.reasoningProcessor = reasoningProcessor;
this.diffProcessor = diffProcessor;
client.setPermissionHandler(permissionHandler);
client.setHandler((msg) => {
logger.debug(`[Codex] MCP message: ${JSON.stringify(msg)}`);
const handleCodexEvent = (msg: Record<string, unknown>) => {
const msgType = asString(msg.type);
if (!msgType) return;
const msgType = typeof msg?.type === 'string' ? msg.type : null;
if (msgType === 'event_msg' || msgType === 'response_item' || msgType === 'session_meta') {
const payloadType = typeof msg?.payload?.type === 'string' ? msg.payload.type : null;
logger.debug(`[Codex] MCP wrapper event type: ${msgType}${payloadType ? ` (payload=${payloadType})` : ''}`);
if (msgType === 'thread_started') {
const threadId = asString(msg.thread_id ?? msg.threadId);
if (threadId) {
this.currentThreadId = threadId;
session.onSessionFound(threadId);
}
return;
}
if (msg.type === 'agent_message') {
messageBuffer.addMessage(msg.message, 'assistant');
} else if (msg.type === 'agent_reasoning_delta') {
} else if (msg.type === 'agent_reasoning') {
messageBuffer.addMessage(`[Thinking] ${msg.text.substring(0, 100)}...`, 'system');
} else if (msg.type === 'exec_command_begin') {
messageBuffer.addMessage(`Executing: ${msg.command}`, 'tool');
} else if (msg.type === 'exec_command_end') {
const output = msg.output || msg.error || 'Command completed';
const truncatedOutput = output.substring(0, 200);
if (msgType === 'task_started') {
const turnId = asString(msg.turn_id ?? msg.turnId);
if (turnId) {
this.currentTurnId = turnId;
}
}
if (msgType === 'task_complete' || msgType === 'turn_aborted' || msgType === 'task_failed') {
this.currentTurnId = null;
}
if (!useAppServer) {
logger.debug(`[Codex] MCP message: ${JSON.stringify(msg)}`);
if (msgType === 'event_msg' || msgType === 'response_item' || msgType === 'session_meta') {
const payload = asRecord(msg.payload);
const payloadType = asString(payload?.type);
logger.debug(`[Codex] MCP wrapper event type: ${msgType}${payloadType ? ` (payload=${payloadType})` : ''}`);
}
}
if (msgType === 'agent_message') {
const message = asString(msg.message);
if (message) {
messageBuffer.addMessage(message, 'assistant');
}
} else if (msgType === 'agent_reasoning') {
const text = asString(msg.text);
if (text) {
messageBuffer.addMessage(`[Thinking] ${text.substring(0, 100)}...`, 'system');
}
} else if (msgType === 'exec_command_begin') {
const command = normalizeCommand(msg.command) ?? 'command';
messageBuffer.addMessage(`Executing: ${command}`, 'tool');
} else if (msgType === 'exec_command_end') {
const output = msg.output ?? msg.error ?? 'Command completed';
const outputText = formatOutputPreview(output);
const truncatedOutput = outputText.substring(0, 200);
messageBuffer.addMessage(
`Result: ${truncatedOutput}${output.length > 200 ? '...' : ''}`,
`Result: ${truncatedOutput}${outputText.length > 200 ? '...' : ''}`,
'result'
);
} else if (msg.type === 'task_started') {
} else if (msgType === 'task_started') {
messageBuffer.addMessage('Starting task...', 'status');
} else if (msg.type === 'task_complete') {
} else if (msgType === 'task_complete') {
messageBuffer.addMessage('Task completed', 'status');
sendReady();
} else if (msg.type === 'turn_aborted') {
} else if (msgType === 'turn_aborted') {
messageBuffer.addMessage('Turn aborted', 'status');
sendReady();
} else if (msgType === 'task_failed') {
const error = asString(msg.error);
messageBuffer.addMessage(error ? `Task failed: ${error}` : 'Task failed', 'status');
sendReady();
}
if (msg.type === 'task_started') {
if (msgType === 'task_started') {
if (useAppServer) {
turnInFlight = true;
}
if (!session.thinking) {
logger.debug('thinking started');
session.onThinkingChange(true);
}
}
if (msg.type === 'task_complete' || msg.type === 'turn_aborted') {
if (msgType === 'task_complete' || msgType === 'turn_aborted' || msgType === 'task_failed') {
if (useAppServer) {
turnInFlight = false;
}
if (session.thinking) {
logger.debug('thinking completed');
session.onThinkingChange(false);
}
diffProcessor.reset();
appServerEventConverter?.reset();
}
if (msg.type === 'agent_reasoning_section_break') {
if (msgType === 'agent_reasoning_section_break') {
reasoningProcessor.handleSectionBreak();
}
if (msg.type === 'agent_reasoning_delta') {
reasoningProcessor.processDelta(msg.delta);
if (msgType === 'agent_reasoning_delta') {
const delta = asString(msg.delta);
if (delta) {
reasoningProcessor.processDelta(delta);
}
}
if (msg.type === 'agent_reasoning') {
reasoningProcessor.complete(msg.text);
if (msgType === 'agent_reasoning') {
const text = asString(msg.text);
if (text) {
reasoningProcessor.complete(text);
}
}
if (msg.type === 'agent_message') {
session.sendCodexMessage({
type: 'message',
message: msg.message,
id: randomUUID()
});
if (msgType === 'agent_message') {
const message = asString(msg.message);
if (message) {
session.sendCodexMessage({
type: 'message',
message,
id: randomUUID()
});
}
}
if (msg.type === 'exec_command_begin' || msg.type === 'exec_approval_request') {
const { call_id, type, ...inputs } = msg;
session.sendCodexMessage({
type: 'tool-call',
name: 'CodexBash',
callId: call_id,
input: inputs,
id: randomUUID()
});
if (msgType === 'exec_command_begin' || msgType === 'exec_approval_request') {
const callId = asString(msg.call_id ?? msg.callId);
if (callId) {
const inputs: Record<string, unknown> = { ...msg };
delete inputs.type;
delete inputs.call_id;
delete inputs.callId;
session.sendCodexMessage({
type: 'tool-call',
name: 'CodexBash',
callId: callId,
input: inputs,
id: randomUUID()
});
}
}
if (msg.type === 'exec_command_end') {
const { call_id, type, ...output } = msg;
session.sendCodexMessage({
type: 'tool-call-result',
callId: call_id,
output,
id: randomUUID()
});
if (msgType === 'exec_command_end') {
const callId = asString(msg.call_id ?? msg.callId);
if (callId) {
const output: Record<string, unknown> = { ...msg };
delete output.type;
delete output.call_id;
delete output.callId;
session.sendCodexMessage({
type: 'tool-call-result',
callId: callId,
output,
id: randomUUID()
});
}
}
if (msg.type === 'token_count') {
if (msgType === 'token_count') {
session.sendCodexMessage({
...msg,
id: randomUUID()
});
}
if (msg.type === 'patch_apply_begin') {
const { call_id, auto_approved, changes } = msg;
if (msgType === 'patch_apply_begin') {
const callId = asString(msg.call_id ?? msg.callId);
if (callId) {
const changes = asRecord(msg.changes) ?? {};
const changeCount = Object.keys(changes).length;
const filesMsg = changeCount === 1 ? '1 file' : `${changeCount} files`;
messageBuffer.addMessage(`Modifying ${filesMsg}...`, 'tool');
const changeCount = Object.keys(changes).length;
const filesMsg = changeCount === 1 ? '1 file' : `${changeCount} files`;
messageBuffer.addMessage(`Modifying ${filesMsg}...`, 'tool');
session.sendCodexMessage({
type: 'tool-call',
name: 'CodexPatch',
callId: call_id,
input: {
auto_approved,
changes
},
id: randomUUID()
});
}
if (msg.type === 'patch_apply_end') {
const { call_id, stdout, stderr, success } = msg;
if (success) {
const message = stdout || 'Files modified successfully';
messageBuffer.addMessage(message.substring(0, 200), 'result');
} else {
const errorMsg = stderr || 'Failed to modify files';
messageBuffer.addMessage(`Error: ${errorMsg.substring(0, 200)}`, 'result');
}
session.sendCodexMessage({
type: 'tool-call-result',
callId: call_id,
output: {
stdout,
stderr,
success
},
id: randomUUID()
});
}
if (msg.type === 'turn_diff') {
if (msg.unified_diff) {
diffProcessor.processDiff(msg.unified_diff);
session.sendCodexMessage({
type: 'tool-call',
name: 'CodexPatch',
callId: callId,
input: {
auto_approved: msg.auto_approved ?? msg.autoApproved,
changes
},
id: randomUUID()
});
}
}
});
if (msgType === 'patch_apply_end') {
const callId = asString(msg.call_id ?? msg.callId);
if (callId) {
const stdout = asString(msg.stdout);
const stderr = asString(msg.stderr);
const success = Boolean(msg.success);
if (success) {
const message = stdout || 'Files modified successfully';
messageBuffer.addMessage(message.substring(0, 200), 'result');
} else {
const errorMsg = stderr || 'Failed to modify files';
messageBuffer.addMessage(`Error: ${errorMsg.substring(0, 200)}`, 'result');
}
session.sendCodexMessage({
type: 'tool-call-result',
callId: callId,
output: {
stdout,
stderr,
success
},
id: randomUUID()
});
}
}
if (msgType === 'turn_diff') {
const diff = asString(msg.unified_diff);
if (diff) {
diffProcessor.processDiff(diff);
}
}
};
if (useAppServer && appServerClient && appServerEventConverter) {
registerAppServerPermissionHandlers({
client: appServerClient,
permissionHandler
});
appServerClient.setNotificationHandler((method, params) => {
const events = appServerEventConverter.handleNotification(method, params);
for (const event of events) {
const eventRecord = asRecord(event) ?? { type: undefined };
handleCodexEvent(eventRecord);
}
});
} else if (mcpClient) {
mcpClient.setPermissionHandler(permissionHandler);
mcpClient.setHandler((msg) => {
const eventRecord = asRecord(msg) ?? { type: undefined };
handleCodexEvent(eventRecord);
});
}
const { server: happyServer, mcpServers } = await buildHapiMcpBridge(session.client);
this.happyServer = happyServer;
@@ -520,19 +469,30 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
};
const syncSessionId = () => {
const clientSessionId = client.getSessionId();
if (!mcpClient) return;
const clientSessionId = mcpClient.getSessionId();
if (clientSessionId && clientSessionId !== session.sessionId) {
session.onSessionFound(clientSessionId);
}
};
await client.connect();
if (useAppServer && appServerClient) {
await appServerClient.connect();
await appServerClient.initialize({
clientInfo: {
name: 'hapi-codex-client',
version: '1.0.0'
}
});
} else if (mcpClient) {
await mcpClient.connect();
}
let wasCreated = false;
let currentModeHash: string | null = null;
let pending: { message: string; mode: EnhancedMode; isolate: boolean; hash: string } | null = null;
let nextExperimentalResume: string | null = null;
let first = true;
let turnInFlight = false;
while (!this.shouldExit) {
logActiveHandles('loop-top');
@@ -556,23 +516,11 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
break;
}
if (wasCreated && currentModeHash && message.hash !== currentModeHash) {
if (!useAppServer && wasCreated && currentModeHash && message.hash !== currentModeHash) {
logger.debug('[Codex] Mode changed – restarting Codex session');
messageBuffer.addMessage('═'.repeat(40), 'status');
messageBuffer.addMessage('Starting new Codex session (mode changed)...', 'status');
try {
const prevSessionId = client.getSessionId();
nextExperimentalResume = findCodexResumeFile(prevSessionId);
if (nextExperimentalResume) {
logger.debug(`[Codex] Found resume file for session ${prevSessionId}: ${nextExperimentalResume}`);
messageBuffer.addMessage('Resuming previous context…', 'status');
} else {
logger.debug('[Codex] No resume file found for previous session');
}
} catch (error) {
logger.debug('[Codex] Error while searching resume file', error);
}
client.clearSession();
mcpClient?.clearSession();
wasCreated = false;
currentModeHash = null;
pending = message;
@@ -588,81 +536,149 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
try {
if (!wasCreated) {
let resumeFile: string | null = null;
if (nextExperimentalResume) {
resumeFile = nextExperimentalResume;
nextExperimentalResume = null;
logger.debug('[Codex] Using resume file from mode change:', resumeFile);
} else if (this.storedSessionIdForResume) {
const abortResumeFile = findCodexResumeFile(this.storedSessionIdForResume);
if (abortResumeFile) {
resumeFile = abortResumeFile;
logger.debug('[Codex] Using resume file from aborted session:', resumeFile);
messageBuffer.addMessage('Resuming from aborted session...', 'status');
if (useAppServer && appServerClient) {
const threadParams = buildThreadStartParams({
mode: message.mode,
mcpServers,
cliOverrides: session.codexCliOverrides
});
const resumeCandidate = session.sessionId;
let threadId: string | null = null;
if (resumeCandidate) {
try {
const resumeResponse = await appServerClient.resumeThread({
threadId: resumeCandidate,
...threadParams
}, {
signal: this.abortController.signal
});
const resumeRecord = asRecord(resumeResponse);
const resumeThread = resumeRecord ? asRecord(resumeRecord.thread) : null;
threadId = asString(resumeThread?.id) ?? resumeCandidate;
logger.debug(`[Codex] Resumed app-server thread ${threadId}`);
} catch (error) {
logger.warn(`[Codex] Failed to resume app-server thread ${resumeCandidate}, starting new thread`, error);
}
}
this.storedSessionIdForResume = null;
} else if (first && session.sessionId) {
const localResumeFile = findCodexResumeFile(session.sessionId);
if (localResumeFile) {
resumeFile = localResumeFile;
logger.debug('[Codex] Using resume file from local session:', localResumeFile);
messageBuffer.addMessage('Resuming from local session log...', 'status');
if (!threadId) {
const threadResponse = await appServerClient.startThread(threadParams, {
signal: this.abortController.signal
});
const threadRecord = asRecord(threadResponse);
const thread = threadRecord ? asRecord(threadRecord.thread) : null;
threadId = asString(thread?.id);
if (!threadId) {
throw new Error('app-server thread/start did not return thread.id');
}
}
if (!threadId) {
throw new Error('app-server resume did not return thread.id');
}
this.currentThreadId = threadId;
session.onSessionFound(threadId);
const turnParams = buildTurnStartParams({
threadId,
message: message.message,
mode: message.mode,
cliOverrides: session.codexCliOverrides
});
turnInFlight = true;
const turnResponse = await appServerClient.startTurn(turnParams, {
signal: this.abortController.signal
});
const turnRecord = asRecord(turnResponse);
const turn = turnRecord ? asRecord(turnRecord.turn) : null;
const turnId = asString(turn?.id);
if (turnId) {
this.currentTurnId = turnId;
}
} else if (mcpClient) {
const startConfig: CodexSessionConfig = buildCodexStartConfig({
message: message.message,
mode: message.mode,
first,
mcpServers,
cliOverrides: session.codexCliOverrides
});
await mcpClient.startSession(startConfig, { signal: this.abortController.signal });
syncSessionId();
}
const developerInstructions = resumeFile
? buildResumeInstructionsFromFile(resumeFile)
: undefined;
const startConfig: CodexSessionConfig = buildCodexStartConfig({
message: message.message,
mode: message.mode,
first,
mcpServers,
cliOverrides: session.codexCliOverrides,
developerInstructions
});
if (resumeFile) {
(startConfig.config as any).experimental_resume = resumeFile;
}
await client.startSession(startConfig, { signal: this.abortController.signal });
wasCreated = true;
first = false;
syncSessionId();
} else {
await client.continueSession(message.message, { signal: this.abortController.signal });
} else if (useAppServer && appServerClient) {
if (!this.currentThreadId) {
logger.debug('[Codex] Missing thread id; restarting app-server thread');
wasCreated = false;
pending = message;
continue;
}
const turnParams = buildTurnStartParams({
threadId: this.currentThreadId,
message: message.message,
mode: message.mode,
cliOverrides: session.codexCliOverrides
});
turnInFlight = true;
const turnResponse = await appServerClient.startTurn(turnParams, {
signal: this.abortController.signal
});
const turnRecord = asRecord(turnResponse);
const turn = turnRecord ? asRecord(turnRecord.turn) : null;
const turnId = asString(turn?.id);
if (turnId) {
this.currentTurnId = turnId;
}
} else if (mcpClient) {
await mcpClient.continueSession(message.message, { signal: this.abortController.signal });
syncSessionId();
}
} catch (error) {
logger.warn('Error in codex session:', error);
const isAbortError = error instanceof Error && error.name === 'AbortError';
if (useAppServer) {
turnInFlight = false;
}
if (isAbortError) {
messageBuffer.addMessage('Aborted by user', 'status');
session.sendSessionEvent({ type: 'message', message: 'Aborted by user' });
wasCreated = false;
currentModeHash = null;
logger.debug('[Codex] Marked session as not created after abort for proper resume');
if (!useAppServer) {
wasCreated = false;
currentModeHash = null;
logger.debug('[Codex] Marked session as not created after abort for proper resume');
}
} else {
messageBuffer.addMessage('Process exited unexpectedly', 'status');
session.sendSessionEvent({ type: 'message', message: 'Process exited unexpectedly' });
if (client.hasActiveSession()) {
this.storedSessionIdForResume = client.storeSessionForResume();
logger.debug('[Codex] Stored session after unexpected error:', this.storedSessionIdForResume);
if (useAppServer) {
this.currentTurnId = null;
this.currentThreadId = null;
wasCreated = false;
}
}
} finally {
permissionHandler.reset();
reasoningProcessor.abort();
diffProcessor.reset();
appServerEventConverter?.reset();
session.onThinkingChange(false);
emitReadyIfIdle({
pending,
queueSize: () => session.queue.size(),
shouldExit: this.shouldExit,
sendReady
});
if (!useAppServer || !turnInFlight) {
emitReadyIfIdle({
pending,
queueSize: () => session.queue.size(),
shouldExit: this.shouldExit,
sendReady
});
}
logActiveHandles('after-turn');
}
}
@@ -671,7 +687,12 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
protected async cleanup(): Promise<void> {
logger.debug('[codex-remote]: cleanup start');
try {
await this.client.disconnect();
if (this.appServerClient) {
await this.appServerClient.disconnect();
}
if (this.mcpClient) {
await this.mcpClient.disconnect();
}
} catch (error) {
logger.debug('[codex-remote]: Error disconnecting client', error);
}
cli/src/codex/utils/appServerConfig.test.ts
+86
View File
@@ -0,0 +1,86 @@
import { describe, expect, it } from 'vitest';
import { buildThreadStartParams, buildTurnStartParams } from './appServerConfig';
import { codexSystemPrompt } from './systemPrompt';
describe('appServerConfig', () => {
const mcpServers = { hapi: { command: 'node', args: ['mcp'] } };
it('applies CLI overrides when permission mode is default', () => {
const params = buildThreadStartParams({
mode: { permissionMode: 'default' },
mcpServers,
cliOverrides: { sandbox: 'danger-full-access', approvalPolicy: 'never' }
});
expect(params.sandbox).toBe('danger-full-access');
expect(params.approvalPolicy).toBe('never');
expect(params.baseInstructions).toBe(codexSystemPrompt);
expect(params.config).toEqual({
'mcp_servers.hapi': {
command: 'node',
args: ['mcp']
}
});
});
it('ignores CLI overrides when permission mode is not default', () => {
const params = buildThreadStartParams({
mode: { permissionMode: 'yolo' },
mcpServers,
cliOverrides: { sandbox: 'read-only', approvalPolicy: 'never' }
});
expect(params.sandbox).toBe('danger-full-access');
expect(params.approvalPolicy).toBe('on-failure');
});
it('builds turn params with mode defaults', () => {
const params = buildTurnStartParams({
threadId: 'thread-1',
message: 'hello',
mode: { permissionMode: 'read-only', model: 'o3' }
});
expect(params.threadId).toBe('thread-1');
expect(params.input).toEqual([{ type: 'text', text: 'hello' }]);
expect(params.approvalPolicy).toBe('never');
expect(params.sandboxPolicy).toEqual({ type: 'readOnly' });
expect(params.model).toBe('o3');
});
it('applies CLI overrides for turns when permission mode is default', () => {
const params = buildTurnStartParams({
threadId: 'thread-1',
message: 'hello',
mode: { permissionMode: 'default' },
cliOverrides: { sandbox: 'danger-full-access', approvalPolicy: 'never' }
});
expect(params.approvalPolicy).toBe('never');
expect(params.sandboxPolicy).toEqual({ type: 'dangerFullAccess' });
});
it('ignores CLI overrides for turns when permission mode is not default', () => {
const params = buildTurnStartParams({
threadId: 'thread-1',
message: 'hello',
mode: { permissionMode: 'safe-yolo' },
cliOverrides: { sandbox: 'read-only', approvalPolicy: 'never' }
});
expect(params.approvalPolicy).toBe('on-failure');
expect(params.sandboxPolicy).toEqual({ type: 'workspaceWrite' });
});
it('prefers turn overrides', () => {
const params = buildTurnStartParams({
threadId: 'thread-1',
message: 'hello',
mode: { permissionMode: 'default' },
overrides: { approvalPolicy: 'on-request', model: 'gpt-5' }
});
expect(params.approvalPolicy).toBe('on-request');
expect(params.model).toBe('gpt-5');
});
});
cli/src/codex/utils/appServerConfig.ts
+145
View File
@@ -0,0 +1,145 @@
import type { EnhancedMode } from '../loop';
import type { CodexCliOverrides } from './codexCliOverrides';
import type { McpServersConfig } from './buildHapiMcpBridge';
import { codexSystemPrompt } from './systemPrompt';
import type {
ApprovalPolicy,
SandboxMode,
SandboxPolicy,
ThreadStartParams,
TurnStartParams
} from '../appServerTypes';
function resolveApprovalPolicy(mode: EnhancedMode): ApprovalPolicy {
switch (mode.permissionMode) {
case 'default': return 'untrusted';
case 'read-only': return 'never';
case 'safe-yolo': return 'on-failure';
case 'yolo': return 'on-failure';
default: {
throw new Error(`Unknown permission mode: ${mode.permissionMode}`);
}
}
}
function resolveSandbox(mode: EnhancedMode): SandboxMode {
switch (mode.permissionMode) {
case 'default': return 'workspace-write';
case 'read-only': return 'read-only';
case 'safe-yolo': return 'workspace-write';
case 'yolo': return 'danger-full-access';
default: {
throw new Error(`Unknown permission mode: ${mode.permissionMode}`);
}
}
}
function resolveSandboxPolicy(mode: EnhancedMode): SandboxPolicy {
switch (mode.permissionMode) {
case 'default': return { type: 'workspaceWrite' };
case 'read-only': return { type: 'readOnly' };
case 'safe-yolo': return { type: 'workspaceWrite' };
case 'yolo': return { type: 'dangerFullAccess' };
default: {
throw new Error(`Unknown permission mode: ${mode.permissionMode}`);
}
}
}
function resolveSandboxPolicyOverride(value: CodexCliOverrides['sandbox'] | undefined): SandboxPolicy | undefined {
switch (value) {
case 'read-only':
return { type: 'readOnly' };
case 'workspace-write':
return { type: 'workspaceWrite' };
case 'danger-full-access':
return { type: 'dangerFullAccess' };
default:
return undefined;
}
}
function buildMcpServerConfig(mcpServers: McpServersConfig): Record<string, unknown> {
const config: Record<string, unknown> = {};
for (const [name, server] of Object.entries(mcpServers)) {
config[`mcp_servers.${name}`] = {
command: server.command,
args: server.args
};
}
return config;
}
export function buildThreadStartParams(args: {
mode: EnhancedMode;
mcpServers: McpServersConfig;
cliOverrides?: CodexCliOverrides;
baseInstructions?: string;
developerInstructions?: string;
}): ThreadStartParams {
const approvalPolicy = resolveApprovalPolicy(args.mode);
const sandbox = resolveSandbox(args.mode);
const allowCliOverrides = args.mode.permissionMode === 'default';
const cliOverrides = allowCliOverrides ? args.cliOverrides : undefined;
const resolvedApprovalPolicy = cliOverrides?.approvalPolicy ?? approvalPolicy;
const resolvedSandbox = cliOverrides?.sandbox ?? sandbox;
const config = buildMcpServerConfig(args.mcpServers);
const baseInstructions = args.baseInstructions ?? codexSystemPrompt;
const params: ThreadStartParams = {
approvalPolicy: resolvedApprovalPolicy,
sandbox: resolvedSandbox,
baseInstructions,
...(args.developerInstructions ? { developerInstructions: args.developerInstructions } : {}),
...(Object.keys(config).length > 0 ? { config } : {})
};
if (args.mode.model) {
params.model = args.mode.model;
}
return params;
}
export function buildTurnStartParams(args: {
threadId: string;
message: string;
mode?: EnhancedMode;
cliOverrides?: CodexCliOverrides;
overrides?: {
approvalPolicy?: TurnStartParams['approvalPolicy'];
sandboxPolicy?: TurnStartParams['sandboxPolicy'];
model?: string;
};
}): TurnStartParams {
const params: TurnStartParams = {
threadId: args.threadId,
input: [{ type: 'text', text: args.message }]
};
const allowCliOverrides = args.mode?.permissionMode === 'default';
const cliOverrides = allowCliOverrides ? args.cliOverrides : undefined;
const approvalPolicy = args.overrides?.approvalPolicy
?? cliOverrides?.approvalPolicy
?? (args.mode ? resolveApprovalPolicy(args.mode) : undefined);
if (approvalPolicy) {
params.approvalPolicy = approvalPolicy;
}
const sandboxPolicy = args.overrides?.sandboxPolicy
?? resolveSandboxPolicyOverride(cliOverrides?.sandbox)
?? (args.mode ? resolveSandboxPolicy(args.mode) : undefined);
if (sandboxPolicy) {
params.sandboxPolicy = sandboxPolicy;
}
const model = args.overrides?.model ?? args.mode?.model;
if (model) {
params.model = model;
}
return params;
}
cli/src/codex/utils/appServerEventConverter.test.ts
+85
View File
@@ -0,0 +1,85 @@
import { describe, expect, it } from 'vitest';
import { AppServerEventConverter } from './appServerEventConverter';
describe('AppServerEventConverter', () => {
it('maps thread/started', () => {
const converter = new AppServerEventConverter();
const events = converter.handleNotification('thread/started', { thread: { id: 'thread-1' } });
expect(events).toEqual([{ type: 'thread_started', thread_id: 'thread-1' }]);
});
it('maps thread/resumed', () => {
const converter = new AppServerEventConverter();
const events = converter.handleNotification('thread/resumed', { thread: { id: 'thread-2' } });
expect(events).toEqual([{ type: 'thread_started', thread_id: 'thread-2' }]);
});
it('maps turn/started and completed statuses', () => {
const converter = new AppServerEventConverter();
const started = converter.handleNotification('turn/started', { turn: { id: 'turn-1' } });
expect(started).toEqual([{ type: 'task_started', turn_id: 'turn-1' }]);
const completed = converter.handleNotification('turn/completed', { turn: { id: 'turn-1' }, status: 'Completed' });
expect(completed).toEqual([{ type: 'task_complete', turn_id: 'turn-1' }]);
const interrupted = converter.handleNotification('turn/completed', { turn: { id: 'turn-1' }, status: 'Interrupted' });
expect(interrupted).toEqual([{ type: 'turn_aborted', turn_id: 'turn-1' }]);
const failed = converter.handleNotification('turn/completed', { turn: { id: 'turn-1' }, status: 'Failed', message: 'boom' });
expect(failed).toEqual([{ type: 'task_failed', turn_id: 'turn-1', error: 'boom' }]);
});
it('accumulates agent message deltas', () => {
const converter = new AppServerEventConverter();
converter.handleNotification('item/agentMessage/delta', { itemId: 'msg-1', delta: 'Hello' });
converter.handleNotification('item/agentMessage/delta', { itemId: 'msg-1', delta: ' world' });
const completed = converter.handleNotification('item/completed', {
item: { id: 'msg-1', type: 'agentMessage' }
});
expect(completed).toEqual([{ type: 'agent_message', message: 'Hello world' }]);
});
it('maps command execution items and output deltas', () => {
const converter = new AppServerEventConverter();
const started = converter.handleNotification('item/started', {
item: { id: 'cmd-1', type: 'commandExecution', command: 'ls' }
});
expect(started).toEqual([{
type: 'exec_command_begin',
call_id: 'cmd-1',
command: 'ls'
}]);
converter.handleNotification('item/commandExecution/outputDelta', { itemId: 'cmd-1', delta: 'ok' });
const completed = converter.handleNotification('item/completed', {
item: { id: 'cmd-1', type: 'commandExecution', exitCode: 0 }
});
expect(completed).toEqual([{
type: 'exec_command_end',
call_id: 'cmd-1',
output: 'ok',
exit_code: 0
}]);
});
it('maps reasoning deltas', () => {
const converter = new AppServerEventConverter();
const events = converter.handleNotification('item/reasoning/textDelta', { itemId: 'r1', delta: 'step' });
expect(events).toEqual([{ type: 'agent_reasoning_delta', delta: 'step' }]);
});
it('maps diff updates', () => {
const converter = new AppServerEventConverter();
const events = converter.handleNotification('turn/diff/updated', { diff: 'diff --git a b' });
expect(events).toEqual([{ type: 'turn_diff', unified_diff: 'diff --git a b' }]);
});
});
cli/src/codex/utils/appServerEventConverter.ts
+313
View File
@@ -0,0 +1,313 @@
import { logger } from '@/ui/logger';
type ConvertedEvent = {
type: string;
[key: string]: unknown;
};
function asRecord(value: unknown): Record<string, unknown> | null {
if (!value || typeof value !== 'object') {
return null;
}
return value as Record<string, unknown>;
}
function asString(value: unknown): string | null {
return typeof value === 'string' && value.length > 0 ? value : null;
}
function asBoolean(value: unknown): boolean | null {
return typeof value === 'boolean' ? value : null;
}
function asNumber(value: unknown): number | null {
return typeof value === 'number' && Number.isFinite(value) ? value : null;
}
function extractItemId(params: Record<string, unknown>): string | null {
const direct = asString(params.itemId ?? params.item_id ?? params.id);
if (direct) return direct;
const item = asRecord(params.item);
if (item) {
return asString(item.id ?? item.itemId ?? item.item_id);
}
return null;
}
function extractItem(params: Record<string, unknown>): Record<string, unknown> | null {
const item = asRecord(params.item);
return item ?? params;
}
function normalizeItemType(value: unknown): string | null {
const raw = asString(value);
if (!raw) return null;
return raw.toLowerCase().replace(/[\s_-]/g, '');
}
function extractCommand(value: unknown): string | null {
if (typeof value === 'string') return value;
if (Array.isArray(value)) {
const parts = value.filter((part): part is string => typeof part === 'string');
return parts.length > 0 ? parts.join(' ') : null;
}
return null;
}
function extractChanges(value: unknown): Record<string, unknown> | null {
const record = asRecord(value);
if (record) return record;
if (Array.isArray(value)) {
const changes: Record<string, unknown> = {};
for (const entry of value) {
const entryRecord = asRecord(entry);
if (!entryRecord) continue;
const path = asString(entryRecord.path ?? entryRecord.file ?? entryRecord.filePath ?? entryRecord.file_path);
if (path) {
changes[path] = entryRecord;
}
}
return Object.keys(changes).length > 0 ? changes : null;
}
return null;
}
export class AppServerEventConverter {
private readonly agentMessageBuffers = new Map<string, string>();
private readonly reasoningBuffers = new Map<string, string>();
private readonly commandOutputBuffers = new Map<string, string>();
private readonly commandMeta = new Map<string, Record<string, unknown>>();
private readonly fileChangeMeta = new Map<string, Record<string, unknown>>();
handleNotification(method: string, params: unknown): ConvertedEvent[] {
const events: ConvertedEvent[] = [];
const paramsRecord = asRecord(params) ?? {};
if (method === 'thread/started' || method === 'thread/resumed') {
const thread = asRecord(paramsRecord.thread) ?? paramsRecord;
const threadId = asString(thread.threadId ?? thread.thread_id ?? thread.id);
if (threadId) {
events.push({ type: 'thread_started', thread_id: threadId });
}
return events;
}
if (method === 'turn/started') {
const turn = asRecord(paramsRecord.turn) ?? paramsRecord;
const turnId = asString(turn.turnId ?? turn.turn_id ?? turn.id);
events.push({ type: 'task_started', ...(turnId ? { turn_id: turnId } : {}) });
return events;
}
if (method === 'turn/completed') {
const turn = asRecord(paramsRecord.turn) ?? paramsRecord;
const statusRaw = asString(paramsRecord.status ?? turn.status);
const status = statusRaw?.toLowerCase();
const turnId = asString(turn.turnId ?? turn.turn_id ?? turn.id);
const errorMessage = asString(paramsRecord.error ?? paramsRecord.message ?? paramsRecord.reason);
if (status === 'interrupted' || status === 'cancelled' || status === 'canceled') {
events.push({ type: 'turn_aborted', ...(turnId ? { turn_id: turnId } : {}) });
return events;
}
if (status === 'failed' || status === 'error') {
events.push({ type: 'task_failed', ...(turnId ? { turn_id: turnId } : {}), ...(errorMessage ? { error: errorMessage } : {}) });
return events;
}
events.push({ type: 'task_complete', ...(turnId ? { turn_id: turnId } : {}) });
return events;
}
if (method === 'turn/diff/updated') {
const diff = asString(paramsRecord.diff ?? paramsRecord.unified_diff ?? paramsRecord.unifiedDiff);
if (diff) {
events.push({ type: 'turn_diff', unified_diff: diff });
}
return events;
}
if (method === 'thread/tokenUsage/updated') {
const info = asRecord(paramsRecord.tokenUsage ?? paramsRecord.token_usage ?? paramsRecord) ?? {};
events.push({ type: 'token_count', info });
return events;
}
if (method === 'error') {
const willRetry = asBoolean(paramsRecord.will_retry ?? paramsRecord.willRetry) ?? false;
if (willRetry) return events;
const message = asString(paramsRecord.message) ?? asString(asRecord(paramsRecord.error)?.message);
if (message) {
events.push({ type: 'task_failed', error: message });
}
return events;
}
if (method === 'item/agentMessage/delta') {
const itemId = extractItemId(paramsRecord);
const delta = asString(paramsRecord.delta ?? paramsRecord.text ?? paramsRecord.message);
if (itemId && delta) {
const prev = this.agentMessageBuffers.get(itemId) ?? '';
this.agentMessageBuffers.set(itemId, prev + delta);
}
return events;
}
if (method === 'item/reasoning/textDelta') {
const itemId = extractItemId(paramsRecord) ?? 'reasoning';
const delta = asString(paramsRecord.delta ?? paramsRecord.text ?? paramsRecord.message);
if (delta) {
const prev = this.reasoningBuffers.get(itemId) ?? '';
this.reasoningBuffers.set(itemId, prev + delta);
events.push({ type: 'agent_reasoning_delta', delta });
}
return events;
}
if (method === 'item/reasoning/summaryPartAdded') {
events.push({ type: 'agent_reasoning_section_break' });
return events;
}
if (method === 'item/commandExecution/outputDelta') {
const itemId = extractItemId(paramsRecord);
const delta = asString(paramsRecord.delta ?? paramsRecord.text ?? paramsRecord.output ?? paramsRecord.stdout);
if (itemId && delta) {
const prev = this.commandOutputBuffers.get(itemId) ?? '';
this.commandOutputBuffers.set(itemId, prev + delta);
}
return events;
}
if (method === 'item/started' || method === 'item/completed') {
const item = extractItem(paramsRecord);
if (!item) return events;
const itemType = normalizeItemType(item.type ?? item.itemType ?? item.kind);
const itemId = extractItemId(paramsRecord) ?? asString(item.id ?? item.itemId ?? item.item_id);
if (!itemType || !itemId) {
return events;
}
if (itemType === 'agentmessage') {
if (method === 'item/completed') {
const text = asString(item.text ?? item.message ?? item.content) ?? this.agentMessageBuffers.get(itemId);
if (text) {
events.push({ type: 'agent_message', message: text });
}
this.agentMessageBuffers.delete(itemId);
}
return events;
}
if (itemType === 'reasoning') {
if (method === 'item/completed') {
const text = asString(item.text ?? item.message ?? item.content) ?? this.reasoningBuffers.get(itemId);
if (text) {
events.push({ type: 'agent_reasoning', text });
}
this.reasoningBuffers.delete(itemId);
}
return events;
}
if (itemType === 'commandexecution') {
if (method === 'item/started') {
const command = extractCommand(item.command ?? item.cmd ?? item.args);
const cwd = asString(item.cwd ?? item.workingDirectory ?? item.working_directory);
const autoApproved = asBoolean(item.autoApproved ?? item.auto_approved);
const meta: Record<string, unknown> = {};
if (command) meta.command = command;
if (cwd) meta.cwd = cwd;
if (autoApproved !== null) meta.auto_approved = autoApproved;
this.commandMeta.set(itemId, meta);
events.push({
type: 'exec_command_begin',
call_id: itemId,
...meta
});
}
if (method === 'item/completed') {
const meta = this.commandMeta.get(itemId) ?? {};
const output = asString(item.output ?? item.result ?? item.stdout) ?? this.commandOutputBuffers.get(itemId);
const stderr = asString(item.stderr);
const error = asString(item.error);
const exitCode = asNumber(item.exitCode ?? item.exit_code ?? item.exitcode);
const status = asString(item.status);
events.push({
type: 'exec_command_end',
call_id: itemId,
...meta,
...(output ? { output } : {}),
...(stderr ? { stderr } : {}),
...(error ? { error } : {}),
...(exitCode !== null ? { exit_code: exitCode } : {}),
...(status ? { status } : {})
});
this.commandMeta.delete(itemId);
this.commandOutputBuffers.delete(itemId);
}
return events;
}
if (itemType === 'filechange') {
if (method === 'item/started') {
const changes = extractChanges(item.changes ?? item.change ?? item.diff);
const autoApproved = asBoolean(item.autoApproved ?? item.auto_approved);
const meta: Record<string, unknown> = {};
if (changes) meta.changes = changes;
if (autoApproved !== null) meta.auto_approved = autoApproved;
this.fileChangeMeta.set(itemId, meta);
events.push({
type: 'patch_apply_begin',
call_id: itemId,
...meta
});
}
if (method === 'item/completed') {
const meta = this.fileChangeMeta.get(itemId) ?? {};
const stdout = asString(item.stdout ?? item.output);
const stderr = asString(item.stderr);
const success = asBoolean(item.success ?? item.ok ?? item.applied ?? item.status === 'completed');
events.push({
type: 'patch_apply_end',
call_id: itemId,
...meta,
...(stdout ? { stdout } : {}),
...(stderr ? { stderr } : {}),
success: success ?? false
});
this.fileChangeMeta.delete(itemId);
}
return events;
}
}
logger.debug('[AppServerEventConverter] Unhandled notification', { method, params });
return events;
}
reset(): void {
this.agentMessageBuffers.clear();
this.reasoningBuffers.clear();
this.commandOutputBuffers.clear();
this.commandMeta.clear();
this.fileChangeMeta.clear();
}
}
cli/src/codex/utils/appServerPermissionAdapter.ts
+94
View File
@@ -0,0 +1,94 @@
import { randomUUID } from 'node:crypto';
import { logger } from '@/ui/logger';
import type { CodexPermissionHandler } from './permissionHandler';
import type { CodexAppServerClient } from '../codexAppServerClient';
type PermissionDecision = 'approved' | 'approved_for_session' | 'denied' | 'abort';
type PermissionResult = {
decision: PermissionDecision;
reason?: string;
};
function asRecord(value: unknown): Record<string, unknown> | null {
if (!value || typeof value !== 'object') {
return null;
}
return value as Record<string, unknown>;
}
function asString(value: unknown): string | undefined {
return typeof value === 'string' && value.length > 0 ? value : undefined;
}
function mapDecision(decision: PermissionDecision): { decision: string } {
switch (decision) {
case 'approved':
return { decision: 'accept' };
case 'approved_for_session':
return { decision: 'acceptForSession' };
case 'denied':
return { decision: 'decline' };
case 'abort':
return { decision: 'cancel' };
}
}
export function registerAppServerPermissionHandlers(args: {
client: CodexAppServerClient;
permissionHandler: CodexPermissionHandler;
onUserInputRequest?: (request: unknown) => Promise<Record<string, string[]>>;
}): void {
const { client, permissionHandler, onUserInputRequest } = args;
client.registerRequestHandler('item/commandExecution/requestApproval', async (params) => {
const record = asRecord(params) ?? {};
const toolCallId = asString(record.itemId) ?? randomUUID();
const reason = asString(record.reason);
const command = record.command;
const cwd = asString(record.cwd);
const result = await permissionHandler.handleToolCall(
toolCallId,
'CodexBash',
{
message: reason,
command,
cwd
}
) as PermissionResult;
return mapDecision(result.decision);
});
client.registerRequestHandler('item/fileChange/requestApproval', async (params) => {
const record = asRecord(params) ?? {};
const toolCallId = asString(record.itemId) ?? randomUUID();
const reason = asString(record.reason);
const grantRoot = asString(record.grantRoot);
const result = await permissionHandler.handleToolCall(
toolCallId,
'CodexPatch',
{
message: reason,
grantRoot
}
) as PermissionResult;
return mapDecision(result.decision);
});
client.registerRequestHandler('item/tool/requestUserInput', async (params) => {
if (!onUserInputRequest) {
logger.debug('[CodexAppServer] No user-input handler registered; cancelling request');
return { decision: 'cancel' };
}
const answers = await onUserInputRequest(params);
return {
decision: 'accept',
answers
};
});
}