From 4290f539bdf110b5403432ffaea4fd34b4ebb8ff Mon Sep 17 00:00:00 2001 From: weishu Date: Mon, 22 Dec 2025 19:13:03 +0800 Subject: [PATCH] refactor(codex): extract session management and launcher logic into separate modules Reorganized runCodex.ts to improve maintainability by extracting: - CodexSession class for session lifecycle management - CodexLocalLauncher and CodexRemoteLauncher for mode-specific initialization - CodexEventConverter for MCP message handling and UI buffer updates - CodexSessionScanner for resume file discovery - emitReadyIfIdle utility for ready event emission Added codexSessionId field to metadata schema for session tracking. Updated UI components to work with refactored architecture. --- cli/src/api/types.ts | 2 + cli/src/codex/codexLocal.ts | 95 +++ cli/src/codex/codexLocalLauncher.ts | 107 +++ cli/src/codex/codexRemoteLauncher.ts | 484 ++++++++++++ cli/src/codex/loop.ts | 67 ++ cli/src/codex/runCodex.ts | 735 +++--------------- cli/src/codex/session.ts | 78 ++ .../codex/utils/codexEventConverter.test.ts | 85 ++ cli/src/codex/utils/codexEventConverter.ts | 224 ++++++ .../codex/utils/codexSessionScanner.test.ts | 76 ++ cli/src/codex/utils/codexSessionScanner.ts | 219 ++++++ cli/src/codex/utils/emitReadyIfIdle.ts | 27 + cli/src/ui/ink/CodexDisplay.tsx | 55 +- cli/src/ui/ink/RemoteModeDisplay.tsx | 6 +- cli/src/ui/ink/messageBuffer.ts | 7 +- 15 files changed, 1606 insertions(+), 661 deletions(-) create mode 100644 cli/src/codex/codexLocal.ts create mode 100644 cli/src/codex/codexLocalLauncher.ts create mode 100644 cli/src/codex/codexRemoteLauncher.ts create mode 100644 cli/src/codex/loop.ts create mode 100644 cli/src/codex/session.ts create mode 100644 cli/src/codex/utils/codexEventConverter.test.ts create mode 100644 cli/src/codex/utils/codexEventConverter.ts create mode 100644 cli/src/codex/utils/codexSessionScanner.test.ts create mode 100644 cli/src/codex/utils/codexSessionScanner.ts create mode 100644 cli/src/codex/utils/emitReadyIfIdle.ts diff --git a/cli/src/api/types.ts b/cli/src/api/types.ts index c098f293..b884ee1c 100644 --- a/cli/src/api/types.ts +++ b/cli/src/api/types.ts @@ -16,6 +16,7 @@ export type Metadata = { } machineId?: string claudeSessionId?: string + codexSessionId?: string tools?: string[] slashCommands?: string[] homeDir: string @@ -44,6 +45,7 @@ export const MetadataSchema = z.object({ }).optional(), machineId: z.string().optional(), claudeSessionId: z.string().optional(), + codexSessionId: z.string().optional(), tools: z.array(z.string()).optional(), slashCommands: z.array(z.string()).optional(), homeDir: z.string(), diff --git a/cli/src/codex/codexLocal.ts b/cli/src/codex/codexLocal.ts new file mode 100644 index 00000000..82b38725 --- /dev/null +++ b/cli/src/codex/codexLocal.ts @@ -0,0 +1,95 @@ +import { spawn } from 'node:child_process'; +import { logger } from '@/ui/logger'; + +export async function codexLocal(opts: { + abort: AbortSignal; + sessionId: string | null; + path: string; + model?: string; + sandbox?: 'read-only' | 'workspace-write' | 'danger-full-access'; + onSessionFound: (id: string) => void; +}): Promise { + const args: string[] = []; + + if (opts.sessionId) { + args.push('resume', opts.sessionId); + opts.onSessionFound(opts.sessionId); + } + + if (opts.model) { + args.push('--model', opts.model); + } + + if (opts.sandbox) { + args.push('--sandbox', opts.sandbox); + } + + logger.debug(`[CodexLocal] Spawning codex with args: ${JSON.stringify(args)}`); + + process.stdin.pause(); + try { + await new Promise((resolve, reject) => { + const child = spawn('codex', args, { + stdio: ['inherit', 'inherit', 'inherit'], + signal: opts.abort, + cwd: opts.path, + env: process.env + }); + + let abortKillTimeout: NodeJS.Timeout | null = null; + const abortHandler = () => { + if (abortKillTimeout) { + return; + } + abortKillTimeout = setTimeout(() => { + if (child.exitCode === null && !child.killed) { + logger.debug('[CodexLocal] Abort timeout reached, sending SIGKILL'); + try { + child.kill('SIGKILL'); + } catch (error) { + logger.debug('[CodexLocal] Failed to send SIGKILL:', error); + } + } + }, 5000); + }; + + if (opts.abort.aborted) { + abortHandler(); + } else { + opts.abort.addEventListener('abort', abortHandler); + } + + const cleanupAbortHandler = () => { + if (abortKillTimeout) { + clearTimeout(abortKillTimeout); + abortKillTimeout = null; + } + opts.abort.removeEventListener('abort', abortHandler); + }; + + child.on('error', (error) => { + cleanupAbortHandler(); + reject(error); + }); + + child.on('exit', (code, signal) => { + cleanupAbortHandler(); + if (signal === 'SIGTERM' && opts.abort.aborted) { + resolve(); + return; + } + if (signal) { + reject(new Error(`Process terminated with signal: ${signal}`)); + return; + } + if (typeof code === 'number' && code !== 0) { + reject(new Error(`Process exited with code: ${code}`)); + return; + } + resolve(); + }); + }); + } finally { + process.stdin.resume(); + } +} diff --git a/cli/src/codex/codexLocalLauncher.ts b/cli/src/codex/codexLocalLauncher.ts new file mode 100644 index 00000000..53d6cb91 --- /dev/null +++ b/cli/src/codex/codexLocalLauncher.ts @@ -0,0 +1,107 @@ +import { logger } from '@/ui/logger'; +import { codexLocal } from './codexLocal'; +import { CodexSession } from './session'; +import { Future } from '@/utils/future'; +import { createCodexSessionScanner } from './utils/codexSessionScanner'; +import { convertCodexEvent } from './utils/codexEventConverter'; + +export async function codexLocalLauncher(session: CodexSession): Promise<'switch' | 'exit'> { + const scanner = await createCodexSessionScanner({ + sessionId: session.sessionId, + onSessionFound: (sessionId) => { + session.onSessionFound(sessionId); + }, + onEvent: (event) => { + const converted = convertCodexEvent(event); + if (converted?.sessionId) { + session.onSessionFound(converted.sessionId); + scanner.onNewSession(converted.sessionId); + } + if (converted?.message) { + session.sendCodexMessage(converted.message); + } + } + }); + + let exitReason: 'switch' | 'exit' | null = null; + const processAbortController = new AbortController(); + const exitFuture = new Future(); + + try { + async function abortProcess() { + if (!processAbortController.signal.aborted) { + processAbortController.abort(); + } + await exitFuture.promise; + } + + async function doAbort() { + logger.debug('[codex-local]: doAbort'); + if (!exitReason) { + exitReason = 'switch'; + } + session.queue.reset(); + await abortProcess(); + } + + async function doSwitch() { + logger.debug('[codex-local]: doSwitch'); + if (!exitReason) { + exitReason = 'switch'; + } + await abortProcess(); + } + + session.client.rpcHandlerManager.registerHandler('abort', doAbort); + session.client.rpcHandlerManager.registerHandler('switch', doSwitch); + session.queue.setOnMessage(() => { + void doSwitch(); + }); + + if (session.queue.size() > 0) { + return 'switch'; + } + + const handleSessionFound = (sessionId: string) => { + session.onSessionFound(sessionId); + scanner.onNewSession(sessionId); + }; + + while (true) { + if (exitReason) { + return exitReason; + } + + logger.debug('[codex-local]: launch'); + try { + await codexLocal({ + path: session.path, + sessionId: session.sessionId, + onSessionFound: handleSessionFound, + abort: processAbortController.signal + }); + + if (!exitReason) { + exitReason = 'exit'; + break; + } + } catch (error) { + logger.debug('[codex-local]: launch error', error); + const message = error instanceof Error ? error.message : String(error); + session.sendSessionEvent({ type: 'message', message: `Local Codex process failed: ${message}` }); + if (!exitReason) { + exitReason = 'switch'; + } + break; + } + } + } finally { + exitFuture.resolve(undefined); + session.client.rpcHandlerManager.registerHandler('abort', async () => {}); + session.client.rpcHandlerManager.registerHandler('switch', async () => {}); + session.queue.setOnMessage(null); + await scanner.cleanup(); + } + + return exitReason || 'exit'; +} diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts new file mode 100644 index 00000000..5c18ddd9 --- /dev/null +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -0,0 +1,484 @@ +import { render } from 'ink'; +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 { CodexPermissionHandler } from './utils/permissionHandler'; +import { ReasoningProcessor } from './utils/reasoningProcessor'; +import { DiffProcessor } from './utils/diffProcessor'; +import { logger } from '@/ui/logger'; +import { MessageBuffer } from '@/ui/ink/messageBuffer'; +import { CodexDisplay } from '@/ui/ink/CodexDisplay'; +import { trimIdent } from '@/utils/trimIdent'; +import type { CodexSessionConfig } from './types'; +import { getHappyCliCommand } from '@/utils/spawnHappyCLI'; +import { startHappyServer } from '@/claude/utils/startHappyServer'; +import { emitReadyIfIdle } from './utils/emitReadyIfIdle'; +import type { CodexSession } from './session'; +import type { EnhancedMode } from './loop'; + +export async function codexRemoteLauncher(session: CodexSession): Promise<'switch' | 'exit'> { + const hasTTY = process.stdout.isTTY && process.stdin.isTTY; + const messageBuffer = new MessageBuffer(); + let inkInstance: any = null; + + let exitReason: 'switch' | 'exit' | null = null; + let shouldExit = false; + + if (hasTTY) { + console.clear(); + inkInstance = render(React.createElement(CodexDisplay, { + messageBuffer, + logPath: process.env.DEBUG ? session.logPath : undefined, + onExit: async () => { + logger.debug('[codex-remote]: Exiting agent via Ctrl-C'); + exitReason = 'exit'; + shouldExit = true; + await handleAbort(); + }, + onSwitchToLocal: async () => { + logger.debug('[codex-remote]: Switching to local mode via double space'); + exitReason = 'switch'; + shouldExit = true; + await handleAbort(); + } + }), { + exitOnCtrlC: false, + patchConsole: false + }); + } + + if (hasTTY) { + process.stdin.resume(); + if (process.stdin.isTTY) { + process.stdin.setRawMode(true); + } + process.stdin.setEncoding('utf8'); + } + + const client = new CodexMcpClient(); + + 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 permissionHandler = new CodexPermissionHandler(session.client); + const reasoningProcessor = new ReasoningProcessor((message) => { + session.sendCodexMessage(message); + }); + const diffProcessor = new DiffProcessor((message) => { + session.sendCodexMessage(message); + }); + + client.setPermissionHandler(permissionHandler); + client.setHandler((msg) => { + logger.debug(`[Codex] MCP message: ${JSON.stringify(msg)}`); + + 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); + messageBuffer.addMessage( + `Result: ${truncatedOutput}${output.length > 200 ? '...' : ''}`, + 'result' + ); + } else if (msg.type === 'task_started') { + messageBuffer.addMessage('Starting task...', 'status'); + } else if (msg.type === 'task_complete') { + messageBuffer.addMessage('Task completed', 'status'); + sendReady(); + } else if (msg.type === 'turn_aborted') { + messageBuffer.addMessage('Turn aborted', 'status'); + sendReady(); + } + + if (msg.type === 'task_started') { + if (!session.thinking) { + logger.debug('thinking started'); + session.onThinkingChange(true); + } + } + if (msg.type === 'task_complete' || msg.type === 'turn_aborted') { + if (session.thinking) { + logger.debug('thinking completed'); + session.onThinkingChange(false); + } + diffProcessor.reset(); + } + if (msg.type === 'agent_reasoning_section_break') { + reasoningProcessor.handleSectionBreak(); + } + if (msg.type === 'agent_reasoning_delta') { + reasoningProcessor.processDelta(msg.delta); + } + if (msg.type === 'agent_reasoning') { + reasoningProcessor.complete(msg.text); + } + if (msg.type === 'agent_message') { + session.sendCodexMessage({ + type: 'message', + message: msg.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 (msg.type === 'exec_command_end') { + const { call_id, type, ...output } = msg; + session.sendCodexMessage({ + type: 'tool-call-result', + callId: call_id, + output: output, + id: randomUUID() + }); + } + if (msg.type === 'token_count') { + session.sendCodexMessage({ + ...msg, + id: randomUUID() + }); + } + if (msg.type === 'patch_apply_begin') { + const { call_id, auto_approved, changes } = msg; + + 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); + } + } + }); + + const happyServer = await startHappyServer(session.client); + const bridgeCommand = getHappyCliCommand(['mcp', '--url', happyServer.url]); + const mcpServers = { + hapi: { + command: bridgeCommand.command, + args: bridgeCommand.args + } + } as const; + + let abortController = new AbortController(); + let storedSessionIdForResume: string | null = null; + + async function handleAbort() { + logger.debug('[Codex] Abort requested - stopping current task'); + try { + if (client.hasActiveSession()) { + storedSessionIdForResume = client.storeSessionForResume(); + logger.debug('[Codex] Stored session for resume:', storedSessionIdForResume); + } + + abortController.abort(); + session.queue.reset(); + permissionHandler.reset(); + reasoningProcessor.abort(); + diffProcessor.reset(); + logger.debug('[Codex] Abort completed - session remains active'); + } catch (error) { + logger.debug('[Codex] Error during abort:', error); + } finally { + abortController = new AbortController(); + } + } + + session.client.rpcHandlerManager.registerHandler('abort', async () => { + await handleAbort(); + }); + + session.client.rpcHandlerManager.registerHandler('switch', async () => { + exitReason = 'switch'; + shouldExit = true; + await handleAbort(); + }); + + function logActiveHandles(tag: string) { + if (!process.env.DEBUG) return; + const anyProc: any = process as any; + const handles = typeof anyProc._getActiveHandles === 'function' ? anyProc._getActiveHandles() : []; + const requests = typeof anyProc._getActiveRequests === 'function' ? anyProc._getActiveRequests() : []; + logger.debug(`[codex][handles] ${tag}: handles=${handles.length} requests=${requests.length}`); + try { + const kinds = handles.map((h: any) => (h && h.constructor ? h.constructor.name : typeof h)); + logger.debug(`[codex][handles] kinds=${JSON.stringify(kinds)}`); + } catch {} + } + + const sendReady = () => { + session.sendSessionEvent({ type: 'ready' }); + }; + + const syncSessionId = () => { + const clientSessionId = client.getSessionId(); + if (clientSessionId && clientSessionId !== session.sessionId) { + session.onSessionFound(clientSessionId); + } + }; + + try { + await client.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; + + while (!shouldExit) { + logActiveHandles('loop-top'); + let message: { message: string; mode: EnhancedMode; isolate: boolean; hash: string } | null = pending; + pending = null; + if (!message) { + const waitSignal = abortController.signal; + const batch = await session.queue.waitForMessagesAndGetAsString(waitSignal); + if (!batch) { + if (waitSignal.aborted && !shouldExit) { + logger.debug('[codex]: Wait aborted while idle; ignoring and continuing'); + continue; + } + logger.debug(`[codex]: batch=${!!batch}, shouldExit=${shouldExit}`); + break; + } + message = batch; + } + + if (!message) { + break; + } + + if (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(); + wasCreated = false; + currentModeHash = null; + pending = message; + permissionHandler.reset(); + reasoningProcessor.abort(); + diffProcessor.reset(); + session.onThinkingChange(false); + continue; + } + + messageBuffer.addMessage(message.message, 'user'); + currentModeHash = message.hash; + + try { + const approvalPolicy = (() => { + switch (message.mode.permissionMode) { + case 'default': return 'untrusted' as const; + case 'read-only': return 'never' as const; + case 'safe-yolo': return 'on-failure' as const; + case 'yolo': return 'on-failure' as const; + } + })(); + const sandbox = (() => { + switch (message.mode.permissionMode) { + case 'default': return 'workspace-write' as const; + case 'read-only': return 'read-only' as const; + case 'safe-yolo': return 'workspace-write' as const; + case 'yolo': return 'danger-full-access' as const; + } + })(); + + if (!wasCreated) { + const startConfig: CodexSessionConfig = { + prompt: first ? message.message + '\n\n' + trimIdent(`Based on this message, call functions.hapi__change_title to change chat session title that would represent the current task. If chat idea would change dramatically - call this function again to update the title.`) : message.message, + sandbox, + 'approval-policy': approvalPolicy, + config: { mcp_servers: mcpServers } + }; + if (message.mode.model) { + startConfig.model = message.mode.model; + } + + let resumeFile: string | null = null; + if (nextExperimentalResume) { + resumeFile = nextExperimentalResume; + nextExperimentalResume = null; + logger.debug('[Codex] Using resume file from mode change:', resumeFile); + } else if (storedSessionIdForResume) { + const abortResumeFile = findCodexResumeFile(storedSessionIdForResume); + if (abortResumeFile) { + resumeFile = abortResumeFile; + logger.debug('[Codex] Using resume file from aborted session:', resumeFile); + messageBuffer.addMessage('Resuming from aborted session...', 'status'); + } + storedSessionIdForResume = null; + } + + if (resumeFile) { + (startConfig.config as any).experimental_resume = resumeFile; + } + + await client.startSession(startConfig, { signal: abortController.signal }); + wasCreated = true; + first = false; + syncSessionId(); + } else { + await client.continueSession(message.message, { signal: abortController.signal }); + syncSessionId(); + } + } catch (error) { + logger.warn('Error in codex session:', error); + const isAbortError = error instanceof Error && error.name === 'AbortError'; + + 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'); + } else { + messageBuffer.addMessage('Process exited unexpectedly', 'status'); + session.sendSessionEvent({ type: 'message', message: 'Process exited unexpectedly' }); + if (client.hasActiveSession()) { + storedSessionIdForResume = client.storeSessionForResume(); + logger.debug('[Codex] Stored session after unexpected error:', storedSessionIdForResume); + } + } + } finally { + permissionHandler.reset(); + reasoningProcessor.abort(); + diffProcessor.reset(); + session.onThinkingChange(false); + emitReadyIfIdle({ + pending, + queueSize: () => session.queue.size(), + shouldExit, + sendReady + }); + logActiveHandles('after-turn'); + } + } + } finally { + logger.debug('[codex-remote]: cleanup start'); + try { + await client.disconnect(); + } catch (error) { + logger.debug('[codex-remote]: Error disconnecting client', error); + } + session.client.rpcHandlerManager.registerHandler('abort', async () => {}); + session.client.rpcHandlerManager.registerHandler('switch', async () => {}); + happyServer.stop(); + permissionHandler.reset(); + reasoningProcessor.abort(); + diffProcessor.reset(); + + if (process.stdin.isTTY) { + try { process.stdin.setRawMode(false); } catch {} + } + if (hasTTY) { + try { process.stdin.pause(); } catch {} + } + if (inkInstance) { + inkInstance.unmount(); + } + messageBuffer.clear(); + logger.debug('[codex-remote]: cleanup done'); + } + + return exitReason || 'exit'; +} diff --git a/cli/src/codex/loop.ts b/cli/src/codex/loop.ts new file mode 100644 index 00000000..07d42f61 --- /dev/null +++ b/cli/src/codex/loop.ts @@ -0,0 +1,67 @@ +import { MessageQueue2 } from '@/utils/MessageQueue2'; +import { logger } from '@/ui/logger'; +import { CodexSession } from './session'; +import { codexLocalLauncher } from './codexLocalLauncher'; +import { codexRemoteLauncher } from './codexRemoteLauncher'; +import { ApiClient, ApiSessionClient } from '@/lib'; + +export type PermissionMode = 'default' | 'read-only' | 'safe-yolo' | 'yolo'; + +export interface EnhancedMode { + permissionMode: PermissionMode; + model?: string; +} + +interface LoopOptions { + path: string; + startingMode?: 'local' | 'remote'; + onModeChange: (mode: 'local' | 'remote') => void; + messageQueue: MessageQueue2; + session: ApiSessionClient; + api: ApiClient; + onSessionReady?: (session: CodexSession) => void; +} + +export async function loop(opts: LoopOptions): Promise { + const logPath = logger.getLogPath(); + const session = new CodexSession({ + api: opts.api, + client: opts.session, + path: opts.path, + sessionId: null, + logPath, + messageQueue: opts.messageQueue, + onModeChange: opts.onModeChange, + mode: opts.startingMode ?? 'local' + }); + + if (opts.onSessionReady) { + opts.onSessionReady(session); + } + + let mode: 'local' | 'remote' = opts.startingMode ?? 'local'; + + while (true) { + logger.debug(`[codex-loop] Iteration with mode: ${mode}`); + + if (mode === 'local') { + const reason = await codexLocalLauncher(session); + if (reason === 'exit') { + return; + } + mode = 'remote'; + session.onModeChange(mode); + continue; + } + + if (mode === 'remote') { + const reason = await codexRemoteLauncher(session); + if (reason === 'exit') { + return; + } + mode = 'local'; + session.onModeChange(mode); + continue; + } + } +} diff --git a/cli/src/codex/runCodex.ts b/cli/src/codex/runCodex.ts index 773fa0a4..5c7cd240 100644 --- a/cli/src/codex/runCodex.ts +++ b/cli/src/codex/runCodex.ts @@ -1,108 +1,53 @@ -import { render } from "ink"; -import React from "react"; -import { ApiClient } from '@/api/api'; -import { CodexMcpClient } from './codexMcpClient'; -import { CodexPermissionHandler } from './utils/permissionHandler'; -import { ReasoningProcessor } from './utils/reasoningProcessor'; -import { DiffProcessor } from './utils/diffProcessor'; -import { randomUUID } from 'node:crypto'; -import { logger } from '@/ui/logger'; -import { readSettings } from '@/persistence'; -import { AgentState, Metadata } from '@/api/types'; -import { initialMachineMetadata } from '@/daemon/run'; -import { configuration } from '@/configuration'; -import packageJson from '../../package.json'; import os from 'node:os'; +import { randomUUID } from 'node:crypto'; +import { resolve } from 'node:path'; + +import { ApiClient } from '@/api/api'; +import { logger } from '@/ui/logger'; +import { loop, type EnhancedMode, type PermissionMode } from './loop'; import { MessageQueue2 } from '@/utils/MessageQueue2'; import { hashObject } from '@/utils/deterministicJson'; +import { readSettings } from '@/persistence'; +import { configuration } from '@/configuration'; +import { notifyDaemonSessionStarted } from '@/daemon/controlClient'; +import { initialMachineMetadata } from '@/daemon/run'; +import { registerKillSessionHandler } from '@/claude/registerKillSessionHandler'; +import type { AgentState, Metadata } from '@/api/types'; +import packageJson from '../../package.json'; import { runtimePath } from '@/projectPath'; -import { getHappyCliCommand } from '@/utils/spawnHappyCLI'; -import { resolve, join } from 'node:path'; -import fs from 'node:fs'; -import { startHappyServer } from '@/claude/utils/startHappyServer'; -import { MessageBuffer } from "@/ui/ink/messageBuffer"; -import { CodexDisplay } from "@/ui/ink/CodexDisplay"; -import { trimIdent } from "@/utils/trimIdent"; -import type { CodexSessionConfig } from './types'; -import { notifyDaemonSessionStarted } from "@/daemon/controlClient"; -import { registerKillSessionHandler } from "@/claude/registerKillSessionHandler"; -import { delay } from "@/utils/time"; +import type { CodexSession } from './session'; -type ReadyEventOptions = { - pending: unknown; - queueSize: () => number; - shouldExit: boolean; - sendReady: () => void; - notify?: () => void; -}; +export { emitReadyIfIdle } from './utils/emitReadyIfIdle'; -/** - * Notify connected clients when Codex finishes processing and the queue is idle. - * Returns true when a ready event was emitted. - */ -export function emitReadyIfIdle({ pending, queueSize, shouldExit, sendReady, notify }: ReadyEventOptions): boolean { - if (shouldExit) { - return false; - } - if (pending) { - return false; - } - if (queueSize() > 0) { - return false; - } - - sendReady(); - notify?.(); - return true; -} - -/** - * Main entry point for the codex command with ink UI - */ export async function runCodex(opts: { startedBy?: 'daemon' | 'terminal'; }): Promise { - type PermissionMode = 'default' | 'read-only' | 'safe-yolo' | 'yolo'; - interface EnhancedMode { - permissionMode: PermissionMode; - model?: string; - } - - // - // Define session - // - + const workingDirectory = process.cwd(); const sessionTag = randomUUID(); - const api = await ApiClient.create(); - // Log startup options logger.debug(`[codex] Starting with options: startedBy=${opts.startedBy || 'terminal'}`); - // - // Machine - // + const api = await ApiClient.create(); const settings = await readSettings(); - let machineId = settings?.machineId; + const machineId = settings?.machineId; if (!machineId) { console.error(`[START] No machine ID found in settings, which is unexpected since authAndSetupMachineIfNeeded should have created it. Please report this issue on ${packageJson.bugs}`); process.exit(1); } logger.debug(`Using machineId: ${machineId}`); + await api.getOrCreateMachine({ machineId, metadata: initialMachineMetadata }); - // - // Create session - // - let state: AgentState = { - controlledByUser: false, - } - let metadata: Metadata = { - path: process.cwd(), + controlledByUser: false + }; + + const metadata: Metadata = { + path: workingDirectory, host: os.hostname(), version: packageJson.version, os: os.platform(), @@ -114,15 +59,14 @@ export async function runCodex(opts: { startedFromDaemon: opts.startedBy === 'daemon', hostPid: process.pid, startedBy: opts.startedBy || 'terminal', - // Initialize lifecycle state lifecycleState: 'running', lifecycleStateSince: Date.now(), flavor: 'codex' }; + const response = await api.getOrCreateSession({ tag: sessionTag, metadata, state }); const session = api.sessionSyncClient(response); - // Always report to daemon if it exists try { logger.debug(`[START] Reporting session ${response.id} to daemon`); const result = await notifyDaemonSessionStarted(response.id, metadata); @@ -135,17 +79,22 @@ export async function runCodex(opts: { logger.debug('[START] Failed to report to daemon (may not be running):', error); } + const startingMode: 'local' | 'remote' = opts.startedBy === 'daemon' ? 'remote' : 'local'; + + session.updateAgentState((currentState) => ({ + ...currentState, + controlledByUser: startingMode === 'local' + })); + const messageQueue = new MessageQueue2((mode) => hashObject({ permissionMode: mode.permissionMode, - model: mode.model, + model: mode.model })); - // Track current overrides to apply per message let currentPermissionMode: PermissionMode | undefined = undefined; let currentModel: string | undefined = undefined; session.onUserMessage((message) => { - // Resolve permission mode (validate) let messagePermissionMode = currentPermissionMode; if (message.meta?.permissionMode) { const validModes: PermissionMode[] = ['default', 'read-only', 'safe-yolo', 'yolo']; @@ -160,7 +109,6 @@ export async function runCodex(opts: { logger.debug(`[Codex] User message received with no permission mode override, using current: ${currentPermissionMode ?? 'default (effective)'}`); } - // Resolve model; explicit null resets to default (undefined) let messageModel = currentModel; if (message.meta?.hasOwnProperty('model')) { messageModel = message.meta.model || undefined; @@ -172,585 +120,88 @@ export async function runCodex(opts: { const enhancedMode: EnhancedMode = { permissionMode: messagePermissionMode || 'default', - model: messageModel, + model: messageModel }; messageQueue.push(message.content.text, enhancedMode); }); - let thinking = false; - session.keepAlive(thinking, 'remote'); - // Periodic keep-alive; store handle so we can clear on exit - const keepAliveInterval = setInterval(() => { - session.keepAlive(thinking, 'remote'); - }, 2000); - const sendReady = () => { - session.sendSessionEvent({ type: 'ready' }); - }; + let sessionWrapper: CodexSession | null = null; - // Debug helper: log active handles/requests if DEBUG is enabled - function logActiveHandles(tag: string) { - if (!process.env.DEBUG) return; - const anyProc: any = process as any; - const handles = typeof anyProc._getActiveHandles === 'function' ? anyProc._getActiveHandles() : []; - const requests = typeof anyProc._getActiveRequests === 'function' ? anyProc._getActiveRequests() : []; - logger.debug(`[codex][handles] ${tag}: handles=${handles.length} requests=${requests.length}`); - try { - const kinds = handles.map((h: any) => (h && h.constructor ? h.constructor.name : typeof h)); - logger.debug(`[codex][handles] kinds=${JSON.stringify(kinds)}`); - } catch { } - } + let cleanupStarted = false; + let exitCode = 0; - // - // Abort handling - // IMPORTANT: There are two different operations: - // 1. Abort (handleAbort): Stops the current inference/task but keeps the session alive - // - Used by the 'abort' RPC from mobile app - // - Similar to Claude Code's abort behavior - // - Allows continuing with new prompts after aborting - // 2. Kill (handleKillSession): Terminates the entire process - // - Used by the 'killSession' RPC - // - Completely exits the CLI process - // - - let abortController = new AbortController(); - let shouldExit = false; - let storedSessionIdForResume: string | null = null; - - /** - * Handles aborting the current task/inference without exiting the process. - * This is the equivalent of Claude Code's abort - it stops what's currently - * happening but keeps the session alive for new prompts. - */ - async function handleAbort() { - logger.debug('[Codex] Abort requested - stopping current task'); - try { - // Store the current session ID before aborting for potential resume - if (client.hasActiveSession()) { - storedSessionIdForResume = client.storeSessionForResume(); - logger.debug('[Codex] Stored session for resume:', storedSessionIdForResume); - } - - abortController.abort(); - messageQueue.reset(); - permissionHandler.reset(); - reasoningProcessor.abort(); - diffProcessor.reset(); - logger.debug('[Codex] Abort completed - session remains active'); - } catch (error) { - logger.debug('[Codex] Error during abort:', error); - } finally { - abortController = new AbortController(); + const cleanup = async (code: number = exitCode) => { + if (cleanupStarted) { + return; } - } - - /** - * Handles session termination and process exit. - * This is called when the session needs to be completely killed (not just aborted). - * Abort stops the current inference but keeps the session alive. - * Kill terminates the entire process. - */ - const handleKillSession = async () => { - logger.debug('[Codex] Kill session requested - terminating process'); - await handleAbort(); - logger.debug('[Codex] Abort completed, proceeding with termination'); - + cleanupStarted = true; + logger.debug('[codex] Cleanup start'); try { - // Update lifecycle state to archived before closing - if (session) { - session.updateMetadata((currentMetadata) => ({ - ...currentMetadata, - lifecycleState: 'archived', - lifecycleStateSince: Date.now(), - archivedBy: 'cli', - archiveReason: 'User terminated' - })); - - // Send session death message - session.sendSessionDeath(); - await session.flush(); - await session.close(); + if (sessionWrapper) { + sessionWrapper.stopKeepAlive(); } - // Stop HAPI MCP server - happyServer.stop(); + session.updateMetadata((currentMetadata) => ({ + ...currentMetadata, + lifecycleState: 'archived', + lifecycleStateSince: Date.now(), + archivedBy: 'cli', + archiveReason: 'User terminated' + })); - logger.debug('[Codex] Session termination complete, exiting'); - process.exit(0); + session.sendSessionDeath(); + await session.flush(); + await session.close(); + + logger.debug('[codex] Cleanup complete, exiting'); + process.exit(code); } catch (error) { - logger.debug('[Codex] Error during session termination:', error); + logger.debug('[codex] Error during cleanup:', error); process.exit(1); } }; - // Register abort handler - session.rpcHandlerManager.registerHandler('abort', handleAbort); + process.on('SIGTERM', () => cleanup(0)); + process.on('SIGINT', () => cleanup(0)); - registerKillSessionHandler(session.rpcHandlerManager, handleKillSession); - - // - // Initialize Ink UI - // - - const messageBuffer = new MessageBuffer(); - const hasTTY = process.stdout.isTTY && process.stdin.isTTY; - let inkInstance: any = null; - - if (hasTTY) { - console.clear(); - inkInstance = render(React.createElement(CodexDisplay, { - messageBuffer, - logPath: process.env.DEBUG ? logger.getLogPath() : undefined, - onExit: async () => { - // Exit the agent - logger.debug('[codex]: Exiting agent via Ctrl-C'); - shouldExit = true; - await handleAbort(); - } - }), { - exitOnCtrlC: false, - patchConsole: false - }); - } - - if (hasTTY) { - process.stdin.resume(); - if (process.stdin.isTTY) { - process.stdin.setRawMode(true); - } - process.stdin.setEncoding("utf8"); - } - - // - // Start Context - // - - const client = new CodexMcpClient(); - - // Helper: find Codex session transcript for a given sessionId - 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'); - - // Recursively collect all files under the sessions directory - 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; // newest first - }); - return candidates[0] || null; - } catch { - return null; - } - } - const permissionHandler = new CodexPermissionHandler(session); - const reasoningProcessor = new ReasoningProcessor((message) => { - // Callback to send messages directly from the processor - session.sendCodexMessage(message); - }); - const diffProcessor = new DiffProcessor((message) => { - // Callback to send messages directly from the processor - session.sendCodexMessage(message); - }); - client.setPermissionHandler(permissionHandler); - client.setHandler((msg) => { - logger.debug(`[Codex] MCP message: ${JSON.stringify(msg)}`); - - // Add messages to the ink UI buffer based on message type - if (msg.type === 'agent_message') { - messageBuffer.addMessage(msg.message, 'assistant'); - } else if (msg.type === 'agent_reasoning_delta') { - // Skip reasoning deltas in the UI to reduce noise - } 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); - messageBuffer.addMessage( - `Result: ${truncatedOutput}${output.length > 200 ? '...' : ''}`, - 'result' - ); - } else if (msg.type === 'task_started') { - messageBuffer.addMessage('Starting task...', 'status'); - } else if (msg.type === 'task_complete') { - messageBuffer.addMessage('Task completed', 'status'); - sendReady(); - } else if (msg.type === 'turn_aborted') { - messageBuffer.addMessage('Turn aborted', 'status'); - sendReady(); - } - - if (msg.type === 'task_started') { - if (!thinking) { - logger.debug('thinking started'); - thinking = true; - session.keepAlive(thinking, 'remote'); - } - } - if (msg.type === 'task_complete' || msg.type === 'turn_aborted') { - if (thinking) { - logger.debug('thinking completed'); - thinking = false; - session.keepAlive(thinking, 'remote'); - } - // Reset diff processor on task end or abort - diffProcessor.reset(); - } - if (msg.type === 'agent_reasoning_section_break') { - // Reset reasoning processor for new section - reasoningProcessor.handleSectionBreak(); - } - if (msg.type === 'agent_reasoning_delta') { - // Process reasoning delta - tool calls are sent automatically via callback - reasoningProcessor.processDelta(msg.delta); - } - if (msg.type === 'agent_reasoning') { - // Complete the reasoning section - tool results or reasoning messages sent via callback - reasoningProcessor.complete(msg.text); - } - if (msg.type === 'agent_message') { - session.sendCodexMessage({ - type: 'message', - message: msg.message, - id: randomUUID() - }); - } - if (msg.type === 'exec_command_begin' || msg.type === 'exec_approval_request') { - let { call_id, type, ...inputs } = msg; - session.sendCodexMessage({ - type: 'tool-call', - name: 'CodexBash', - callId: call_id, - input: inputs, - id: randomUUID() - }); - } - if (msg.type === 'exec_command_end') { - let { call_id, type, ...output } = msg; - session.sendCodexMessage({ - type: 'tool-call-result', - callId: call_id, - output: output, - id: randomUUID() - }); - } - if (msg.type === 'token_count') { - session.sendCodexMessage({ - ...msg, - id: randomUUID() - }); - } - if (msg.type === 'patch_apply_begin') { - // Handle the start of a patch operation - let { call_id, auto_approved, changes } = msg; - - // Add UI feedback for patch operation - const changeCount = Object.keys(changes).length; - const filesMsg = changeCount === 1 ? '1 file' : `${changeCount} files`; - messageBuffer.addMessage(`Modifying ${filesMsg}...`, 'tool'); - - // Send tool call message - session.sendCodexMessage({ - type: 'tool-call', - name: 'CodexPatch', - callId: call_id, - input: { - auto_approved, - changes - }, - id: randomUUID() - }); - } - if (msg.type === 'patch_apply_end') { - // Handle the end of a patch operation - let { call_id, stdout, stderr, success } = msg; - - // Add UI feedback for completion - 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'); - } - - // Send tool call result message - session.sendCodexMessage({ - type: 'tool-call-result', - callId: call_id, - output: { - stdout, - stderr, - success - }, - id: randomUUID() - }); - } - if (msg.type === 'turn_diff') { - // Handle turn_diff messages and track unified_diff changes - if (msg.unified_diff) { - diffProcessor.processDiff(msg.unified_diff); - } - } + process.on('uncaughtException', (error) => { + logger.debug('[codex] Uncaught exception:', error); + exitCode = 1; + cleanup(1); }); - // Start HAPI MCP server (HTTP) and prepare STDIO bridge config for Codex - const happyServer = await startHappyServer(session); - const bridgeCommand = getHappyCliCommand(['mcp', '--url', happyServer.url]); - const mcpServers = { - hapi: { - command: bridgeCommand.command, - args: bridgeCommand.args - } - } as const; - let first = true; + process.on('unhandledRejection', (reason) => { + logger.debug('[codex] Unhandled rejection:', reason); + exitCode = 1; + cleanup(1); + }); + registerKillSessionHandler(session.rpcHandlerManager, cleanup); + + let loopError: unknown = null; try { - logger.debug('[codex]: client.connect begin'); - await client.connect(); - logger.debug('[codex]: client.connect done'); - let wasCreated = false; - let currentModeHash: string | null = null; - let pending: { message: string; mode: EnhancedMode; isolate: boolean; hash: string } | null = null; - // If we restart (e.g., mode change), use this to carry a resume file - let nextExperimentalResume: string | null = null; - - while (!shouldExit) { - logActiveHandles('loop-top'); - // Get next batch; respect mode boundaries like Claude - let message: { message: string; mode: EnhancedMode; isolate: boolean; hash: string } | null = pending; - pending = null; - if (!message) { - // Capture the current signal to distinguish idle-abort from queue close - const waitSignal = abortController.signal; - const batch = await messageQueue.waitForMessagesAndGetAsString(waitSignal); - if (!batch) { - // If wait was aborted (e.g., remote abort with no active inference), ignore and continue - if (waitSignal.aborted && !shouldExit) { - logger.debug('[codex]: Wait aborted while idle; ignoring and continuing'); - continue; - } - logger.debug(`[codex]: batch=${!!batch}, shouldExit=${shouldExit}`); - break; - } - message = batch; + await loop({ + path: workingDirectory, + startingMode, + messageQueue, + api, + session, + onModeChange: (newMode) => { + session.sendSessionEvent({ type: 'switch', mode: newMode }); + session.updateAgentState((currentState) => ({ + ...currentState, + controlledByUser: newMode === 'local' + })); + }, + onSessionReady: (instance) => { + sessionWrapper = instance; } - - // Defensive check for TS narrowing - if (!message) { - break; - } - - // If a session exists and mode changed, restart on next iteration - if (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'); - // Capture previous sessionId and try to find its transcript to resume - 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 (e) { - logger.debug('[Codex] Error while searching resume file', e); - } - client.clearSession(); - wasCreated = false; - currentModeHash = null; - pending = message; - // Reset processors/permissions like end-of-turn cleanup - permissionHandler.reset(); - reasoningProcessor.abort(); - diffProcessor.reset(); - thinking = false; - session.keepAlive(thinking, 'remote'); - continue; - } - - // Display user messages in the UI - messageBuffer.addMessage(message.message, 'user'); - currentModeHash = message.hash; - - try { - // Map permission mode to approval policy and sandbox for startSession - const approvalPolicy = (() => { - switch (message.mode.permissionMode) { - case 'default': return 'untrusted' as const; - case 'read-only': return 'never' as const; - case 'safe-yolo': return 'on-failure' as const; - case 'yolo': return 'on-failure' as const; - } - })(); - const sandbox = (() => { - switch (message.mode.permissionMode) { - case 'default': return 'workspace-write' as const; - case 'read-only': return 'read-only' as const; - case 'safe-yolo': return 'workspace-write' as const; - case 'yolo': return 'danger-full-access' as const; - } - })(); - - if (!wasCreated) { - const startConfig: CodexSessionConfig = { - prompt: first ? message.message + '\n\n' + trimIdent(`Based on this message, call functions.hapi__change_title to change chat session title that would represent the current task. If chat idea would change dramatically - call this function again to update the title.`) : message.message, - sandbox, - 'approval-policy': approvalPolicy, - config: { mcp_servers: mcpServers } - }; - if (message.mode.model) { - startConfig.model = message.mode.model; - } - - // Check for resume file from multiple sources - let resumeFile: string | null = null; - - // Priority 1: Explicit resume file from mode change - if (nextExperimentalResume) { - resumeFile = nextExperimentalResume; - nextExperimentalResume = null; // consume once - logger.debug('[Codex] Using resume file from mode change:', resumeFile); - } - // Priority 2: Resume from stored abort session - else if (storedSessionIdForResume) { - const abortResumeFile = findCodexResumeFile(storedSessionIdForResume); - if (abortResumeFile) { - resumeFile = abortResumeFile; - logger.debug('[Codex] Using resume file from aborted session:', resumeFile); - messageBuffer.addMessage('Resuming from aborted session...', 'status'); - } - storedSessionIdForResume = null; // consume once - } - - // Apply resume file if found - if (resumeFile) { - (startConfig.config as any).experimental_resume = resumeFile; - } - - await client.startSession( - startConfig, - { signal: abortController.signal } - ); - wasCreated = true; - first = false; - } else { - const response = await client.continueSession( - message.message, - { signal: abortController.signal } - ); - logger.debug('[Codex] continueSession response:', response); - } - } catch (error) { - logger.warn('Error in codex session:', error); - const isAbortError = error instanceof Error && error.name === 'AbortError'; - - if (isAbortError) { - messageBuffer.addMessage('Aborted by user', 'status'); - session.sendSessionEvent({ type: 'message', message: 'Aborted by user' }); - // Session was already stored in handleAbort(), no need to store again - // Mark session as not created to force proper resume on next message - 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' }); - // For unexpected exits, try to store session for potential recovery - if (client.hasActiveSession()) { - storedSessionIdForResume = client.storeSessionForResume(); - logger.debug('[Codex] Stored session after unexpected error:', storedSessionIdForResume); - } - } - } finally { - // Reset permission handler, reasoning processor, and diff processor - permissionHandler.reset(); - reasoningProcessor.abort(); // Use abort to properly finish any in-progress tool calls - diffProcessor.reset(); - thinking = false; - session.keepAlive(thinking, 'remote'); - emitReadyIfIdle({ - pending, - queueSize: () => messageQueue.size(), - shouldExit, - sendReady, - }); - logActiveHandles('after-turn'); - } - } - + }); + } catch (error) { + loopError = error; + exitCode = 1; + logger.debug('[codex] Loop error:', error); } finally { - // Clean up resources when main loop exits - logger.debug('[codex]: Final cleanup start'); - logActiveHandles('cleanup-start'); - try { - logger.debug('[codex]: sendSessionDeath'); - session.sendSessionDeath(); - logger.debug('[codex]: flush begin'); - await session.flush(); - logger.debug('[codex]: flush done'); - logger.debug('[codex]: session.close begin'); - await session.close(); - logger.debug('[codex]: session.close done'); - } catch (e) { - logger.debug('[codex]: Error while closing session', e); - } - logger.debug('[codex]: client.disconnect begin'); - await client.disconnect(); - logger.debug('[codex]: client.disconnect done'); - // Stop HAPI MCP server - logger.debug('[codex]: happyServer.stop'); - happyServer.stop(); - - // Clean up ink UI - if (process.stdin.isTTY) { - logger.debug('[codex]: setRawMode(false)'); - try { process.stdin.setRawMode(false); } catch { } - } - // Stop reading from stdin so the process can exit - if (hasTTY) { - logger.debug('[codex]: stdin.pause()'); - try { process.stdin.pause(); } catch { } - } - // Clear periodic keep-alive to avoid keeping event loop alive - logger.debug('[codex]: clearInterval(keepAlive)'); - clearInterval(keepAliveInterval); - if (inkInstance) { - logger.debug('[codex]: inkInstance.unmount()'); - inkInstance.unmount(); - } - messageBuffer.clear(); - - logActiveHandles('cleanup-end'); - logger.debug('[codex]: Final cleanup completed'); + await cleanup(loopError ? 1 : exitCode); } } diff --git a/cli/src/codex/session.ts b/cli/src/codex/session.ts new file mode 100644 index 00000000..37a663a5 --- /dev/null +++ b/cli/src/codex/session.ts @@ -0,0 +1,78 @@ +import { ApiClient, ApiSessionClient } from '@/lib'; +import { MessageQueue2 } from '@/utils/MessageQueue2'; +import { logger } from '@/ui/logger'; +import type { EnhancedMode } from './loop'; + +export class CodexSession { + readonly path: string; + readonly logPath: string; + readonly api: ApiClient; + readonly client: ApiSessionClient; + readonly queue: MessageQueue2; + readonly _onModeChange: (mode: 'local' | 'remote') => void; + + sessionId: string | null; + mode: 'local' | 'remote' = 'local'; + thinking: boolean = false; + private keepAliveInterval: NodeJS.Timeout | null = null; + + constructor(opts: { + api: ApiClient; + client: ApiSessionClient; + path: string; + logPath: string; + sessionId: string | null; + messageQueue: MessageQueue2; + onModeChange: (mode: 'local' | 'remote') => void; + mode?: 'local' | 'remote'; + }) { + this.path = opts.path; + this.api = opts.api; + this.client = opts.client; + this.logPath = opts.logPath; + this.sessionId = opts.sessionId; + this.queue = opts.messageQueue; + this._onModeChange = opts.onModeChange; + this.mode = opts.mode ?? 'local'; + + this.client.keepAlive(this.thinking, this.mode); + this.keepAliveInterval = setInterval(() => { + this.client.keepAlive(this.thinking, this.mode); + }, 2000); + } + + onThinkingChange = (thinking: boolean) => { + this.thinking = thinking; + this.client.keepAlive(thinking, this.mode); + }; + + onModeChange = (mode: 'local' | 'remote') => { + this.mode = mode; + this.client.keepAlive(this.thinking, mode); + this._onModeChange(mode); + }; + + onSessionFound = (sessionId: string) => { + this.sessionId = sessionId; + this.client.updateMetadata((metadata) => ({ + ...metadata, + codexSessionId: sessionId + })); + logger.debug(`[CodexSession] Codex session ID ${sessionId} added to metadata`); + }; + + sendCodexMessage = (message: unknown): void => { + this.client.sendCodexMessage(message); + }; + + sendSessionEvent = (event: Parameters[0]): void => { + this.client.sendSessionEvent(event); + }; + + stopKeepAlive = (): void => { + if (this.keepAliveInterval) { + clearInterval(this.keepAliveInterval); + this.keepAliveInterval = null; + } + }; +} diff --git a/cli/src/codex/utils/codexEventConverter.test.ts b/cli/src/codex/utils/codexEventConverter.test.ts new file mode 100644 index 00000000..446d2e13 --- /dev/null +++ b/cli/src/codex/utils/codexEventConverter.test.ts @@ -0,0 +1,85 @@ +import { describe, expect, it } from 'vitest'; +import { convertCodexEvent } from './codexEventConverter'; + +describe('convertCodexEvent', () => { + it('extracts session_meta id', () => { + const result = convertCodexEvent({ + type: 'session_meta', + payload: { id: 'session-123' } + }); + + expect(result).toEqual({ sessionId: 'session-123' }); + }); + + it('converts agent_message events', () => { + const result = convertCodexEvent({ + type: 'event_msg', + payload: { type: 'agent_message', message: 'hello' } + }); + + expect(result?.message).toMatchObject({ + type: 'message', + message: 'hello' + }); + }); + + it('converts reasoning events', () => { + const result = convertCodexEvent({ + type: 'event_msg', + payload: { type: 'agent_reasoning', text: 'thinking' } + }); + + expect(result?.message).toMatchObject({ + type: 'reasoning', + message: 'thinking' + }); + }); + + it('converts reasoning delta events', () => { + const result = convertCodexEvent({ + type: 'event_msg', + payload: { type: 'agent_reasoning_delta', delta: 'step' } + }); + + expect(result?.message).toEqual({ + type: 'reasoning-delta', + delta: 'step' + }); + }); + + it('converts function_call items', () => { + const result = convertCodexEvent({ + type: 'response_item', + payload: { + type: 'function_call', + name: 'ToolName', + call_id: 'call-1', + arguments: '{"foo":"bar"}' + } + }); + + expect(result?.message).toMatchObject({ + type: 'tool-call', + name: 'ToolName', + callId: 'call-1', + input: { foo: 'bar' } + }); + }); + + it('converts function_call_output items', () => { + const result = convertCodexEvent({ + type: 'response_item', + payload: { + type: 'function_call_output', + call_id: 'call-2', + output: { ok: true } + } + }); + + expect(result?.message).toMatchObject({ + type: 'tool-call-result', + callId: 'call-2', + output: { ok: true } + }); + }); +}); diff --git a/cli/src/codex/utils/codexEventConverter.ts b/cli/src/codex/utils/codexEventConverter.ts new file mode 100644 index 00000000..e783d92a --- /dev/null +++ b/cli/src/codex/utils/codexEventConverter.ts @@ -0,0 +1,224 @@ +import { randomUUID } from 'node:crypto'; +import { z } from 'zod'; +import { logger } from '@/ui/logger'; + +const CodexSessionEventSchema = z.object({ + timestamp: z.string().optional(), + type: z.string(), + payload: z.unknown().optional() +}).passthrough(); + +export type CodexSessionEvent = z.infer; + +export type CodexMessage = { + type: 'message'; + message: string; + id: string; +} | { + type: 'reasoning'; + message: string; + id: string; +} | { + type: 'reasoning-delta'; + delta: string; +} | { + type: 'token_count'; + info: Record; + id: string; +} | { + type: 'tool-call'; + name: string; + callId: string; + input: unknown; + id: string; +} | { + type: 'tool-call-result'; + callId: string; + output: unknown; + id: string; +}; + +export type CodexConversionResult = { + sessionId?: string; + message?: CodexMessage; +}; + +function asRecord(value: unknown): Record | null { + if (!value || typeof value !== 'object') { + return null; + } + return value as Record; +} + +function asString(value: unknown): string | null { + return typeof value === 'string' && value.length > 0 ? value : null; +} + +function parseArguments(value: unknown): unknown { + if (typeof value !== 'string') { + return value; + } + + const trimmed = value.trim(); + if (trimmed.startsWith('{') || trimmed.startsWith('[')) { + try { + return JSON.parse(trimmed); + } catch (error) { + logger.debug('[codexEventConverter] Failed to parse function_call arguments as JSON:', error); + } + } + + return value; +} + +function extractCallId(payload: Record): string | null { + const candidates = [ + 'call_id', + 'callId', + 'tool_call_id', + 'toolCallId', + 'id' + ]; + + for (const key of candidates) { + const value = payload[key]; + if (typeof value === 'string' && value.length > 0) { + return value; + } + } + + return null; +} + +export function convertCodexEvent(rawEvent: unknown): CodexConversionResult | null { + const parsed = CodexSessionEventSchema.safeParse(rawEvent); + if (!parsed.success) { + return null; + } + + const { type, payload } = parsed.data; + const payloadRecord = asRecord(payload); + + if (type === 'session_meta') { + const sessionId = payloadRecord ? asString(payloadRecord.id) : null; + if (!sessionId) { + return null; + } + return { sessionId }; + } + + if (!payloadRecord) { + return null; + } + + if (type === 'event_msg') { + const eventType = asString(payloadRecord.type); + if (!eventType) { + return null; + } + + if (eventType === 'user_message') { + return null; + } + + if (eventType === 'agent_message') { + const message = asString(payloadRecord.message); + if (!message) { + return null; + } + return { + message: { + type: 'message', + message, + id: randomUUID() + } + }; + } + + if (eventType === 'agent_reasoning') { + const message = asString(payloadRecord.text) ?? asString(payloadRecord.message); + if (!message) { + return null; + } + return { + message: { + type: 'reasoning', + message, + id: randomUUID() + } + }; + } + + if (eventType === 'agent_reasoning_delta') { + const delta = asString(payloadRecord.delta) ?? asString(payloadRecord.text) ?? asString(payloadRecord.message); + if (!delta) { + return null; + } + return { + message: { + type: 'reasoning-delta', + delta + } + }; + } + + if (eventType === 'token_count') { + const info = asRecord(payloadRecord.info); + if (!info) { + return null; + } + return { + message: { + type: 'token_count', + info, + id: randomUUID() + } + }; + } + + return null; + } + + if (type === 'response_item') { + const itemType = asString(payloadRecord.type); + if (!itemType) { + return null; + } + + if (itemType === 'function_call') { + const name = asString(payloadRecord.name); + const callId = extractCallId(payloadRecord); + if (!name || !callId) { + return null; + } + return { + message: { + type: 'tool-call', + name, + callId, + input: parseArguments(payloadRecord.arguments), + id: randomUUID() + } + }; + } + + if (itemType === 'function_call_output') { + const callId = extractCallId(payloadRecord); + if (!callId) { + return null; + } + return { + message: { + type: 'tool-call-result', + callId, + output: payloadRecord.output, + id: randomUUID() + } + }; + } + + return null; + } + + return null; +} diff --git a/cli/src/codex/utils/codexSessionScanner.test.ts b/cli/src/codex/utils/codexSessionScanner.test.ts new file mode 100644 index 00000000..8ea8e78c --- /dev/null +++ b/cli/src/codex/utils/codexSessionScanner.test.ts @@ -0,0 +1,76 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { mkdir, writeFile, appendFile, rm } from 'node:fs/promises'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { existsSync } from 'node:fs'; +import { createCodexSessionScanner } from './codexSessionScanner'; +import type { CodexSessionEvent } from './codexEventConverter'; + +const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe('codexSessionScanner', () => { + let testDir: string; + let sessionsDir: string; + let sessionFile: string; + let originalCodexHome: string | undefined; + let scanner: Awaited> | null = null; + let events: CodexSessionEvent[] = []; + + beforeEach(async () => { + testDir = join(tmpdir(), `codex-scanner-${Date.now()}`); + sessionsDir = join(testDir, 'sessions', '2025', '12', '22'); + await mkdir(sessionsDir, { recursive: true }); + + originalCodexHome = process.env.CODEX_HOME; + process.env.CODEX_HOME = testDir; + + events = []; + }); + + afterEach(async () => { + if (scanner) { + await scanner.cleanup(); + scanner = null; + } + + if (originalCodexHome === undefined) { + delete process.env.CODEX_HOME; + } else { + process.env.CODEX_HOME = originalCodexHome; + } + + if (existsSync(testDir)) { + await rm(testDir, { recursive: true, force: true }); + } + }); + + it('emits only new events after startup', async () => { + const sessionId = 'session-123'; + sessionFile = join(sessionsDir, `codex-${sessionId}.jsonl`); + + const initialLines = [ + JSON.stringify({ type: 'session_meta', payload: { id: sessionId } }), + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'hello' } }) + ]; + + await writeFile(sessionFile, initialLines.join('\n') + '\n'); + + scanner = await createCodexSessionScanner({ + sessionId, + onEvent: (event) => events.push(event) + }); + + await wait(150); + expect(events).toHaveLength(0); + + const newLine = JSON.stringify({ + type: 'response_item', + payload: { type: 'function_call', name: 'Tool', call_id: 'call-1', arguments: '{}' } + }); + await appendFile(sessionFile, newLine + '\n'); + + await wait(200); + expect(events).toHaveLength(1); + expect(events[0].type).toBe('response_item'); + }); +}); diff --git a/cli/src/codex/utils/codexSessionScanner.ts b/cli/src/codex/utils/codexSessionScanner.ts new file mode 100644 index 00000000..900d307b --- /dev/null +++ b/cli/src/codex/utils/codexSessionScanner.ts @@ -0,0 +1,219 @@ +import { InvalidateSync } from '@/utils/sync'; +import { startFileWatcher } from '@/modules/watcher/startFileWatcher'; +import { logger } from '@/ui/logger'; +import { join } from 'node:path'; +import { homedir } from 'node:os'; +import { readFile, readdir, stat } from 'node:fs/promises'; +import type { CodexSessionEvent } from './codexEventConverter'; + +interface CodexSessionScannerOptions { + sessionId: string | null; + onEvent: (event: CodexSessionEvent) => void; + onSessionFound?: (sessionId: string) => void; +} + +interface CodexSessionScanner { + cleanup: () => Promise; + onNewSession: (sessionId: string) => void; +} + +export async function createCodexSessionScanner(opts: CodexSessionScannerOptions): Promise { + const codexHomeDir = process.env.CODEX_HOME || join(homedir(), '.codex'); + const sessionsRoot = join(codexHomeDir, 'sessions'); + + const processedLineCounts = new Map(); + const watchers = new Map void>(); + const sessionIdByFile = new Map(); + + let activeSessionId: string | null = opts.sessionId; + let reportedSessionId: string | null = opts.sessionId; + let isClosing = false; + + const reportSessionId = (sessionId: string) => { + if (reportedSessionId === sessionId) { + return; + } + reportedSessionId = sessionId; + opts.onSessionFound?.(sessionId); + }; + + const setActiveSessionId = (sessionId: string) => { + activeSessionId = sessionId; + reportSessionId(sessionId); + }; + + async function listSessionFiles(dir: string): Promise { + try { + const entries = await readdir(dir, { withFileTypes: true }); + const results: string[] = []; + for (const entry of entries) { + const full = join(dir, entry.name); + if (entry.isDirectory()) { + results.push(...await listSessionFiles(full)); + } else if (entry.isFile() && entry.name.endsWith('.jsonl')) { + results.push(full); + } + } + return results; + } catch (error) { + return []; + } + } + + async function readSessionFile(filePath: string, startLine: number): Promise<{ events: CodexSessionEvent[]; totalLines: number }> { + let content: string; + try { + content = await readFile(filePath, 'utf-8'); + } catch (error) { + return { events: [], totalLines: startLine }; + } + + const events: CodexSessionEvent[] = []; + const lines = content.split('\n'); + const hasTrailingEmpty = lines.length > 0 && lines[lines.length - 1] === ''; + const totalLines = hasTrailingEmpty ? lines.length - 1 : lines.length; + let effectiveStartLine = startLine; + if (effectiveStartLine > totalLines) { + effectiveStartLine = 0; + } + + const hasSessionId = sessionIdByFile.has(filePath); + const parseFrom = hasSessionId ? effectiveStartLine : 0; + + for (let index = parseFrom; index < lines.length; index += 1) { + const trimmed = lines[index].trim(); + if (!trimmed) { + continue; + } + try { + const parsed = JSON.parse(trimmed); + if (parsed?.type === 'session_meta') { + const payload = asRecord(parsed.payload); + const sessionId = payload ? asString(payload.id) : null; + if (sessionId) { + sessionIdByFile.set(filePath, sessionId); + } + } + if (index >= effectiveStartLine) { + events.push(parsed); + } + } catch (error) { + logger.debug(`[CODEX_SESSION_SCANNER] Failed to parse line: ${error}`); + } + } + + return { events, totalLines }; + } + + async function initializeProcessedMessages(): Promise { + const files = await listSessionFiles(sessionsRoot); + for (const filePath of files) { + const { totalLines } = await readSessionFile(filePath, 0); + processedLineCounts.set(filePath, totalLines); + if (!isClosing && !watchers.has(filePath)) { + watchers.set(filePath, startFileWatcher(filePath, () => sync.invalidate())); + } + } + } + + const sync = new InvalidateSync(async () => { + if (isClosing) { + return; + } + const files = await listSessionFiles(sessionsRoot); + const sortedFiles = await sortFilesByMtime(files); + + for (const filePath of sortedFiles) { + if (isClosing) { + return; + } + if (!watchers.has(filePath)) { + watchers.set(filePath, startFileWatcher(filePath, () => sync.invalidate())); + } + + const fileSessionId = sessionIdByFile.get(filePath); + if (activeSessionId && fileSessionId && fileSessionId !== activeSessionId) { + continue; + } + if (activeSessionId && !fileSessionId && !filePath.endsWith(`-${activeSessionId}.jsonl`)) { + continue; + } + + const lastProcessedLine = processedLineCounts.get(filePath) ?? 0; + const { events, totalLines } = await readSessionFile(filePath, lastProcessedLine); + processedLineCounts.set(filePath, totalLines); + let emittedForFile = 0; + + for (const event of events) { + const payload = asRecord(event.payload); + const payloadSessionId = payload ? asString(payload.id) : null; + const eventSessionId = payloadSessionId ?? fileSessionId ?? null; + + if (!activeSessionId && eventSessionId) { + setActiveSessionId(eventSessionId); + } + + if (activeSessionId && eventSessionId && eventSessionId !== activeSessionId) { + continue; + } + + opts.onEvent(event); + emittedForFile += 1; + } + + if (emittedForFile > 0) { + logger.debug(`[CODEX_SESSION_SCANNER] Emitted ${emittedForFile} new events from ${filePath}`); + } + } + }); + + await initializeProcessedMessages(); + await sync.invalidateAndAwait(); + const intervalId = setInterval(() => sync.invalidate(), 2000); + + return { + cleanup: async () => { + isClosing = true; + clearInterval(intervalId); + sync.stop(); + for (const stop of watchers.values()) { + stop(); + } + watchers.clear(); + }, + onNewSession: (sessionId: string) => { + if (activeSessionId === sessionId) { + return; + } + logger.debug(`[CODEX_SESSION_SCANNER] Switching to new session: ${sessionId}`); + setActiveSessionId(sessionId); + sync.invalidate(); + } + }; +} + +async function sortFilesByMtime(files: string[]): Promise { + const entries = await Promise.all(files.map(async (file) => { + try { + const stats = await stat(file); + return { file, mtimeMs: stats.mtimeMs }; + } catch { + return { file, mtimeMs: 0 }; + } + })); + + return entries + .sort((a, b) => b.mtimeMs - a.mtimeMs) + .map((entry) => entry.file); +} + +function asRecord(value: unknown): Record | null { + if (!value || typeof value !== 'object') { + return null; + } + return value as Record; +} + +function asString(value: unknown): string | null { + return typeof value === 'string' && value.length > 0 ? value : null; +} diff --git a/cli/src/codex/utils/emitReadyIfIdle.ts b/cli/src/codex/utils/emitReadyIfIdle.ts new file mode 100644 index 00000000..1b70962e --- /dev/null +++ b/cli/src/codex/utils/emitReadyIfIdle.ts @@ -0,0 +1,27 @@ +export type ReadyEventOptions = { + pending: unknown; + queueSize: () => number; + shouldExit: boolean; + sendReady: () => void; + notify?: () => void; +}; + +/** + * Notify connected clients when Codex finishes processing and the queue is idle. + * Returns true when a ready event was emitted. + */ +export function emitReadyIfIdle({ pending, queueSize, shouldExit, sendReady, notify }: ReadyEventOptions): boolean { + if (shouldExit) { + return false; + } + if (pending) { + return false; + } + if (queueSize() > 0) { + return false; + } + + sendReady(); + notify?.(); + return true; +} diff --git a/cli/src/ui/ink/CodexDisplay.tsx b/cli/src/ui/ink/CodexDisplay.tsx index f05f4b66..4d042c2b 100644 --- a/cli/src/ui/ink/CodexDisplay.tsx +++ b/cli/src/ui/ink/CodexDisplay.tsx @@ -6,12 +6,13 @@ interface CodexDisplayProps { messageBuffer: MessageBuffer logPath?: string onExit?: () => void + onSwitchToLocal?: () => void } -export const CodexDisplay: React.FC = ({ messageBuffer, logPath, onExit }) => { +export const CodexDisplay: React.FC = ({ messageBuffer, logPath, onExit, onSwitchToLocal }) => { const [messages, setMessages] = useState([]) - const [confirmationMode, setConfirmationMode] = useState(false) - const [actionInProgress, setActionInProgress] = useState(false) + const [confirmationMode, setConfirmationMode] = useState<'exit' | 'switch' | null>(null) + const [actionInProgress, setActionInProgress] = useState<'exiting' | 'switching' | null>(null) const confirmationTimeoutRef = useRef(null) const { stdout } = useStdout() const terminalWidth = stdout.columns || 80 @@ -33,15 +34,15 @@ export const CodexDisplay: React.FC = ({ messageBuffer, logPa }, [messageBuffer]) const resetConfirmation = useCallback(() => { - setConfirmationMode(false) + setConfirmationMode(null) if (confirmationTimeoutRef.current) { clearTimeout(confirmationTimeoutRef.current) confirmationTimeoutRef.current = null } }, []) - const setConfirmationWithTimeout = useCallback(() => { - setConfirmationMode(true) + const setConfirmationWithTimeout = useCallback((mode: 'exit' | 'switch') => { + setConfirmationMode(mode) if (confirmationTimeoutRef.current) { clearTimeout(confirmationTimeoutRef.current) } @@ -56,25 +57,38 @@ export const CodexDisplay: React.FC = ({ messageBuffer, logPa // Handle Ctrl-C - exits the agent directly instead of switching modes if (key.ctrl && input === 'c') { - if (confirmationMode) { + if (confirmationMode === 'exit') { // Second Ctrl-C, exit resetConfirmation() - setActionInProgress(true) + setActionInProgress('exiting') // Small delay to show the status message await new Promise(resolve => setTimeout(resolve, 100)) onExit?.() } else { // First Ctrl-C, show confirmation - setConfirmationWithTimeout() + setConfirmationWithTimeout('exit') + } + return + } + + const isSpace = input === ' ' || key.name === 'space' + + if (isSpace && onSwitchToLocal) { + if (confirmationMode === 'switch') { + resetConfirmation() + setActionInProgress('switching') + await new Promise(resolve => setTimeout(resolve, 100)) + onSwitchToLocal() + } else { + setConfirmationWithTimeout('switch') } return } - // Any other key cancels confirmation if (confirmationMode) { resetConfirmation() } - }, [confirmationMode, actionInProgress, onExit, setConfirmationWithTimeout, resetConfirmation])) + }, [confirmationMode, actionInProgress, onExit, onSwitchToLocal, setConfirmationWithTimeout, resetConfirmation])) const getMessageColor = (type: BufferedMessage['type']): string => { switch (type) { @@ -140,7 +154,8 @@ export const CodexDisplay: React.FC = ({ messageBuffer, logPa borderStyle="round" borderColor={ actionInProgress ? "gray" : - confirmationMode ? "red" : + confirmationMode === 'exit' ? "red" : + confirmationMode === 'switch' ? "yellow" : "green" } paddingX={2} @@ -149,18 +164,26 @@ export const CodexDisplay: React.FC = ({ messageBuffer, logPa flexDirection="column" > - {actionInProgress ? ( + {actionInProgress === 'exiting' ? ( Exiting agent... - ) : confirmationMode ? ( + ) : actionInProgress === 'switching' ? ( + + Switching to local mode... + + ) : confirmationMode === 'exit' ? ( ⚠️ Press Ctrl-C again to exit the agent + ) : confirmationMode === 'switch' ? ( + + ⏸️ Press space again to switch to local mode + ) : ( <> - 🤖 Codex Agent Running • Ctrl-C to exit + 🤖 Codex Agent Running {onSwitchToLocal ? '• Press space to switch to local mode • Ctrl-C to exit' : '• Ctrl-C to exit'} )} @@ -173,4 +196,4 @@ export const CodexDisplay: React.FC = ({ messageBuffer, logPa ) -} \ No newline at end of file +} diff --git a/cli/src/ui/ink/RemoteModeDisplay.tsx b/cli/src/ui/ink/RemoteModeDisplay.tsx index 2a569148..a7213bb7 100644 --- a/cli/src/ui/ink/RemoteModeDisplay.tsx +++ b/cli/src/ui/ink/RemoteModeDisplay.tsx @@ -71,8 +71,10 @@ export const RemoteModeDisplay: React.FC = ({ messageBuf return } + const isSpace = input === ' ' || key.name === 'space' + // Handle double space - if (input === ' ') { + if (isSpace) { if (confirmationMode === 'switch') { // Second space, switch to local resetConfirmation() @@ -199,4 +201,4 @@ export const RemoteModeDisplay: React.FC = ({ messageBuf ) -} \ No newline at end of file +} diff --git a/cli/src/ui/ink/messageBuffer.ts b/cli/src/ui/ink/messageBuffer.ts index cf12b058..f3b8a187 100644 --- a/cli/src/ui/ink/messageBuffer.ts +++ b/cli/src/ui/ink/messageBuffer.ts @@ -5,6 +5,8 @@ export interface BufferedMessage { type: 'user' | 'assistant' | 'system' | 'tool' | 'result' | 'status' } +const MAX_MESSAGE_COUNT = 500 + export class MessageBuffer { private messages: BufferedMessage[] = [] private listeners: Array<(messages: BufferedMessage[]) => void> = [] @@ -18,6 +20,9 @@ export class MessageBuffer { type } this.messages.push(message) + if (this.messages.length > MAX_MESSAGE_COUNT) { + this.messages.splice(0, this.messages.length - MAX_MESSAGE_COUNT) + } this.notifyListeners() } @@ -45,4 +50,4 @@ export class MessageBuffer { const messages = this.getMessages() this.listeners.forEach(listener => listener(messages)) } -} \ No newline at end of file +}