fix(codex): approvals still appear after many stopped Codex turns

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<string | number> = [];
+      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();

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<CodexError.CodexAppServerError>());
     const terminationSignal = yield* Deferred.make<void>();
     const activeRequestHandlers = yield* Ref.make(0);
+    const requestHandlers = yield* FiberMap.make<string | number>().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,
       );