fix(server): stop Codex threads with queued follow-ups

apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts

@@ -245,4 +245,53 @@ describe("CodexSessionRuntime collab integration", () => {
       yield* runtime.close;
     }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
   );
+
+  it.live("Stop targets the active turn when Codex has accepted a queued follow-up", () =>
+    Effect.gen(function* () {
+      const activeTurnId = "019fe3e8-f908-7f31-8d51-283f4a47897a";
+      const queuedTurnId = "019fe3eb-8faf-7de3-a85b-ac64c7f9c8c3";
+      const script = {
+        rootThreadId: ROOT,
+        holdTurnOpen: true,
+        onlyFirstTurnStarts: true,
+        turnIds: [activeTurnId, queuedTurnId],
+        expectedActiveTurnId: activeTurnId,
+        notifications: [],
+      };
+      // @effect-diagnostics-next-line preferSchemaOverJson:off
+      NodeFS.writeFileSync(scriptPath, JSON.stringify(script), "utf8");
+      const interruptsPath = `${scriptPath}.interrupts`;
+      NodeFS.rmSync(interruptsPath, { force: true });
+      yield* Effect.addFinalizer(() =>
+        Effect.sync(() => {
+          NodeFS.rmSync(scriptPath, { force: true });
+          NodeFS.rmSync(interruptsPath, { force: true });
+        }),
+      );
+
+      const runtime = yield* makeCodexSessionRuntime({
+        threadId: ThreadId.make("thread-codex-queued-stop"),
+        binaryPath: peerPath,
+        cwd: "/tmp",
+        runtimeMode: "full-access",
+        environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath },
+      });
+
+      yield* runtime.start();
+      yield* runtime.sendTurn({ input: "keep working" });
+      yield* runtime.sendTurn({ input: "queued follow-up" });
+      yield* runtime.interruptTurn();
+
+      const interrupts = NodeFS.readFileSync(interruptsPath, "utf8")
+        .trim()
+        .split("\n")
+        .map((line) => JSON.parse(line) as { threadId?: string; turnId?: string });
+      assert.deepEqual(interrupts.at(-1), {
+        threadId: ROOT,
+        turnId: activeTurnId,
+      });
+
+      yield* runtime.close;
+    }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
+  );
 });

apps/server/src/provider/Layers/CodexSessionRuntime.ts

@@ -814,13 +814,13 @@ function currentProviderThreadId(session: ProviderSession): string | undefined {
 
 function updateSession(
   sessionRef: Ref.Ref<ProviderSession>,
-  updates: Partial<ProviderSession>,
+  updates: Partial<ProviderSession> | ((session: ProviderSession) => Partial<ProviderSession>),
 ): Effect.Effect<void> {
   return Effect.gen(function* () {
     const updatedAt = DateTime.formatIso(yield* DateTime.now);
     yield* Ref.update(sessionRef, (session) => ({
       ...session,
-      ...updates,
+      ...(typeof updates === "function" ? updates(session) : updates),
       updatedAt,
     }));
   });
@@ -1782,11 +1782,14 @@ export const makeCodexSessionRuntime = (
             ),
           );
           const turnId = TurnId.make(response.turn.id);
-          yield* updateSession(sessionRef, {
+          yield* updateSession(sessionRef, (session) => ({
             status: "running",
-            activeTurnId: turnId,
+            // Codex accepts follow-ups while the current turn is still
+            // running. The response contains the queued turn id, but
+            // turn/interrupt only accepts the id that is active now.
+            activeTurnId: session.activeTurnId ?? turnId,
             ...(normalizedModel ? { model: normalizedModel } : {}),
-          });
+          }));
           const resumedProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef));
           return {
             threadId: options.threadId,

apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs

@@ -16,6 +16,7 @@ const fixture = JSON.parse(
 const script = JSON.parse(NodeFS.readFileSync(process.env.T3_CODEX_COLLAB_SCRIPT, "utf8"));
 
 const write = (message) => process.stdout.write(`${JSON.stringify(message)}\n`);
+let turnStartCount = 0;
 
 const rl = NodeReadline.createInterface({ input: process.stdin });
 rl.on("line", (line) => {
@@ -43,14 +44,20 @@ rl.on("line", (line) => {
     return;
   }
   if (method === "turn/start") {
-    write({ id, result: fixture.responses.turnStart });
+    const turnId = script.turnIds?.[turnStartCount];
+    const turn = turnId
+      ? { ...fixture.responses.turnStart.turn, id: turnId }
+      : fixture.responses.turnStart.turn;
+    turnStartCount += 1;
+    write({ id, result: { ...fixture.responses.turnStart, turn } });
     const rootThreadId = script.rootThreadId;
-    const turn = fixture.responses.turnStart.turn;
-    write({
-      jsonrpc: "2.0",
-      method: "turn/started",
-      params: { threadId: rootThreadId, turn },
-    });
+    if (script.onlyFirstTurnStarts !== true || turnStartCount === 1) {
+      write({
+        jsonrpc: "2.0",
+        method: "turn/started",
+        params: { threadId: rootThreadId, turn },
+      });
+    }
     for (const notification of script.notifications) {
       write({ jsonrpc: "2.0", method: notification.method, params: notification.params });
     }
@@ -75,6 +82,20 @@ rl.on("line", (line) => {
       `${process.env.T3_CODEX_COLLAB_SCRIPT}.interrupts`,
       `${JSON.stringify({ threadId: target, turnId: message.params?.turnId })}\n`,
     );
+    if (
+      script.expectedActiveTurnId &&
+      message.params?.threadId === script.rootThreadId &&
+      message.params?.turnId !== script.expectedActiveTurnId
+    ) {
+      write({
+        id,
+        error: {
+          code: -32000,
+          message: `expected active turn id ${message.params?.turnId} but found ${script.expectedActiveTurnId}`,
+        },
+      });
+      return;
+    }
     if (script.failInterruptFor && script.failInterruptFor === target) {
       write({ id, error: { code: -32000, message: "thread already closed" } });
       return;