diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 10e979c650108..dcb6da696f799 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -417,6 +417,75 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { }), ); + it.effect("releases a waiting request handler when Codex resolves its request", () => + Effect.gen(function* () { + const { stdio, input, output } = yield* makeInMemoryStdio(); + yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + onRequest: (request) => + request.id === 33 ? Effect.succeed({ decision: "accept" }) : Effect.never, + }); + + const messages = Array.from({ length: 32 }, (_, index) => + encodeUnknownJsonString({ + id: index + 1, + method: "item/commandExecution/requestApproval", + params: {}, + }), + ); + for (let requestId = 1; requestId <= 32; requestId++) { + messages.push( + encodeUnknownJsonString({ + method: "serverRequest/resolved", + params: { threadId: "thread-1", requestId }, + }), + ); + } + messages.push( + encodeUnknownJsonString({ + id: 33, + method: "item/commandExecution/requestApproval", + params: {}, + }), + ); + yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`)); + + assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), { + id: 33, + result: { decision: "accept" }, + }); + }), + ); + + it.effect("keeps numeric and string request ids apart when Codex resolves one", () => + Effect.gen(function* () { + const { stdio, input, output } = yield* makeInMemoryStdio(); + const interrupted: Array = []; + yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + onRequest: (request) => + request.id === 2 + ? Effect.sync(() => ({ interrupted })) + : Effect.never.pipe( + Effect.onInterrupt(() => Effect.sync(() => interrupted.push(request.id))), + ), + }); + + const messages = [ + { id: 1, method: "item/commandExecution/requestApproval", params: {} }, + { id: "1", method: "item/commandExecution/requestApproval", params: {} }, + { method: "serverRequest/resolved", params: { threadId: "thread-1", requestId: "1" } }, + { id: 2, method: "item/commandExecution/requestApproval", params: {} }, + ].map((message) => encodeUnknownJsonString(message)); + yield* Queue.offer(input, encoder.encode(`${messages.join("\n")}\n`)); + + assert.deepEqual(yield* decodeJson(yield* Queue.take(output)), { + id: 2, + result: { interrupted: ["1"] }, + }); + }), + ); + it.effect("interrupts pending request handlers when the protocol terminates", () => Effect.gen(function* () { const { stdio, input } = yield* makeInMemoryStdio(); diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index bf67174033f4d..33abd2bc14b04 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -2,6 +2,7 @@ import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as FiberMap from "effect/FiberMap"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; @@ -175,6 +176,9 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa const terminationFailure = yield* Ref.make(Option.none()); const terminationSignal = yield* Deferred.make(); const activeRequestHandlers = yield* Ref.make(0); + const requestHandlers = yield* FiberMap.make().pipe( + Scope.provide(requestHandlerScope), + ); const logProtocol = (event: CodexAppServerProtocolLogEvent) => { if (event.direction === "incoming" && !options.logIncoming) { @@ -342,7 +346,7 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa Effect.asVoid, ), ), - Effect.forkIn(requestHandlerScope, { startImmediately: true }), + FiberMap.run(requestHandlers, request.id), Effect.asVoid, ); }), @@ -351,8 +355,17 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa Effect.asVoid, ); + // Codex settles requests itself when their turn ends; a handler still waiting would hold its slot. + const releaseResolvedRequest = (notification: CodexAppServerIncomingNotification) => + notification.method === "serverRequest/resolved" && + isObject(notification.params) && + isJsonRpcId(notification.params.requestId) + ? FiberMap.remove(requestHandlers, notification.params.requestId) + : Effect.void; + const handleNotification = (notification: CodexAppServerIncomingNotification) => - Queue.offer(incomingNotifications, notification).pipe( + releaseResolvedRequest(notification).pipe( + Effect.andThen(Queue.offer(incomingNotifications, notification)), Effect.andThen(options.onNotification ? options.onNotification(notification) : Effect.void), Effect.asVoid, );