authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-08 20:52:06-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-08 17:52:06-07:00
log89c320df0b0884a8c4df1cf596564c6bc725eb54
tree020f899c6f00a2605cb3244acc3eac75f780b42a
parent7b2cf4374f5d92cc01eff482ad1af92ab7a87f41
signature Signed by PGP key B5690EEEBB952194

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

3 files changed, 85 insertions(+), 12 deletions(-)

apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts+49
...@@ -245,4 +245,53 @@ describe("CodexSessionRuntime collab integration", () => {...@@ -245,4 +245,53 @@ describe("CodexSessionRuntime collab integration", () => {
245 yield* runtime.close;245 yield* runtime.close;
246 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),246 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
247 );247 );
248
249 it.live("Stop targets the active turn when Codex has accepted a queued follow-up", () =>
250 Effect.gen(function* () {
251 const activeTurnId = "019fe3e8-f908-7f31-8d51-283f4a47897a";
252 const queuedTurnId = "019fe3eb-8faf-7de3-a85b-ac64c7f9c8c3";
253 const script = {
254 rootThreadId: ROOT,
255 holdTurnOpen: true,
256 onlyFirstTurnStarts: true,
257 turnIds: [activeTurnId, queuedTurnId],
258 expectedActiveTurnId: activeTurnId,
259 notifications: [],
260 };
261 // @effect-diagnostics-next-line preferSchemaOverJson:off
262 NodeFS.writeFileSync(scriptPath, JSON.stringify(script), "utf8");
263 const interruptsPath = `${scriptPath}.interrupts`;
264 NodeFS.rmSync(interruptsPath, { force: true });
265 yield* Effect.addFinalizer(() =>
266 Effect.sync(() => {
267 NodeFS.rmSync(scriptPath, { force: true });
268 NodeFS.rmSync(interruptsPath, { force: true });
269 }),
270 );
271
272 const runtime = yield* makeCodexSessionRuntime({
273 threadId: ThreadId.make("thread-codex-queued-stop"),
274 binaryPath: peerPath,
275 cwd: "/tmp",
276 runtimeMode: "full-access",
277 environment: { ...process.env, T3_CODEX_COLLAB_SCRIPT: scriptPath },
278 });
279
280 yield* runtime.start();
281 yield* runtime.sendTurn({ input: "keep working" });
282 yield* runtime.sendTurn({ input: "queued follow-up" });
283 yield* runtime.interruptTurn();
284
285 const interrupts = NodeFS.readFileSync(interruptsPath, "utf8")
286 .trim()
287 .split("\n")
288 .map((line) => JSON.parse(line) as { threadId?: string; turnId?: string });
289 assert.deepEqual(interrupts.at(-1), {
290 threadId: ROOT,
291 turnId: activeTurnId,
292 });
293
294 yield* runtime.close;
295 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
296 );
248});297});
apps/server/src/provider/Layers/CodexSessionRuntime.ts+8-5
...@@ -814,13 +814,13 @@ function currentProviderThreadId(session: ProviderSession): string | undefined {...@@ -814,13 +814,13 @@ function currentProviderThreadId(session: ProviderSession): string | undefined {
814814
815function updateSession(815function updateSession(
816 sessionRef: Ref.Ref<ProviderSession>,816 sessionRef: Ref.Ref<ProviderSession>,
817 updates: Partial<ProviderSession>,817 updates: Partial<ProviderSession> | ((session: ProviderSession) => Partial<ProviderSession>),
818): Effect.Effect<void> {818): Effect.Effect<void> {
819 return Effect.gen(function* () {819 return Effect.gen(function* () {
820 const updatedAt = DateTime.formatIso(yield* DateTime.now);820 const updatedAt = DateTime.formatIso(yield* DateTime.now);
821 yield* Ref.update(sessionRef, (session) => ({821 yield* Ref.update(sessionRef, (session) => ({
822 ...session,822 ...session,
823 ...updates,823 ...(typeof updates === "function" ? updates(session) : updates),
824 updatedAt,824 updatedAt,
825 }));825 }));
826 });826 });
...@@ -1782,11 +1782,14 @@ export const makeCodexSessionRuntime = (...@@ -1782,11 +1782,14 @@ export const makeCodexSessionRuntime = (
1782 ),1782 ),
1783 );1783 );
1784 const turnId = TurnId.make(response.turn.id);1784 const turnId = TurnId.make(response.turn.id);
1785 yield* updateSession(sessionRef, {1785 yield* updateSession(sessionRef, (session) => ({
1786 status: "running",1786 status: "running",
1787 activeTurnId: turnId,1787 // Codex accepts follow-ups while the current turn is still
1788 // running. The response contains the queued turn id, but
1789 // turn/interrupt only accepts the id that is active now.
1790 activeTurnId: session.activeTurnId ?? turnId,
1788 ...(normalizedModel ? { model: normalizedModel } : {}),1791 ...(normalizedModel ? { model: normalizedModel } : {}),
1789 });1792 }));
1790 const resumedProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef));1793 const resumedProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef));
1791 return {1794 return {
1792 threadId: options.threadId,1795 threadId: options.threadId,
apps/server/src/provider/testFixtures/codexCollabMockPeer.mjs+28-7
...@@ -16,6 +16,7 @@ const fixture = JSON.parse(...@@ -16,6 +16,7 @@ const fixture = JSON.parse(
16const script = JSON.parse(NodeFS.readFileSync(process.env.T3_CODEX_COLLAB_SCRIPT, "utf8"));16const script = JSON.parse(NodeFS.readFileSync(process.env.T3_CODEX_COLLAB_SCRIPT, "utf8"));
1717
18const write = (message) => process.stdout.write(`${JSON.stringify(message)}\n`);18const write = (message) => process.stdout.write(`${JSON.stringify(message)}\n`);
19let turnStartCount = 0;
1920
20const rl = NodeReadline.createInterface({ input: process.stdin });21const rl = NodeReadline.createInterface({ input: process.stdin });
21rl.on("line", (line) => {22rl.on("line", (line) => {
...@@ -43,14 +44,20 @@ rl.on("line", (line) => {...@@ -43,14 +44,20 @@ rl.on("line", (line) => {
43 return;44 return;
44 }45 }
45 if (method === "turn/start") {46 if (method === "turn/start") {
46 write({ id, result: fixture.responses.turnStart });47 const turnId = script.turnIds?.[turnStartCount];
48 const turn = turnId
49 ? { ...fixture.responses.turnStart.turn, id: turnId }
50 : fixture.responses.turnStart.turn;
51 turnStartCount += 1;
52 write({ id, result: { ...fixture.responses.turnStart, turn } });
47 const rootThreadId = script.rootThreadId;53 const rootThreadId = script.rootThreadId;
48 const turn = fixture.responses.turnStart.turn;54 if (script.onlyFirstTurnStarts !== true || turnStartCount === 1) {
49 write({55 write({
50 jsonrpc: "2.0",56 jsonrpc: "2.0",
51 method: "turn/started",57 method: "turn/started",
52 params: { threadId: rootThreadId, turn },58 params: { threadId: rootThreadId, turn },
53 });59 });
60 }
54 for (const notification of script.notifications) {61 for (const notification of script.notifications) {
55 write({ jsonrpc: "2.0", method: notification.method, params: notification.params });62 write({ jsonrpc: "2.0", method: notification.method, params: notification.params });
56 }63 }
...@@ -75,6 +82,20 @@ rl.on("line", (line) => {...@@ -75,6 +82,20 @@ rl.on("line", (line) => {
75 `${process.env.T3_CODEX_COLLAB_SCRIPT}.interrupts`,82 `${process.env.T3_CODEX_COLLAB_SCRIPT}.interrupts`,
76 `${JSON.stringify({ threadId: target, turnId: message.params?.turnId })}\n`,83 `${JSON.stringify({ threadId: target, turnId: message.params?.turnId })}\n`,
77 );84 );
85 if (
86 script.expectedActiveTurnId &&
87 message.params?.threadId === script.rootThreadId &&
88 message.params?.turnId !== script.expectedActiveTurnId
89 ) {
90 write({
91 id,
92 error: {
93 code: -32000,
94 message: `expected active turn id ${message.params?.turnId} but found ${script.expectedActiveTurnId}`,
95 },
96 });
97 return;
98 }
78 if (script.failInterruptFor && script.failInterruptFor === target) {99 if (script.failInterruptFor && script.failInterruptFor === target) {
79 write({ id, error: { code: -32000, message: "thread already closed" } });100 write({ id, error: { code: -32000, message: "thread already closed" } });
80 return;101 return;