Merge pull request #2 from JetBrains/st

Windows fixes and error handling

Author: slapoguzov

readme-dev.md

@@ -4,6 +4,10 @@
 
 Last checked version: `codex-cli 0.64.0-alpha.9`)
 
+###### Develop on Windows?
+- Download and install [bum](https://bun.com/docs/installation#windows)
+- Download and install [C++ redistributable package](https://learn.microsoft.com/en-us/cpp/windows/latest-supported-vc-redist?view=msvc-170#latest-supported-redistributable-version)
+
 #### Adjust acp config for IDE
 
 Run from sources

src/CodexAcpServer.ts

@@ -17,24 +17,27 @@ export class CodexAcpServer implements acp.Agent {
     private readonly codexAcpClient: CodexAcpClient;
     private readonly connection: acp.AgentSideConnection;
     private readonly defaultAuthRequest: CodexAuthRequest | null;
+    private readonly getExitCode: () => number | null;
 
     private readonly sessions: Map<string, SessionState>;
 
     constructor(
         connection: acp.AgentSideConnection,
         codexAcpClient: CodexAcpClient,
         defaultAuthRequest?: CodexAuthRequest,
+        getExitCode?: () => number | null,
     ) {
         this.sessions = new Map();
         this.connection = connection;
-        this.defaultAuthRequest = defaultAuthRequest ?? null;
         this.codexAcpClient = codexAcpClient;
+        this.defaultAuthRequest = defaultAuthRequest ?? null;
+        this.getExitCode = getExitCode ?? (() => null);
     }
 
     async initialize(
         _params: acp.InitializeRequest,
     ): Promise<acp.InitializeResponse> {
-        await this.codexAcpClient.initialize(_params);
+        await this.runWithProcessCheck(() => this.codexAcpClient.initialize(_params));
         return {
             protocolVersion: acp.PROTOCOL_VERSION,
             agentCapabilities: {
@@ -47,15 +50,15 @@ export class CodexAcpServer implements acp.Agent {
     async newSession(
         _params: acp.NewSessionRequest,
     ): Promise<acp.NewSessionResponse> {
-        if (await this.codexAcpClient.authRequired()) {
+        if (await this.runWithProcessCheck(() => this.codexAcpClient.authRequired())) {
             if (this.defaultAuthRequest) {
                 await this.authenticate(this.defaultAuthRequest)
             } else {
                 throw RequestError.authRequired();
             }
         }
 
-        const sessionMetadata = await this.codexAcpClient.newSession(_params);
+        const sessionMetadata = await this.runWithProcessCheck(() => this.codexAcpClient.newSession(_params));
         const {sessionId, currentModelId, models} = sessionMetadata;
         this.sessions.set(sessionId, {
             sessionMetadata: sessionMetadata,
@@ -76,7 +79,7 @@ export class CodexAcpServer implements acp.Agent {
     async authenticate(
         _params: acp.AuthenticateRequest,
     ): Promise<acp.AuthenticateResponse> {
-        const isAuthenticated = await this.codexAcpClient.authenticate(_params);
+        const isAuthenticated = await this.runWithProcessCheck(() => this.codexAcpClient.authenticate(_params));
         if (!isAuthenticated) {
             throw RequestError.invalidParams();
         }
@@ -118,10 +121,10 @@ export class CodexAcpServer implements acp.Agent {
         }
 
 
-        await this.codexAcpClient.setModel({
+        await this.runWithProcessCheck(() => this.codexAcpClient.setModel({
             model: model.model,
             reasoningEffort,
-        });
+        }));
         sessionState.sessionMetadata.currentModelId = ModelId.fromComponents(model, reasoningEffort).toString();
 
         return {};
@@ -155,7 +158,7 @@ export class CodexAcpServer implements acp.Agent {
 
         try {
             const messageHandler = new CodexEventHandler(this.connection, sessionState);
-            await this.codexAcpClient.sendPrompt(params, (event) => messageHandler.handleNotification(event));
+            await this.runWithProcessCheck(() => this.codexAcpClient.sendPrompt(params, (event) => messageHandler.handleNotification(event)));
         } catch (err) {
             if (sessionState.pendingPrompt.signal.aborted) {
                 return {stopReason: "cancelled"};
@@ -171,6 +174,22 @@ export class CodexAcpServer implements acp.Agent {
         };
     }
 
+    private async runWithProcessCheck<T>(operation: () => Promise<T>): Promise<T> {
+        try {
+            return await operation();
+        } catch (err) {
+            const exitCode = this.getExitCode();
+            const requestErrorCode = 1001 // Just some magic number
+            if (exitCode == 3221225781) {
+                throw new RequestError(requestErrorCode, `VC++ redistributable should be installed`);
+            }
+            if (exitCode !== null) {
+                throw new RequestError(requestErrorCode, `Codex process has exited with code ${exitCode}`);
+            }
+            throw err;
+        }
+    }
+
     async cancel(params: acp.CancelNotification): Promise<void> {
         //TODO not supported yet
         this.sessions.get(params.sessionId)?.pendingPrompt?.abort();

src/CodexEventHandler.ts

@@ -7,6 +7,7 @@ import {ACPSessionConnection, type UpdateSessionEvent} from "./ACPSessionConnect
 import type {
     AgentMessageDeltaNotification,
     CommandAction,
+    ErrorNotification,
     FileUpdateChange,
     ItemCompletedNotification,
     ItemStartedNotification,
@@ -50,6 +51,8 @@ export class CodexEventHandler {
                 return await this.completeItemEvent(notification.params);
             case "turn/plan/updated":
                 return await this.updatePlan(notification.params);
+            case "error":
+                return await this.createErrorEvent(notification.params);
             case "item/reasoning/summaryTextDelta": //TODO streaming reasoning?
             case "item/reasoning/summaryPartAdded":
             //skipped events
@@ -59,7 +62,6 @@ export class CodexEventHandler {
             case "turn/diff/updated":
             case "item/commandExecution/outputDelta":
             case "item/fileChange/outputDelta":
-            case "error":
             case "thread/tokenUsage/updated":
             case "item/mcpToolCall/progress":
             case "account/updated":
@@ -219,4 +221,14 @@ export class CodexEventHandler {
             entries: plan,
         }
     }
+
+    private async createErrorEvent(params: ErrorNotification): Promise<UpdateSessionEvent> {
+        return {
+            sessionUpdate: "agent_message_chunk",
+            content: {
+                type: "text",
+                text: `❌ ${params.error.message}\n\n`
+            }
+        }
+    }
 }

src/CodexJsonRpcConnection.ts

@@ -7,8 +7,15 @@ import fs from "node:fs";
 import path from "node:path";
 import {createJSONRPCReader, createJSONRPCWriter} from "./StdUtils";
 
-export function startCodexConnection(codexPath: string, logPath?: string): MessageConnection {
-    const codex: ChildProcessWithoutNullStreams = spawn(codexPath, ["app-server"]);
+export interface CodexConnection {
+    readonly connection: MessageConnection
+    readonly process: ChildProcessWithoutNullStreams;
+}
+
+export function startCodexConnection(codexPath: string, logPath?: string): CodexConnection {
+    const codex: ChildProcessWithoutNullStreams = process.platform === 'win32'
+        ? spawn(`"${codexPath}" app-server`, { shell: true })
+        : spawn(codexPath, ['app-server']);
 
     if (logPath) {
         attachLogs(codex, logPath);
@@ -21,7 +28,12 @@ export function startCodexConnection(codexPath: string, logPath?: string): Messa
 
     connection.listen();
 
-    return connection;
+    // Terminate all current activities on process termination
+    codex.on("exit", _ => {
+        connection.dispose();
+    });
+
+    return {connection: connection, process: codex};
 }
 
 function attachLogs(proc: ChildProcessWithoutNullStreams, logPath: string) {
@@ -39,4 +51,7 @@ function attachLogs(proc: ChildProcessWithoutNullStreams, logPath: string) {
     proc.stdout.on("data", (data: Buffer) => {
         log("[STDOUT] " + data.toString());
     });
+    proc.on("exit", (code) => {
+        log("[EXIT] " + code?.toString());
+    });
 }

src/__tests__/CodexACPAgent/CodexAcpClient.test.ts

@@ -4,7 +4,7 @@ import {createTestFixture, type TestFixture} from "../acp-test-utils";
 import type {ServerNotification} from "../../app-server";
 import type {SessionState} from "../../CodexAcpServer";
 
-describe('ACP server test', () => {
+describe('ACP server test', { timeout: 40_000 }, () => {
 
     let fixture: TestFixture;
     beforeEach(() => {
@@ -20,7 +20,7 @@ describe('ACP server test', () => {
 
         fixture.getCodexAcpClient().authRequired = vi.fn().mockResolvedValue(false);
         const newSessionResponse = await codexAcpAgent.newSession({cwd: "", mcpServers: []});
-        codexAcpAgent.prompt({ sessionId: newSessionResponse.sessionId, prompt: [{type: "text", text: "Hi!"}] });
+        codexAcpAgent.prompt({sessionId: newSessionResponse.sessionId, prompt: [{type: "text", text: "Hi!"}]});
 
         const transportDump = fixture.getCodexConnectionDump(ignoredFields);
         await expect(transportDump).toMatchFileSnapshot("data/start-conversation.json");
@@ -93,7 +93,7 @@ describe('ACP server test', () => {
 
         expect(fixture.getAcpConnectionDump([])).toMatchFileSnapshot("data/output-acp-events.json");
 
-    }, 90_000);
+    });
 
     //dev-time test
     it.skip('should convert session notification to acp events', async () => {
@@ -105,5 +105,5 @@ describe('ACP server test', () => {
         const newSessionResponse = await codexAcpAgent.newSession({cwd: "/home/alex/work/spring-petclinic/", mcpServers: []});
         fixture.clearCodexConnectionDump();
         await codexAcpAgent.prompt({ sessionId: newSessionResponse.sessionId, prompt: [{type: "text", text: "Add method `minus` to Math Utils."}] });
-    }, 90_000);
+    });
 });

src/__tests__/acp-test-utils.ts

@@ -3,6 +3,8 @@ import {type CodexConnectionEvent, CodexAppServerClient} from '../CodexAppServer
 import {startCodexConnection} from "../CodexJsonRpcConnection";
 import {CodexAcpServer} from "../CodexAcpServer";
 import type {AgentSideConnection} from "@agentclientprotocol/sdk";
+import path from "node:path";
+import fs from "node:fs";
 
 export type MethodCallEvent = { method: string; args: any[] };
 
@@ -32,17 +34,22 @@ export interface TestFixture {
 }
 
 export function createTestFixture(): TestFixture {
-    const pathToCodex = "././node_modules/.bin/codex"
+    const pathToCodex = path.resolve(process.cwd(), "node_modules", ".bin", process.platform === 'win32' ? "codex.cmd" : "codex");
+    if (!fs.existsSync(pathToCodex)) {
+        throw new Error(`Codex binary not found at ${pathToCodex}. Did you run 'npm install'?`);
+    }
+    
     const acpConnectionEvents: MethodCallEvent[] = []
     const acpEventHandlers: ((event: MethodCallEvent) => void)[] = [];
     const acpConnection = createSmartMock<AgentSideConnection>((event) => {
         acpConnectionEvents.push(event);
         acpEventHandlers.forEach(handler => handler(event));
     });
-    const codexAppServerClient = new CodexAppServerClient(startCodexConnection(pathToCodex));
+    const codexConnection = startCodexConnection(pathToCodex);
+    const codexAppServerClient = new CodexAppServerClient(codexConnection.connection);
 
     const codexAcpClient = new CodexAcpClient(codexAppServerClient);
-    const codexAcpAgent = new CodexAcpServer(acpConnection, codexAcpClient);
+    const codexAcpAgent = new CodexAcpServer(acpConnection, codexAcpClient, undefined, () => codexConnection.process.exitCode);
 
     const transportEvents: CodexConnectionEvent[] = []
     const codexEventHandlers: ((event: CodexConnectionEvent) => void)[] = [];

src/index.ts

@@ -17,7 +17,7 @@ if (process.argv.includes("--version")) {
 const codexPath = process.env["CODEX_PATH"] ?? "codex";
 const logPath = process.env["APP_SERVER_LOGS"];
 
-const appServerConnection = startCodexConnection(codexPath, logPath);
+const codexConnection = startCodexConnection(codexPath, logPath);
 const acpJsonStream = createJsonStream(process.stdin, process.stdout);
 
 function createAgent(connection: acp.AgentSideConnection): CodexAcpServer {
@@ -27,9 +27,9 @@ function createAgent(connection: acp.AgentSideConnection): CodexAcpServer {
     const authRequestString = process.env["DEFAULT_AUTH_REQUEST"];
     const parsedRequest = authRequestString ? JSON.parse(authRequestString) : undefined;
     const defaultAuthRequest = parsedRequest && isCodexAuthRequest(parsedRequest) ? parsedRequest : undefined;
-    const appServerClient = new CodexAppServerClient(appServerConnection);
+    const appServerClient = new CodexAppServerClient(codexConnection.connection);
     const codexClient = new CodexAcpClient(appServerClient, config, modelProvider)
-    return new CodexAcpServer(connection, codexClient, defaultAuthRequest);
+    return new CodexAcpServer(connection, codexClient, defaultAuthRequest, () => codexConnection.process.exitCode);
 }
 
 new acp.AgentSideConnection(createAgent, acpJsonStream);