fix(codex): tolerate launcher stdout preambles

packages/effect-codex-app-server/src/client.test.ts

@@ -154,4 +154,29 @@ it.layer(NodeServices.layer)("effect-codex-app-server client", (it) => {
       assert.equal(initialized.userAgent, "mock-codex-app-server");
     }),
   );
+
+  it.effect("initializes through a child launcher that writes a non-JSON preamble", () =>
+    Effect.gen(function* () {
+      const handle = yield* makeHandle({ CODEX_APP_SERVER_TEST_STDOUT_PREAMBLE: "1" });
+      const scope = yield* Scope.make();
+      const context = yield* Layer.buildWithScope(CodexClient.layerChildProcess(handle), scope);
+
+      const initialized = yield* Effect.gen(function* () {
+        const client = yield* CodexClient.CodexAppServerClient;
+        return yield* client.request("initialize", {
+          clientInfo: {
+            name: "effect-codex-app-server-test",
+            title: "Effect Codex App Server Test",
+            version: "0.0.0",
+          },
+          capabilities: {
+            experimentalApi: true,
+            optOutNotificationMethods: null,
+          },
+        });
+      }).pipe(Effect.provide(context), Effect.ensuring(Scope.close(scope, Exit.void)));
+
+      assert.equal(initialized.userAgent, "mock-codex-app-server");
+    }),
+  );
 });

packages/effect-codex-app-server/src/client.ts

@@ -19,6 +19,7 @@ import {
 import { makeChildStdio, makeTerminationError } from "./_internal/stdio.ts";
 
 export interface CodexAppServerClientOptions {
+  readonly ignoreNonJsonPreamble?: boolean;
   readonly logIncoming?: boolean;
   readonly logOutgoing?: boolean;
   readonly logger?: (
@@ -186,6 +187,9 @@ export const make = Effect.fn("effect-codex-app-server/CodexAppServerClient.make
 
   const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
     stdio,
+    ...(options.ignoreNonJsonPreamble !== undefined
+      ? { ignoreNonJsonPreamble: options.ignoreNonJsonPreamble }
+      : {}),
     ...(terminationError ? { terminationError } : {}),
     ...(options.logIncoming !== undefined ? { logIncoming: options.logIncoming } : {}),
     ...(options.logOutgoing !== undefined ? { logOutgoing: options.logOutgoing } : {}),
@@ -265,5 +269,9 @@ const makeChildProcessClient = Effect.fn(
   "effect-codex-app-server/CodexAppServerClient.makeChildProcessClient",
 )(function* (handle: ChildProcessSpawner.ChildProcessHandle, options: CodexAppServerClientOptions) {
   yield* Stream.runDrain(handle.stderr).pipe(Effect.ignore, Effect.forkScoped);
-  return yield* make(makeChildStdio(handle), options, makeTerminationError(handle));
+  return yield* make(
+    makeChildStdio(handle),
+    { ignoreNonJsonPreamble: true, ...options },
+    makeTerminationError(handle),
+  );
 });

packages/effect-codex-app-server/src/protocol.test.ts

@@ -328,6 +328,62 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => {
     }),
   );
 
+  it.effect("ignores non-JSON launcher output before the first protocol message", () =>
+    Effect.gen(function* () {
+      const { stdio, input, output } = yield* makeInMemoryStdio();
+      const events: Array<CodexProtocol.CodexAppServerProtocolLogEvent> = [];
+      const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
+        stdio,
+        ignoreNonJsonPreamble: true,
+        logIncoming: true,
+        logger: (event) =>
+          Effect.sync(() => {
+            events.push(event);
+          }),
+      });
+
+      const response = yield* transport.request("initialize", {}).pipe(Effect.forkScoped);
+      yield* Queue.take(output);
+      yield* Queue.offer(
+        input,
+        encoder.encode("mise ~/.config/mise/config.toml tools: codex@0.149.0\n"),
+      );
+      yield* Queue.offer(input, encodeJsonl({ id: 1, result: { userAgent: "codex/0.149.0" } }));
+
+      assert.deepEqual(yield* Fiber.join(response), { userAgent: "codex/0.149.0" });
+      assert.deepInclude(
+        events.find(({ stage }) => stage === "ignored_preamble"),
+        {
+          direction: "incoming",
+          stage: "ignored_preamble",
+          payload: { byteLength: 52 },
+        },
+      );
+    }),
+  );
+
+  it.effect("rejects non-JSON output after protocol messages begin", () =>
+    Effect.gen(function* () {
+      const { stdio, input, output } = yield* makeInMemoryStdio();
+      const termination = yield* Deferred.make<CodexError.CodexAppServerError>();
+      const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({
+        stdio,
+        ignoreNonJsonPreamble: true,
+        onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid),
+      });
+
+      const response = yield* transport.request("initialize", {}).pipe(Effect.forkScoped);
+      yield* Queue.take(output);
+      yield* Queue.offer(input, encodeJsonl({ id: 1, result: {} }));
+      yield* Fiber.join(response);
+      yield* Queue.offer(input, encoder.encode("unexpected output\n"));
+
+      const error = yield* Deferred.await(termination);
+      assert.instanceOf(error, CodexError.CodexAppServerProtocolParseError);
+      assert.equal(error.operation, "decode-wire-message");
+    }),
+  );
+
   it.effect("correlates response errors with the originating request", () =>
     Effect.gen(function* () {
       const { stdio, input, output } = yield* makeInMemoryStdio();

packages/effect-codex-app-server/src/protocol.ts

@@ -1,6 +1,7 @@
 import * as Cause from "effect/Cause";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Option from "effect/Option";
 import * as Queue from "effect/Queue";
 import * as Ref from "effect/Ref";
 import * as Scope from "effect/Scope";
@@ -13,10 +14,11 @@ import { JsonRpcId, JsonRpcResponseEnvelope } from "./_internal/shared.ts";
 const isJsonRpcId = Schema.is(JsonRpcId);
 const isJsonRpcResponseEnvelope = Schema.is(JsonRpcResponseEnvelope);
 const isCodexAppServerError = Schema.is(CodexError.CodexAppServerError);
+const textEncoder = new TextEncoder();
 
 export interface CodexAppServerProtocolLogEvent {
   readonly direction: "incoming" | "outgoing";
-  readonly stage: "raw" | "decoded" | "decode_failed";
+  readonly stage: "raw" | "decoded" | "decode_failed" | "ignored_preamble";
   readonly payload: unknown;
 }
 
@@ -33,6 +35,7 @@ export interface CodexAppServerIncomingRequest {
 
 export interface CodexAppServerPatchedProtocolOptions {
   readonly stdio: Stdio.Stdio;
+  readonly ignoreNonJsonPreamble?: boolean;
   readonly terminationError?: Effect.Effect<CodexError.CodexAppServerError>;
   readonly logIncoming?: boolean;
   readonly logOutgoing?: boolean;
@@ -158,6 +161,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
     const pending = yield* Ref.make(new Map<string, CodexAppServerPendingRequest>());
     const nextRequestId = yield* Ref.make(1);
     const remainder = yield* Ref.make("");
+    const receivedProtocolMessage = yield* Ref.make(false);
     const terminationHandled = yield* Ref.make(false);
 
     const logProtocol = (event: CodexAppServerProtocolLogEvent) => {
@@ -323,12 +327,39 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
         stage: "raw",
         payload: line,
       }).pipe(
-        Effect.flatMap(() => decodeWireMessage(line)),
-        Effect.tap((decoded) =>
-          logProtocol({
-            direction: "incoming",
-            stage: "decoded",
-            payload: decoded,
+        Effect.flatMap(() =>
+          decodeWireMessage(line).pipe(
+            Effect.map(Option.some),
+            Effect.catchTags({
+              CodexAppServerProtocolParseError: (error) =>
+                options.ignoreNonJsonPreamble
+                  ? Ref.get(receivedProtocolMessage).pipe(
+                      Effect.flatMap((received) =>
+                        received
+                          ? Effect.fail(error)
+                          : logProtocol({
+                              direction: "incoming",
+                              stage: "ignored_preamble",
+                              payload: { byteLength: textEncoder.encode(line).byteLength },
+                            }).pipe(Effect.as(Option.none<unknown>())),
+                      ),
+                    )
+                  : Effect.fail(error),
+            }),
+          ),
+        ),
+        Effect.flatMap(
+          Option.match({
+            onNone: () => Effect.void,
+            onSome: (decoded) =>
+              logProtocol({
+                direction: "incoming",
+                stage: "decoded",
+                payload: decoded,
+              }).pipe(
+                Effect.andThen(routeMessage(decoded)),
+                Effect.andThen(Ref.set(receivedProtocolMessage, true)),
+              ),
           }),
         ),
         Effect.tapErrorTag("CodexAppServerProtocolParseError", (error) =>
@@ -347,7 +378,6 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa
             },
           }),
         ),
-        Effect.flatMap(routeMessage),
       );
     };
 

packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts

@@ -4,6 +4,10 @@ let nextServerRequestId = 10_000;
 let pendingSkillsListRequestId: number | string | null = null;
 let pendingUserInputRequestId: number | null = null;
 
+if (process.env.CODEX_APP_SERVER_TEST_STDOUT_PREAMBLE === "1") {
+  process.stdout.write("mise ~/.config/mise/config.toml tools: codex@0.149.0\n");
+}
+
 const writeMessage = (message: unknown) => {
   process.stdout.write(`${JSON.stringify(message)}\n`);
 };