diff --git a/packages/effect-codex-app-server/src/client.test.ts b/packages/effect-codex-app-server/src/client.test.ts index 3830c5fc5f6f..ab2377990e16 100644 --- a/packages/effect-codex-app-server/src/client.test.ts +++ b/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"); + }), + ); }); diff --git a/packages/effect-codex-app-server/src/client.ts b/packages/effect-codex-app-server/src/client.ts index c0cb5b1dc23a..41ef13b8e431 100644 --- a/packages/effect-codex-app-server/src/client.ts +++ b/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), + ); }); diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index a7e0397b4adb..b57b67d69de3 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/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 = []; + 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(); + 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(); diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index 17bfaed2b64c..fc9d5fa7aee4 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/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; readonly logIncoming?: boolean; readonly logOutgoing?: boolean; @@ -158,6 +161,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa const pending = yield* Ref.make(new Map()); 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())), + ), + ) + : 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), ); }; diff --git a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts b/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts index 3f2a213d38c7..4d6928dffb28 100644 --- a/packages/effect-codex-app-server/test/fixtures/codex-app-server-mock-peer.ts +++ b/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`); };