diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 3a02c45b23f..38e0e0a7b2c 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/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)), + ); }); diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 57a1162dd08..58c012bd63e 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -814,13 +814,13 @@ function currentProviderThreadId(session: ProviderSession): string | undefined { function updateSession( sessionRef: Ref.Ref, - updates: Partial, + updates: Partial | ((session: ProviderSession) => Partial), ): Effect.Effect { 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, diff --git a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs b/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs index 59580d2c7e6..f06e984c9aa 100644 --- a/apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs +++ b/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;