diff --git a/bun.lock b/bun.lock index f99aa81..7ba4144 100644 --- a/bun.lock +++ b/bun.lock @@ -61,8 +61,7 @@ "name": "ejclaw-codex-runner", "version": "1.0.0", "dependencies": { - "@openai/codex": "0.136.0", - "@openai/codex-sdk": "0.136.0", + "@openai/codex": "0.133.0", "ejclaw-runners-shared": "workspace:*", }, "devDependencies": { @@ -222,21 +221,19 @@ "@napi-rs/wasm-runtime": ["@napi-rs/wasm-runtime@1.1.4", "", { "dependencies": { "@tybys/wasm-util": "^0.10.1" }, "peerDependencies": { "@emnapi/core": "^1.7.1", "@emnapi/runtime": "^1.7.1" } }, "sha512-3NQNNgA1YSlJb/kMH1ildASP9HW7/7kYnRI2szWJaofaS1hWmbGI4H+d3+22aGzXXN9IJ+n+GiFVcGipJP18ow=="], - "@openai/codex": ["@openai/codex@0.136.0", "", { "optionalDependencies": { "@openai/codex-darwin-arm64": "npm:@openai/codex@0.136.0-darwin-arm64", "@openai/codex-darwin-x64": "npm:@openai/codex@0.136.0-darwin-x64", "@openai/codex-linux-arm64": "npm:@openai/codex@0.136.0-linux-arm64", "@openai/codex-linux-x64": "npm:@openai/codex@0.136.0-linux-x64", "@openai/codex-win32-arm64": "npm:@openai/codex@0.136.0-win32-arm64", "@openai/codex-win32-x64": "npm:@openai/codex@0.136.0-win32-x64" }, "bin": { "codex": "bin/codex.js" } }, "sha512-yu+diXznSOt8h296/CDOf2CNi55z1raA7aZ6sejjGy1QWPqn72YkBc2ByTzZG6G+1yiUqlm2G5Hid54hm3/kaA=="], + "@openai/codex": ["@openai/codex@0.133.0", "", { "optionalDependencies": { "@openai/codex-darwin-arm64": "npm:@openai/codex@0.133.0-darwin-arm64", "@openai/codex-darwin-x64": "npm:@openai/codex@0.133.0-darwin-x64", "@openai/codex-linux-arm64": "npm:@openai/codex@0.133.0-linux-arm64", "@openai/codex-linux-x64": "npm:@openai/codex@0.133.0-linux-x64", "@openai/codex-win32-arm64": "npm:@openai/codex@0.133.0-win32-arm64", "@openai/codex-win32-x64": "npm:@openai/codex@0.133.0-win32-x64" }, "bin": { "codex": "bin/codex.js" } }, "sha512-Gh42kLLBo/6gpnHmDzUWDVvyS57ekCB1+1Dz0RG2oIl3Lhk1uwrjSj/PwaJWWh4Rw/rUp1RqkwrMugFfFEOlqQ=="], - "@openai/codex-darwin-arm64": ["@openai/codex@0.136.0-darwin-arm64", "", { "os": "darwin", "cpu": "arm64" }, "sha512-9xnEohnUO9l6qtPSDbG0Y2UXX5mBZ9Lsn0VMgjtkREGfWL9TawNC1d7D1ERMSJtHTlVvieoaFyK9LHPLQCO9Vw=="], + "@openai/codex-darwin-arm64": ["@openai/codex@0.133.0-darwin-arm64", "", { "os": "darwin", "cpu": "arm64" }, "sha512-W7f8+DckLujnqGlptKCzgJU+ooeHKMuk6KYgMFP6A9asn7YUsGUgJqjiBaX8oNcXO6w/pTbKGRARx1kCNS8lIg=="], - "@openai/codex-darwin-x64": ["@openai/codex@0.136.0-darwin-x64", "", { "os": "darwin", "cpu": "x64" }, "sha512-HD9YLalPg0cv+CiZSmRWOl/8sefuJlxJcAmxyzb8mSaAMchD3V+2yS6M6E1Z2I6NX1irp4Polt1fPsjGdlHYhQ=="], + "@openai/codex-darwin-x64": ["@openai/codex@0.133.0-darwin-x64", "", { "os": "darwin", "cpu": "x64" }, "sha512-Ek8ikvLOiXZ8emcIJVBXxK6fm8ratBy0kaEt3JNisTNszxGshUHf/R4xxDxIyKNcUkYYXjW7A/rMwW3iu3OFlg=="], - "@openai/codex-linux-arm64": ["@openai/codex@0.136.0-linux-arm64", "", { "os": "linux", "cpu": "arm64" }, "sha512-xDxTOk5ClEX/nzlQseqEoiubjIzVZfcWKn9nmKkHOLKknz+c7n6tbmw7eY4mwt29zTAAoFoecdRnbbsPZUHJ1g=="], + "@openai/codex-linux-arm64": ["@openai/codex@0.133.0-linux-arm64", "", { "os": "linux", "cpu": "arm64" }, "sha512-uKXYYSJ3mY16sp4hcG/4BMNRjva/ZS4oARiI1+7k8+NiuoAhdCGWNe5u4KJ3sMuL3tp/IXcmc6B56EFX1+WDBQ=="], - "@openai/codex-linux-x64": ["@openai/codex@0.136.0-linux-x64", "", { "os": "linux", "cpu": "x64" }, "sha512-Q6yZHRtUWD+27B5yAiOYyPAtM0BzW0Mm23YGaXDmfNEO5BYgHuUT80VDofUpjSyo2+ShYoZQUFVQAsNPKqd+Sw=="], + "@openai/codex-linux-x64": ["@openai/codex@0.133.0-linux-x64", "", { "os": "linux", "cpu": "x64" }, "sha512-9YfyqrfUj/UZ2+aXE4zBz47t6RXbVni95ZorGsNh857vxYK/asVpUtR2cymo9lB3JaI4mQaKFfV/t7IRItqkuA=="], - "@openai/codex-sdk": ["@openai/codex-sdk@0.136.0", "", { "dependencies": { "@openai/codex": "0.136.0" } }, "sha512-qrSirrVxrpVHR1YIQQ4WSouneaQrQrRAAlQWJYRflvMCcruFXorsmIPeIbDqFXOa8A9H9FHHyc+2U4tdbEGGiQ=="], + "@openai/codex-win32-arm64": ["@openai/codex@0.133.0-win32-arm64", "", { "os": "win32", "cpu": "arm64" }, "sha512-mRzND0PSGHRoLk0X41GTSoc3tFjZSF4HgDlfjU5fiQcWVi0/kLb7Ku6/tPFT/X2hOLa3YdJkbIcHC0Hc9ni80g=="], - "@openai/codex-win32-arm64": ["@openai/codex@0.136.0-win32-arm64", "", { "os": "win32", "cpu": "arm64" }, "sha512-mAHZXfygQxBg1JjZ9+bAZfBXe6fg8MabJhuDYhj5z0VF++orKSFsC1lMiv6PksQN36ikQefgVjc9EOlaE4lt7g=="], - - "@openai/codex-win32-x64": ["@openai/codex@0.136.0-win32-x64", "", { "os": "win32", "cpu": "x64" }, "sha512-zS6DAmvjdWeAB1CL9KTUMkwzTwfXtxHy8GAtePw2a93jIqawoG07fBxAXuyoHZ3QXQkwEgqBx1zEEh33gdIKAw=="], + "@openai/codex-win32-x64": ["@openai/codex@0.133.0-win32-x64", "", { "os": "win32", "cpu": "x64" }, "sha512-u3ji78DIPZCGJeELuovsAnaZH+vK9gsA4F6M1y+Uy2s80Sz7/i1S0KL81qGReYji3urSjgBpkQuNP47GXOqxrQ=="], "@oxc-project/types": ["@oxc-project/types@0.127.0", "", {}, "sha512-aIYXQBo4lCbO4z0R3FHeucQHpF46l2LbMdxRvqvuRuW2OxdnSkcng5B8+K12spgLDj93rtN3+J2Vac/TIO+ciQ=="], diff --git a/runners/codex-runner/package.json b/runners/codex-runner/package.json index 4387092..ea04ccc 100644 --- a/runners/codex-runner/package.json +++ b/runners/codex-runner/package.json @@ -10,8 +10,7 @@ "start": "bun dist/index.js" }, "dependencies": { - "@openai/codex": "0.136.0", - "@openai/codex-sdk": "0.136.0", + "@openai/codex": "0.133.0", "ejclaw-runners-shared": "workspace:*" }, "devDependencies": { diff --git a/runners/codex-runner/src/codex-sdk-client.ts b/runners/codex-runner/src/codex-sdk-client.ts deleted file mode 100644 index 2ea1fb6..0000000 --- a/runners/codex-runner/src/codex-sdk-client.ts +++ /dev/null @@ -1,301 +0,0 @@ -import { - Codex, - type CodexOptions, - type Input, - type ModelReasoningEffort, - type ThreadEvent, - type ThreadOptions, - type Usage, - type UserInput, -} from '@openai/codex-sdk'; - -import type { AppServerInputItem } from './app-server-client.js'; - -export interface CodexSdkThreadPort { - id: string | null; - runStreamed( - input: Input | unknown, - options?: { signal?: AbortSignal }, - ): Promise<{ events: AsyncIterable }>; -} - -export interface CodexSdkPort { - startThread(options?: ThreadOptions): CodexSdkThreadPort; - resumeThread(id: string, options?: ThreadOptions): CodexSdkThreadPort; -} - -export interface CodexSdkClientOptions { - codex?: CodexSdkPort; - env?: Record; - log: (message: string) => void; - codexPathOverride?: string; - config?: CodexOptions['config']; -} - -export interface CodexSdkThreadStartOptions { - cwd: string; - model?: string; - effort?: string; -} - -export interface CodexSdkTurnOptions extends CodexSdkThreadStartOptions { - onProgress?: (message: string) => void; -} - -export interface CodexSdkTurnState { - status: 'pending' | 'inProgress' | 'completed' | 'failed' | 'interrupted'; - threadId: string | null; - finalAnswer: string | null; - latestAgentMessage: string | null; - errorMessage: string | null; - usage: Usage | null; -} - -export interface CodexSdkTurnResult { - state: CodexSdkTurnState; - result: string | null; - threadId: string | null; -} - -export function createInitialCodexSdkTurnState(): CodexSdkTurnState { - return { - status: 'pending', - threadId: null, - finalAnswer: null, - latestAgentMessage: null, - errorMessage: null, - usage: null, - }; -} - -function isKnownReasoningEffort( - effort: string | undefined, -): effort is ModelReasoningEffort { - return ( - effort === 'minimal' || - effort === 'low' || - effort === 'medium' || - effort === 'high' || - effort === 'xhigh' - ); -} - -function normalizeReasoningEffort( - effort: string | undefined, -): ModelReasoningEffort | undefined { - if (effort === 'minimal') { - return 'low'; - } - return isKnownReasoningEffort(effort) ? effort : undefined; -} - -function buildThreadOptions( - options: CodexSdkThreadStartOptions, -): ThreadOptions { - return { - approvalPolicy: 'never', - model: options.model, - modelReasoningEffort: normalizeReasoningEffort(options.effort), - networkAccessEnabled: true, - sandboxMode: 'danger-full-access', - workingDirectory: options.cwd, - }; -} - -export function toCodexSdkInput(input: AppServerInputItem[]): UserInput[] { - return input.map((item) => { - if (item.type === 'text') { - return { type: 'text', text: item.text }; - } - return { type: 'local_image', path: item.path }; - }); -} - -export function reduceCodexSdkTurnState( - state: CodexSdkTurnState, - event: ThreadEvent | unknown, -): CodexSdkTurnState { - if (!event || typeof event !== 'object' || !('type' in event)) { - return state; - } - - const typed = event as ThreadEvent; - if (typed.type === 'thread.started') { - return { ...state, threadId: typed.thread_id }; - } - - if (typed.type === 'turn.started') { - return { ...state, status: 'inProgress' }; - } - - if (typed.type === 'item.completed') { - if (typed.item.type === 'agent_message') { - const text = typed.item.text.trim(); - if (!text) return state; - return { - ...state, - finalAnswer: text, - latestAgentMessage: text, - }; - } - if (typed.item.type === 'error') { - return { - ...state, - errorMessage: typed.item.message, - }; - } - return state; - } - - if (typed.type === 'turn.completed') { - return { - ...state, - status: state.status === 'failed' ? 'failed' : 'completed', - usage: typed.usage, - }; - } - - if (typed.type === 'turn.failed') { - return { - ...state, - status: 'failed', - errorMessage: typed.error.message, - }; - } - - if (typed.type === 'error') { - return { - ...state, - status: 'failed', - errorMessage: typed.message, - }; - } - - return state; -} - -function isAbortError(error: unknown): boolean { - return ( - error instanceof Error && - (error.name === 'AbortError' || - error.message.toLowerCase().includes('abort')) - ); -} - -export class CodexSdkClient { - private readonly codex: CodexSdkPort; - private readonly log: (message: string) => void; - private readonly threads = new Map(); - private nextPendingThreadId = 1; - - constructor(options: CodexSdkClientOptions) { - this.log = options.log; - this.codex = - options.codex ?? - new Codex({ - codexPathOverride: options.codexPathOverride, - config: options.config, - env: options.env, - }); - } - - async startOrResumeThread( - sessionId: string | undefined, - options: CodexSdkThreadStartOptions, - ): Promise { - const threadOptions = buildThreadOptions(options); - const thread = sessionId - ? this.codex.resumeThread(sessionId, threadOptions) - : this.codex.startThread(threadOptions); - const handle = - sessionId ?? `pending-sdk-thread-${this.nextPendingThreadId++}`; - this.threads.set(handle, thread); - return handle; - } - - async startTurn( - threadHandle: string, - input: AppServerInputItem[], - options: CodexSdkTurnOptions, - ): Promise<{ - turnId: string; - steer: (nextInput: AppServerInputItem[]) => Promise; - interrupt: () => Promise; - wait: () => Promise; - }> { - const thread = this.threads.get(threadHandle); - if (!thread) { - throw new Error(`Unknown Codex SDK thread handle: ${threadHandle}`); - } - - const abortController = new AbortController(); - const waitPromise = this.consumeTurn( - threadHandle, - thread, - input, - options, - abortController.signal, - ); - - return { - turnId: threadHandle, - steer: async () => { - throw new Error( - 'Codex SDK runner does not support mid-turn steering; queue the message for the next turn.', - ); - }, - interrupt: async () => { - abortController.abort(); - }, - wait: async () => waitPromise, - }; - } - - private async consumeTurn( - threadHandle: string, - thread: CodexSdkThreadPort, - input: AppServerInputItem[], - options: CodexSdkTurnOptions, - signal: AbortSignal, - ): Promise { - let state = createInitialCodexSdkTurnState(); - try { - const { events } = await thread.runStreamed(toCodexSdkInput(input), { - signal, - }); - for await (const event of events) { - state = reduceCodexSdkTurnState(state, event); - if (state.threadId && state.threadId !== threadHandle) { - this.threads.set(state.threadId, thread); - this.threads.delete(threadHandle); - } - if ( - state.latestAgentMessage && - state.latestAgentMessage !== state.finalAnswer - ) { - options.onProgress?.(state.latestAgentMessage); - } - } - } catch (error) { - if (isAbortError(error)) { - state = { ...state, status: 'interrupted' }; - } else { - const message = error instanceof Error ? error.message : String(error); - this.log(`Codex SDK turn failed: ${message}`); - state = { ...state, status: 'failed', errorMessage: message }; - } - } - - const threadId = state.threadId ?? thread.id; - if (threadId && threadId !== threadHandle) { - this.threads.set(threadId, thread); - this.threads.delete(threadHandle); - } - - return { - state: { ...state, threadId }, - result: state.finalAnswer, - threadId, - }; - } -} diff --git a/runners/codex-runner/src/index.ts b/runners/codex-runner/src/index.ts index 14b5680..0db0b5e 100644 --- a/runners/codex-runner/src/index.ts +++ b/runners/codex-runner/src/index.ts @@ -1,10 +1,7 @@ /** * EJClaw Codex Runner * - * App-server by default, with optional CODEX_RUNTIME=sdk lane. - * SDK mode uses @openai/codex-sdk / codex exec JSON events and leaves - * mid-turn IPC follow-ups queued for the next run because SDK steering is - * not currently available. + * App-server only runtime. * * Input protocol: * Stdin: Full RunnerInput JSON (read until EOF) @@ -30,8 +27,6 @@ import { import { CodexAppServerClient } from './app-server-client.js'; import { parseAppServerInput } from './app-server-input.js'; -import { CodexSdkClient } from './codex-sdk-client.js'; -import { resolveCodexRuntimeMode } from './runtime-mode.js'; import { prependRoomRoleHeader, type RoomRoleContext, @@ -109,15 +104,6 @@ function log(message: string): void { console.error(`[codex-runner] ${message}`); } -function toStringEnv(env: NodeJS.ProcessEnv): Record { - return Object.fromEntries( - Object.entries(env).filter((entry): entry is [string, string] => { - const [, value] = entry; - return typeof value === 'string'; - }), - ); -} - function isCodexGoalsEnabled(runnerInput: RunnerInput): boolean { return runnerInput.codexGoals === true || process.env.CODEX_GOALS === 'true'; } @@ -297,117 +283,6 @@ async function executeAppServerTurn( } } -function countQueuedIpcInputFiles(): number { - try { - fs.mkdirSync(IPC_INPUT_DIR, { recursive: true }); - return fs - .readdirSync(IPC_INPUT_DIR) - .filter((file) => file.endsWith('.json')).length; - } catch (err) { - log( - `IPC queued input count error: ${err instanceof Error ? err.message : String(err)}`, - ); - return 0; - } -} - -async function executeSdkTurn( - client: CodexSdkClient, - threadHandle: string, - prompt: string, - retryCount = 0, -): Promise<{ - result: string | null; - error?: string; - threadId: string; -}> { - let lastProgressMessage: string | null = null; - const activeTurn = await client.startTurn( - threadHandle, - parseAppServerInput(prompt, log), - { - cwd: EFFECTIVE_CWD, - model: CODEX_MODEL || undefined, - effort: CODEX_EFFORT || undefined, - onProgress: (message) => { - const trimmed = message.trim(); - if (!trimmed || trimmed === lastProgressMessage) { - return; - } - lastProgressMessage = trimmed; - writeOutput({ - status: 'success', - phase: 'progress', - ...normalizeStructuredOutput(trimmed), - newSessionId: threadHandle, - }); - }, - }, - ); - - let elapsedMs = 0; - let polling = true; - let lastQueuedInputCount = 0; - const pollDuringTurn = async () => { - if (!polling) return; - - if (consumeCloseSentinel()) { - log('Close sentinel detected during SDK turn, interrupting'); - polling = false; - try { - await activeTurn.interrupt(); - } catch (err) { - log( - `Failed to interrupt active SDK turn: ${ - err instanceof Error ? err.message : String(err) - }`, - ); - } - return; - } - - const queuedInputCount = countQueuedIpcInputFiles(); - if (queuedInputCount > 0 && queuedInputCount !== lastQueuedInputCount) { - lastQueuedInputCount = queuedInputCount; - log( - `SDK runtime cannot steer active turns; leaving ${queuedInputCount} queued input file(s) for the next run`, - ); - } - - elapsedMs += IPC_POLL_MS; - if (elapsedMs > 0 && elapsedMs % 60000 === 0) { - log(`SDK turn in progress... (${formatProgressElapsed(elapsedMs)})`); - } - setTimeout(() => void pollDuringTurn(), IPC_POLL_MS); - }; - - setTimeout(() => void pollDuringTurn(), IPC_POLL_MS); - - try { - const { state, result, threadId } = await activeTurn.wait(); - const finalThreadId = threadId ?? threadHandle; - if (state.status === 'completed') { - return { result, threadId: finalThreadId }; - } - if (state.status === 'interrupted' && consumeCloseSentinel()) { - return { result, threadId: finalThreadId }; - } - if (state.status === 'interrupted' && retryCount < 1) { - log('Codex SDK turn interrupted, retrying once...'); - return executeSdkTurn(client, finalThreadId, prompt, retryCount + 1); - } - return { - result, - error: - state.errorMessage || - `Codex SDK turn finished with status ${state.status}`, - threadId: finalThreadId, - }; - } finally { - polling = false; - } -} - async function runAppServerCompact( client: CodexAppServerClient, threadId: string | undefined, @@ -530,64 +405,6 @@ async function runAppServerSession( } } -async function runSdkSession( - runnerInput: RunnerInput, - prompt: string, -): Promise { - const reviewerRuntime = - isReviewerRuntimeEnvEnabled(process.env) || - isReviewerRuntime(runnerInput.roomRoleContext); - const readonlyRuntime = - reviewerRuntime || isArbiterRuntimeEnvEnabled(process.env); - const clientEnv = buildReviewerGitGuardEnv(process.env, reviewerRuntime); - assertReadonlyWorkspaceRepoConnectivity(clientEnv, readonlyRuntime); - const client = new CodexSdkClient({ - env: toStringEnv(clientEnv), - log, - }); - - const threadHandle = await client.startOrResumeThread(runnerInput.sessionId, { - cwd: EFFECTIVE_CWD, - model: CODEX_MODEL || undefined, - effort: CODEX_EFFORT || undefined, - }); - log( - runnerInput.sessionId - ? `SDK thread resumed (${threadHandle})` - : `SDK thread started (${threadHandle})`, - ); - - log('Starting SDK turn...'); - const { result, error, threadId } = await executeSdkTurn( - client, - threadHandle, - prompt, - ); - - if (error) { - const normalized = normalizeStructuredOutput(result || null); - log(`SDK turn error: ${error}`); - writeOutput({ - status: 'error', - ...normalized, - newSessionId: threadId, - error, - }); - } else { - const normalized = normalizeStructuredOutput(result || null); - writeOutput({ - status: 'success', - ...normalized, - ...(result ? { phase: 'final' as const } : {}), - newSessionId: threadId, - }); - } - - if (consumeCloseSentinel()) { - log('Close sentinel detected, exiting SDK runtime'); - } -} - // ── Main ────────────────────────────────────────────────────────── async function main(): Promise { @@ -638,20 +455,8 @@ async function main(): Promise { } try { - const runtimeMode = resolveCodexRuntimeMode( - process.env, - { - codexGoals: isCodexGoalsEnabled(runnerInput), - roomRole: runnerInput.roomRoleContext?.role, - }, - rawPrompt, - ); - log(`Runtime selected: ${runtimeMode.mode} (${runtimeMode.reason})`); - if (runtimeMode.mode === 'sdk') { - await runSdkSession(runnerInput, prompt); - } else { - await runAppServerSession(runnerInput, prompt); - } + log('Runtime selected: app-server'); + await runAppServerSession(runnerInput, prompt); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); log(`Runner error: ${errorMessage}`); diff --git a/runners/codex-runner/src/runtime-mode.ts b/runners/codex-runner/src/runtime-mode.ts deleted file mode 100644 index f675ed8..0000000 --- a/runners/codex-runner/src/runtime-mode.ts +++ /dev/null @@ -1,85 +0,0 @@ -export type CodexRuntimeMode = 'app-server' | 'sdk'; - -export interface CodexRuntimeModeInput { - codexGoals?: boolean; - roomRole?: string; -} - -export interface CodexRuntimeModeResolution { - mode: CodexRuntimeMode; - reason: - | 'default' - | 'CODEX_RUNTIME=app-server' - | 'CODEX_RUNTIME=sdk' - | 'sdk-unsupported-goals' - | 'sdk-unsupported-session-command' - | 'sdk-role-not-enabled' - | 'invalid-CODEX_RUNTIME'; -} - -function normalizeRuntimeMode( - raw: string | undefined, -): CodexRuntimeMode | null { - if (!raw) return null; - const normalized = raw.trim().toLowerCase(); - if (normalized === 'app-server' || normalized === 'appserver') { - return 'app-server'; - } - if (normalized === 'sdk') { - return 'sdk'; - } - return null; -} - -export interface CodexRuntimeEnv { - [key: string]: string | undefined; - CODEX_RUNTIME?: string; - CODEX_RUNTIME_SDK_ROLES?: string; -} - -function sdkRoleEnabled( - env: CodexRuntimeEnv, - roomRole: string | undefined, -): boolean { - const raw = env.CODEX_RUNTIME_SDK_ROLES?.trim(); - if (!raw) return true; - if (!roomRole) return false; - const allowed = raw - .split(',') - .map((role) => role.trim().toLowerCase()) - .filter(Boolean); - return allowed.includes(roomRole.trim().toLowerCase()); -} - -export function resolveCodexRuntimeMode( - env: CodexRuntimeEnv, - input: CodexRuntimeModeInput, - rawPrompt: string, -): CodexRuntimeModeResolution { - const requested = normalizeRuntimeMode(env.CODEX_RUNTIME); - if (requested === 'sdk') { - if (rawPrompt.trim() === '/compact') { - return { - mode: 'app-server', - reason: 'sdk-unsupported-session-command', - }; - } - if (input.codexGoals === true) { - return { mode: 'app-server', reason: 'sdk-unsupported-goals' }; - } - if (!sdkRoleEnabled(env, input.roomRole)) { - return { mode: 'app-server', reason: 'sdk-role-not-enabled' }; - } - return { mode: 'sdk', reason: 'CODEX_RUNTIME=sdk' }; - } - - if (requested === 'app-server') { - return { mode: 'app-server', reason: 'CODEX_RUNTIME=app-server' }; - } - - if (env.CODEX_RUNTIME && env.CODEX_RUNTIME.trim()) { - return { mode: 'app-server', reason: 'invalid-CODEX_RUNTIME' }; - } - - return { mode: 'app-server', reason: 'default' }; -} diff --git a/runners/codex-runner/test/runtime-mode.test.ts b/runners/codex-runner/test/runtime-mode.test.ts deleted file mode 100644 index 40e4245..0000000 --- a/runners/codex-runner/test/runtime-mode.test.ts +++ /dev/null @@ -1,64 +0,0 @@ -import { describe, expect, it } from 'vitest'; - -import { resolveCodexRuntimeMode } from '../src/runtime-mode.js'; - -describe('codex runner runtime mode selection', () => { - it('keeps app-server as the default runtime', () => { - expect( - resolveCodexRuntimeMode({}, { codexGoals: false }, 'do work'), - ).toEqual({ - mode: 'app-server', - reason: 'default', - }); - }); - - it('selects SDK mode when CODEX_RUNTIME=sdk', () => { - expect( - resolveCodexRuntimeMode( - { CODEX_RUNTIME: 'sdk' }, - { codexGoals: false }, - 'do work', - ), - ).toEqual({ mode: 'sdk', reason: 'CODEX_RUNTIME=sdk' }); - }); - - it('canary-limits SDK mode to configured paired roles', () => { - expect( - resolveCodexRuntimeMode( - { CODEX_RUNTIME: 'sdk', CODEX_RUNTIME_SDK_ROLES: 'owner,arbiter' }, - { codexGoals: false, roomRole: 'owner' }, - 'do work', - ), - ).toEqual({ mode: 'sdk', reason: 'CODEX_RUNTIME=sdk' }); - expect( - resolveCodexRuntimeMode( - { CODEX_RUNTIME: 'sdk', CODEX_RUNTIME_SDK_ROLES: 'owner,arbiter' }, - { codexGoals: false, roomRole: 'reviewer' }, - 'do work', - ), - ).toEqual({ mode: 'app-server', reason: 'sdk-role-not-enabled' }); - }); - - it('falls back to app-server for /compact because SDK lacks compaction', () => { - expect( - resolveCodexRuntimeMode( - { CODEX_RUNTIME: 'sdk' }, - { codexGoals: false }, - '/compact', - ), - ).toEqual({ - mode: 'app-server', - reason: 'sdk-unsupported-session-command', - }); - }); - - it('falls back to app-server when Codex goals are enabled', () => { - expect( - resolveCodexRuntimeMode( - { CODEX_RUNTIME: 'sdk' }, - { codexGoals: true }, - 'do work', - ), - ).toEqual({ mode: 'app-server', reason: 'sdk-unsupported-goals' }); - }); -}); diff --git a/runners/codex-runner/test/sdk-client.test.ts b/runners/codex-runner/test/sdk-client.test.ts deleted file mode 100644 index 243a509..0000000 --- a/runners/codex-runner/test/sdk-client.test.ts +++ /dev/null @@ -1,194 +0,0 @@ -import { describe, expect, it } from 'vitest'; - -import { - CodexSdkClient, - createInitialCodexSdkTurnState, - reduceCodexSdkTurnState, - toCodexSdkInput, - type CodexSdkThreadPort, -} from '../src/codex-sdk-client.js'; - -const usage = { - input_tokens: 1, - cached_input_tokens: 0, - output_tokens: 2, - reasoning_output_tokens: 0, -}; - -async function* events(items: unknown[]): AsyncGenerator { - for (const item of items) { - yield item; - } -} - -class FakeThread implements CodexSdkThreadPort { - id: string | null = null; - lastInput: unknown = null; - lastSignal: AbortSignal | undefined; - - constructor(private readonly scriptedEvents: unknown[]) {} - - async runStreamed(input: unknown, options?: { signal?: AbortSignal }) { - this.lastInput = input; - this.lastSignal = options?.signal; - return { events: events(this.scriptedEvents) }; - } -} - -class FakeCodex { - readonly started: unknown[] = []; - readonly resumed: unknown[] = []; - readonly threads: FakeThread[] = []; - - constructor(private readonly scriptedEvents: unknown[]) {} - - startThread(options?: unknown): FakeThread { - this.started.push(options); - const thread = new FakeThread(this.scriptedEvents); - this.threads.push(thread); - return thread; - } - - resumeThread(id: string, options?: unknown): FakeThread { - this.resumed.push({ id, options }); - const thread = new FakeThread(this.scriptedEvents); - thread.id = id; - this.threads.push(thread); - return thread; - } -} - -describe('Codex SDK runner spike', () => { - it('maps existing app-server text/image input into Codex SDK input shape', () => { - expect( - toCodexSdkInput([ - { type: 'text', text: 'inspect this' }, - { type: 'localImage', path: '/tmp/screenshot.png' }, - ]), - ).toEqual([ - { type: 'text', text: 'inspect this' }, - { type: 'local_image', path: '/tmp/screenshot.png' }, - ]); - }); - - it('reduces SDK thread events into the runner final result and thread id', () => { - let state = createInitialCodexSdkTurnState(); - state = reduceCodexSdkTurnState(state, { - type: 'thread.started', - thread_id: 'thread-1', - }); - state = reduceCodexSdkTurnState(state, { type: 'turn.started' }); - state = reduceCodexSdkTurnState(state, { - type: 'item.completed', - item: { - id: 'msg-1', - type: 'agent_message', - text: 'STEP_DONE\n검증 완료', - }, - }); - state = reduceCodexSdkTurnState(state, { - type: 'turn.completed', - usage, - }); - - expect(state).toMatchObject({ - status: 'completed', - threadId: 'thread-1', - finalAnswer: 'STEP_DONE\n검증 완료', - usage, - }); - }); - - it('runs a streamed SDK turn and returns the first real thread id', async () => { - const codex = new FakeCodex([ - { type: 'thread.started', thread_id: 'thread-42' }, - { type: 'turn.started' }, - { - type: 'item.completed', - item: { id: 'msg-1', type: 'agent_message', text: 'TASK_DONE\nSDK_OK' }, - }, - { type: 'turn.completed', usage }, - ]); - const client = new CodexSdkClient({ - codex, - env: { PATH: '/usr/bin' }, - log: () => undefined, - }); - - const handle = await client.startOrResumeThread(undefined, { - cwd: '/repo', - model: 'gpt-5.5', - effort: 'high', - }); - const turn = await client.startTurn( - handle, - [{ type: 'text', text: 'respond' }], - { cwd: '/repo', model: 'gpt-5.5', effort: 'high' }, - ); - - await expect(turn.wait()).resolves.toMatchObject({ - threadId: 'thread-42', - result: 'TASK_DONE\nSDK_OK', - state: { status: 'completed' }, - }); - expect(codex.started).toEqual([ - { - approvalPolicy: 'never', - model: 'gpt-5.5', - modelReasoningEffort: 'high', - networkAccessEnabled: true, - sandboxMode: 'danger-full-access', - workingDirectory: '/repo', - }, - ]); - expect(codex.threads[0].lastInput).toEqual([ - { type: 'text', text: 'respond' }, - ]); - }); - - it('coerces minimal effort to low because Codex SDK exec keeps tools enabled', async () => { - const codex = new FakeCodex([ - { type: 'thread.started', thread_id: 'thread-minimal' }, - { type: 'turn.started' }, - { - type: 'item.completed', - item: { id: 'msg-1', type: 'agent_message', text: 'SDK_OK' }, - }, - { type: 'turn.completed', usage }, - ]); - const client = new CodexSdkClient({ codex, log: () => undefined }); - - await client.startOrResumeThread(undefined, { - cwd: '/repo', - effort: 'minimal', - }); - - expect(codex.started[0]).toMatchObject({ modelReasoningEffort: 'low' }); - }); - - it('documents the SDK gap: mid-turn steering is not available', async () => { - const codex = new FakeCodex([ - { type: 'thread.started', thread_id: 'thread-1' }, - { type: 'turn.started' }, - { type: 'turn.completed', usage }, - ]); - const client = new CodexSdkClient({ codex, log: () => undefined }); - const handle = await client.startOrResumeThread('thread-1', { - cwd: '/repo', - }); - const turn = await client.startTurn( - handle, - [{ type: 'text', text: 'start' }], - { - cwd: '/repo', - }, - ); - - await expect( - turn.steer([{ type: 'text', text: 'follow-up' }]), - ).rejects.toThrow('Codex SDK runner does not support mid-turn steering'); - await expect(turn.wait()).resolves.toMatchObject({ - state: { status: 'completed' }, - }); - }); -}); diff --git a/spikes/001-codex-sdk-runner/README.md b/spikes/001-codex-sdk-runner/README.md deleted file mode 100644 index ec5767c..0000000 --- a/spikes/001-codex-sdk-runner/README.md +++ /dev/null @@ -1,81 +0,0 @@ -# 001: codex-sdk-runner - -## Question - -Given EJClaw currently wraps `codex app-server` JSON-RPC directly, when we switch the runner core to the official `@openai/codex-sdk`, can we preserve the key runner guarantees: thread resume, final result extraction, cancellation, and structured event visibility? - -## Approach - -- Added `@openai/codex-sdk@0.136.0` beside `@openai/codex@0.136.0` in the codex runner workspace. -- Built a `CodexSdkClient` adapter that maps EJClaw's existing app-server input shape into SDK input. -- Added `CODEX_RUNTIME=sdk` runtime selection while keeping app-server as the default. -- Added `CODEX_RUNTIME_SDK_ROLES=owner,arbiter` canary limiting so production can try SDK only on selected paired roles. -- Fallbacks: `/compact` and `codexGoals` stay on app-server because SDK lacks those APIs. -- Wrote unit tests around SDK event reduction, thread-id propagation, runtime selection, and unsupported steering. -- Ran real smoke turns through both the SDK adapter and the integrated codex-runner entrypoint against temporary git repos. - -## Evidence - -Commands run: - -```bash -bunx vitest run runners/codex-runner/test/sdk-client.test.ts -bunx vitest run runners/codex-runner/test/runtime-mode.test.ts -bunx tsc --noEmit -p runners/codex-runner/tsconfig.json -bun run --cwd runners/codex-runner build -bunx vitest run \ - runners/codex-runner/test/app-server-client.test.ts \ - runners/codex-runner/test/app-server-state.test.ts \ - runners/codex-runner/test/runtime-mode.test.ts \ - runners/codex-runner/test/sdk-client.test.ts -bun run check -CODEX_RUNTIME=sdk CODEX_EFFORT=minimal node runners/codex-runner/dist/index.js < /tmp/input.json -``` - -Observed adapter smoke result: - -```json -{ - "status": "completed", - "result": "SDK_OK", - "error": null -} -``` - -Observed integrated runner smoke result: - -```json -{ - "status": "success", - "result": "SDK_RUNNER_OK", - "phase": "final", - "newSessionId": "019e..." -} -``` - -## What worked - -- SDK import and runner workspace build work on Node/Bun in this repo. -- SDK streaming events can be reduced into EJClaw-style `status`, `threadId`, `result`, `error`, and `usage`. -- `thread.started` gives the real thread id during the first streamed turn, so a pending local handle can be replaced with the real session id. -- `AbortController` gives a clean replacement for close-sentinel cancellation. -- Real SDK smoke and integrated runner smoke completed successfully. -- Runtime flag selection works: default app-server; `CODEX_RUNTIME=sdk` selects SDK; `CODEX_RUNTIME_SDK_ROLES` canaries SDK to chosen paired roles; `/compact` and goals fall back to app-server. - -## What did not work / gaps - -- SDK does **not** expose app-server `turn/steer`; mid-turn Discord follow-up steering would need to be queued for the next turn or stay on the app-server path. -- SDK TypeScript wrapper uses `codex exec --experimental-json`, not persistent `codex app-server`, so process-per-turn overhead and session behavior must be measured under real long tasks. -- Passing `modelReasoningEffort: "minimal"` failed in live smoke because Codex exec still enabled tools that are incompatible with minimal reasoning. The spike adapter coerces `minimal` to `low`. -- Existing Codex app-server-only features (`thread/goal/*`, `thread/compact/start`) are not available through the current SDK API. - -## Verdict: FEATURE-FLAGGED - -SDK is now viable as an optional **exec-backed Codex runner lane** behind `CODEX_RUNTIME=sdk`. The default remains app-server, and `CODEX_RUNTIME_SDK_ROLES=owner,arbiter` can limit SDK to safer paired roles while reviewer/non-paired turns keep the old path. - -## Recommended rollout - -1. Merge as a draft/feature-flagged implementation without changing production env. -2. Enable `CODEX_RUNTIME=sdk` for one low-risk owner/arbiter lane first. -3. Keep app-server mode for rooms/tasks that require live steering, goals, or compaction. -4. Watch `turn.failed`, `error`, `item.completed:error`, empty final output, and process-per-turn latency before promoting SDK broadly. diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index b699d0f..9ab63c3 100644 --- a/src/agent-runner.test.ts +++ b/src/agent-runner.test.ts @@ -852,53 +852,4 @@ describe('agent-runner Codex goals', () => { | undefined; expect(spawnEnv?.CODEX_GOALS).toBe('true'); }); - - it('marks SDK codex processes so queued IPC follow-ups drain after the active run', async () => { - const previousCodexGoals = process.env.CODEX_GOALS; - delete process.env.CODEX_GOALS; - const onProcess = vi.fn(); - const codexGroup: RegisteredGroup = { - ...testGroup, - agentType: 'codex', - }; - - try { - const resultPromise = runAgentProcess( - codexGroup, - { - ...testInput, - roomRoleContext: { - serviceId: 'codex-main', - role: 'owner', - ownerServiceId: 'codex-main', - reviewerServiceId: 'claude', - failoverOwner: false, - }, - }, - onProcess, - async () => {}, - { - CODEX_RUNTIME: 'sdk', - CODEX_RUNTIME_SDK_ROLES: 'owner', - }, - ); - - fakeProc.emit('close', 0); - const result = await resultPromise; - expect(result.status).toBe('success'); - - expect(onProcess).toHaveBeenCalledWith( - fakeProc, - expect.any(String), - expect.any(String), - { drainFollowUpsAfterRun: true }, - ); - } finally { - if (previousCodexGoals === undefined) { - delete process.env.CODEX_GOALS; - } else { - process.env.CODEX_GOALS = previousCodexGoals; - } - } - }); }); diff --git a/src/agent-runner.ts b/src/agent-runner.ts index 1ecd1c5..3b7a7dd 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -61,30 +61,6 @@ export interface AgentOutput { }; } -interface ActiveProcessOptions { - drainFollowUpsAfterRun?: boolean; -} - -function isCodexSdkRuntimeMode( - env: Record, - input: AgentInput, -): boolean { - const runtime = env.CODEX_RUNTIME?.trim().toLowerCase(); - if (runtime !== 'sdk') return false; - if (input.prompt.trim() === '/compact') return false; - if (input.codexGoals === true || env.CODEX_GOALS === 'true') return false; - - const rawRoles = env.CODEX_RUNTIME_SDK_ROLES?.trim(); - if (!rawRoles) return true; - const role = input.roomRoleContext?.role?.trim().toLowerCase(); - if (!role) return false; - return rawRoles - .split(',') - .map((item) => item.trim().toLowerCase()) - .filter(Boolean) - .includes(role); -} - function readRoomSkillOverridesForRunner( chatJid: string, ): StoredRoomSkillOverride[] { @@ -150,7 +126,6 @@ export async function runAgentProcess( proc: ChildProcess, processName: string, runtimeIpcDir: string, - options?: ActiveProcessOptions, ) => void, onOutput?: (output: AgentOutput) => Promise, envOverrides?: Record, @@ -211,11 +186,6 @@ export async function runAgentProcess( const safeName = group.folder.replace(/[^a-zA-Z0-9-]/g, '-'); const processSuffix = input.runId || `${Date.now()}`; const processName = `ejclaw-${safeName}-${processSuffix}`; - const activeProcessOptions: ActiveProcessOptions = { - drainFollowUpsAfterRun: - (group.agentType || 'claude-code') === 'codex' && - isCodexSdkRuntimeMode(env, input), - }; const finalizeCodexAuthSessionOnce = createCodexAuthSessionFinalizer( () => codexSessionAuth, ); @@ -258,7 +228,7 @@ export async function runAgentProcess( env, }); - onProcess(proc, processName, env[EJCLAW_ENV.ipcDir], activeProcessOptions); + onProcess(proc, processName, env[EJCLAW_ENV.ipcDir]); const runnerInput: AgentInput = { ...input, diff --git a/src/group-queue-state.ts b/src/group-queue-state.ts index 12e6f03..8719b6f 100644 --- a/src/group-queue-state.ts +++ b/src/group-queue-state.ts @@ -37,7 +37,6 @@ export interface GroupState { pendingTasks: QueuedTask[]; process: ChildProcess | null; processName: string | null; - drainFollowUpsAfterRun: boolean; ipcDir: string | null; retryCount: number; retryTimer: ReturnType | null; @@ -61,7 +60,6 @@ export function createGroupState(): GroupState { pendingTasks: [], process: null, processName: null, - drainFollowUpsAfterRun: false, ipcDir: null, retryCount: 0, retryTimer: null, @@ -147,7 +145,6 @@ export function resetRunState(state: GroupState, groupJid: string): void { state.startedAt = null; state.process = null; state.processName = null; - state.drainFollowUpsAfterRun = false; state.ipcDir = null; state.directTerminalDeliveries.clear(); transitionRunPhase(state, groupJid, 'idle'); @@ -163,8 +160,7 @@ export function assertRunPhaseInvariants( state.currentRunId != null || state.runningTaskId != null || state.process != null || - state.processName != null || - state.drainFollowUpsAfterRun + state.processName != null ) { logger.error( { @@ -174,7 +170,6 @@ export function assertRunPhaseInvariants( runningTaskId: state.runningTaskId, hasProcess: state.process != null, processName: state.processName, - drainFollowUpsAfterRun: state.drainFollowUpsAfterRun, }, 'Invariant violation: idle phase has stale run/task ID or process', ); diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index 76b2a60..d2991d1 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -40,6 +40,8 @@ describe('GroupQueue', () => { vi.useRealTimers(); }); + // --- Single group at a time --- + it('only runs one agent per group at a time', async () => { let concurrentCount = 0; let maxConcurrent = 0; @@ -88,7 +90,6 @@ describe('GroupQueue', () => { releaseRun(true); await vi.advanceTimersByTimeAsync(10); - expect(processMessages).toHaveBeenCalledTimes(1); }); it('does not pipe follow-up messages after stdin close was requested', async () => { @@ -790,46 +791,3 @@ describe('GroupQueue close reason tracking', () => { expect(queue.getCloseReasonForRun('group1@g.us', activeRunId)).toBeNull(); }); }); - -describe('GroupQueue SDK follow-up draining', () => { - let queue: GroupQueue; - - beforeEach(() => { - vi.useFakeTimers(); - vi.clearAllMocks(); - queue = new GroupQueue(); - }); - - afterEach(() => { - vi.useRealTimers(); - }); - - it('drains follow-up messages after active processes that cannot consume queued IPC', async () => { - const ipcDir = '/tmp/ejclaw-test-data/ipc/group-folder'; - let releaseRun!: (value: boolean) => void; - let activeRun = true; - const blocker = new Promise((resolve) => { - releaseRun = resolve; - }); - - const processMessages = vi.fn(async () => { - if (!activeRun) return true; - activeRun = false; - return await blocker; - }); - - queue.setProcessMessagesFn(processMessages); - queue.enqueueMessageCheck('group1@g.us', ipcDir); - await vi.advanceTimersByTimeAsync(10); - - queue.registerProcess('group1@g.us', {} as any, 'sdk-agent', ipcDir, { - drainFollowUpsAfterRun: true, - }); - expect(queue.sendMessage('group1@g.us', '후속 메시지')).toBe(true); - - releaseRun(true); - await vi.advanceTimersByTimeAsync(10); - - expect(processMessages).toHaveBeenCalledTimes(2); - }); -}); diff --git a/src/group-queue.ts b/src/group-queue.ts index 3d141f0..c92d576 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -65,6 +65,7 @@ export class GroupQueue { this.processMessagesFn = fn; } + /** Limit concurrency after restart to avoid API rate-limit storms. */ enterRecoveryMode(): void { this.recoveryMode = true; logger.info( @@ -100,6 +101,7 @@ export class GroupQueue { const state = this.getGroup(groupJid); + // Pre-set IPC dir so sendMessage can pipe follow-ups while agent process starts if (ipcDir && !state.ipcDir) { state.ipcDir = ipcDir; } @@ -155,6 +157,7 @@ export class GroupQueue { const state = this.getGroup(groupJid); + // Prevent double-queuing: check both pending and currently-running task if (state.runningTaskId === taskId) { logger.debug({ groupJid, taskId }, 'Task already running, skipping'); return; @@ -193,6 +196,7 @@ export class GroupQueue { return; } + // Run immediately this.runTask(groupJid, { id: taskId, groupJid, fn }).catch((err) => logger.error({ groupJid, taskId, err }, 'Unhandled error in runTask'), ); @@ -203,12 +207,10 @@ export class GroupQueue { proc: ChildProcess, processName: string, ipcDir?: string, - options?: { drainFollowUpsAfterRun?: boolean }, ): void { const state = this.getGroup(groupJid); state.process = proc; state.processName = processName; - state.drainFollowUpsAfterRun = options?.drainFollowUpsAfterRun === true; if (ipcDir) state.ipcDir = ipcDir; logger.info( { @@ -239,7 +241,6 @@ export class GroupQueue { try { const filename = queueFollowUpMessage(state.ipcDir, text); - state.pendingMessages ||= state.drainFollowUpsAfterRun; logger.info( { groupJid, @@ -247,7 +248,6 @@ export class GroupQueue { ipcDir: state.ipcDir, textLength: text.length, filename, - pendingMessages: state.pendingMessages, }, 'Queued follow-up message for active agent', ); diff --git a/src/message-agent-executor-attempt-runner.ts b/src/message-agent-executor-attempt-runner.ts index 2b07de4..64a9126 100644 --- a/src/message-agent-executor-attempt-runner.ts +++ b/src/message-agent-executor-attempt-runner.ts @@ -100,7 +100,6 @@ interface RunMessageAgentAttemptArgs { : never, processName: string, ipcDir?: string, - options?: { drainFollowUpsAfterRun?: boolean }, ) => void; onOutput?: (output: AgentOutput) => Promise; pairedExecutionLifecycle: { diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 846e1ce..6b85cdb 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -274,8 +274,8 @@ export async function runAgentForGroup( deps.persistSession(sessionFolder, sessionId); currentSessionId = sessionId; }, - registerProcess: (proc, processName, ipcDir, options) => - deps.queue.registerProcess(chatJid, proc, processName, ipcDir, options), + registerProcess: (proc, processName, ipcDir) => + deps.queue.registerProcess(chatJid, proc, processName, ipcDir), onOutput, pairedExecutionLifecycle, log,