diff --git a/plugins/provider-codex/src/bridge/app-server-connection.ts b/plugins/provider-codex/src/bridge/app-server-connection.ts index d055b8e54e..d876babdce 100644 --- a/plugins/provider-codex/src/bridge/app-server-connection.ts +++ b/plugins/provider-codex/src/bridge/app-server-connection.ts @@ -44,7 +44,7 @@ interface CodexAppServerRequestArgs { export interface CodexAppServerConnection { request(args: CodexAppServerRequestArgs): Promise; notify(method: string, params?: unknown): void; - kill(): void; + kill(): Promise; readonly exited: boolean; } @@ -112,6 +112,10 @@ export function createCodexAppServerConnection( } | null = null; let closeGraceTimer: NodeJS.Timeout | null = null; let stdoutLines: Interface | null = null; + let resolveExit!: () => void; + const exitPromise = new Promise((resolve) => { + resolveExit = resolve; + }); function writeLine(message: object): void { const stdin = child.stdin; @@ -155,7 +159,11 @@ export function createCodexAppServerConnection( { spawnFailed }, ), ); - options.onExit({ ...status, stderrTail, spawnFailed }); + try { + options.onExit({ ...status, stderrTail, spawnFailed }); + } finally { + resolveExit(); + } } if (child.stdout) { @@ -313,7 +321,7 @@ export function createCodexAppServerConnection( kill() { if (finalized) { - return; + return exitPromise; } const escalation = setTimeout(() => { if (!finalized) { @@ -322,6 +330,7 @@ export function createCodexAppServerConnection( }, KILL_ESCALATION_MS); escalation.unref?.(); child.kill("SIGTERM"); + return exitPromise; }, }; } diff --git a/plugins/provider-codex/src/bridge/bridge-process.test-support.ts b/plugins/provider-codex/src/bridge/bridge-process.test-support.ts new file mode 100644 index 0000000000..30b2467005 --- /dev/null +++ b/plugins/provider-codex/src/bridge/bridge-process.test-support.ts @@ -0,0 +1,130 @@ +import { existsSync, readFileSync, rmSync } from "node:fs"; +import type { experimental_createBridgeJsonRpcTestHarness as createBridgeJsonRpcTestHarness } from "@get-bb/plugin-sdk/provider-bridge/testing"; + +type BridgeJsonRpcTestHarness = ReturnType< + typeof createBridgeJsonRpcTestHarness +>; + +const PROCESS_POLL_INTERVAL_MS = 20; + +function readProcessLog(processLogPath: string): string { + if (!existsSync(processLogPath)) { + return ""; + } + try { + return readFileSync(processLogPath, "utf8"); + } catch (error) { + if (error instanceof Error && Reflect.get(error, "code") === "ENOENT") { + return ""; + } + throw error; + } +} + +export function spawnedAppServerPids(processLogPath: string): number[] { + return readProcessLog(processLogPath) + .split("\n") + .filter((line) => line.startsWith("spawn:")) + .map((line) => Number(line.split(":")[1])) + .filter((pid) => Number.isSafeInteger(pid) && pid > 0); +} + +function processIsAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + if (error instanceof Error && Reflect.get(error, "code") === "ESRCH") { + return false; + } + throw error; + } +} + +export async function waitForAppServerChildrenToExit( + processLogPath: string, + timeoutMs = 8_000, +): Promise { + const childPids = spawnedAppServerPids(processLogPath); + const deadline = Date.now() + timeoutMs; + while (childPids.some(processIsAlive)) { + if (Date.now() > deadline) { + throw new Error( + `Timed out waiting for app-server children to exit: ${JSON.stringify(childPids.filter(processIsAlive))}`, + ); + } + await new Promise((resolveTick) => + setTimeout(resolveTick, PROCESS_POLL_INTERVAL_MS), + ); + } +} + +export async function waitForAppServerProcessStep( + processLogPath: string, + step: string, + timeoutMs = 5_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (readProcessLog(processLogPath).includes(`${step}:`)) { + return; + } + await new Promise((resolveTick) => + setTimeout(resolveTick, PROCESS_POLL_INTERVAL_MS), + ); + } + throw new Error(`Timed out waiting for app-server process step: ${step}`); +} + +export async function cleanupBridgeProcessTest(args: { + harness: BridgeJsonRpcTestHarness | undefined; + cleanupId: number; + threadId: string; + providerThreadId: string; + processLogPath: string; + workspaceDir: string; + unstubEnvs: () => void; +}): Promise { + const errors: unknown[] = []; + if (args.harness !== undefined) { + try { + args.harness.sendRequest(args.cleanupId, "thread/stop", { + threadId: args.threadId, + providerThreadId: args.providerThreadId, + intent: "release", + activeTurnId: null, + }); + await args.harness.waitForResponse(args.cleanupId).catch(() => undefined); + } catch (error) { + errors.push(error); + } + } + try { + await waitForAppServerChildrenToExit(args.processLogPath); + } catch (error) { + errors.push(error); + } + try { + args.harness?.restore(); + } catch (error) { + errors.push(error); + } + try { + args.unstubEnvs(); + } catch (error) { + errors.push(error); + } + try { + if (args.workspaceDir !== "") { + rmSync(args.workspaceDir, { recursive: true, force: true }); + } + } catch (error) { + errors.push(error); + } + if (errors.length === 1) { + throw errors[0]; + } + if (errors.length > 1) { + throw new AggregateError(errors, "Bridge process test cleanup failed"); + } +} diff --git a/plugins/provider-codex/src/bridge/bridge.archived-rebuild.test.ts b/plugins/provider-codex/src/bridge/bridge.archived-rebuild.test.ts index c09a785a06..8a451e51ed 100644 --- a/plugins/provider-codex/src/bridge/bridge.archived-rebuild.test.ts +++ b/plugins/provider-codex/src/bridge/bridge.archived-rebuild.test.ts @@ -1,4 +1,4 @@ -import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { mkdtempSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; @@ -6,6 +6,10 @@ import { afterEach, beforeEach, expect, it, vi } from "vitest"; import { experimental_createBridgeJsonRpcTestHarness as createBridgeJsonRpcTestHarness } from "@get-bb/plugin-sdk/provider-bridge/testing"; import type { BridgeJsonRpcOutputMessage } from "@get-bb/plugin-sdk/provider-bridge/testing"; import { handleLine } from "./bridge.js"; +import { + cleanupBridgeProcessTest, + spawnedAppServerPids, +} from "./bridge-process.test-support.js"; const THREAD_ID = "thr_archived_rebuild_1"; const PROVIDER_THREAD_ID = "rebuild-rollout-1"; @@ -31,9 +35,9 @@ const changedSessionOptions = { const turnInput = [{ type: "text", text: "hello", mentions: [] }]; let harness: ReturnType; -let workspaceDir: string; -let archiveStatePath: string; -let processLogPath: string; +let workspaceDir = ""; +let archiveStatePath = ""; +let processLogPath = ""; beforeEach(() => { workspaceDir = mkdtempSync(join(tmpdir(), "bb-codex-archived-rebuild-")); @@ -53,52 +57,17 @@ beforeEach(() => { }); afterEach(async () => { - const cleanupId = 993_001; - harness.sendRequest(cleanupId, "thread/stop", { + await cleanupBridgeProcessTest({ + harness, + cleanupId: 993_001, threadId: THREAD_ID, providerThreadId: PROVIDER_THREAD_ID, - intent: "release", - activeTurnId: null, + processLogPath, + workspaceDir, + unstubEnvs: vi.unstubAllEnvs, }); - await harness.waitForResponse(cleanupId).catch(() => undefined); - await waitForAppServerChildrenToExit(); - harness.restore(); - vi.unstubAllEnvs(); - rmSync(workspaceDir, { recursive: true, force: true }); }); -function spawnedAppServerPids(): number[] { - return readFileSync(processLogPath, "utf8") - .split("\n") - .filter((line) => line.startsWith("spawn:")) - .map((line) => Number(line.split(":")[1])); -} - -function processIsAlive(pid: number): boolean { - try { - process.kill(pid, 0); - return true; - } catch (error) { - if (error instanceof Error && Reflect.get(error, "code") === "ESRCH") { - return false; - } - throw error; - } -} - -async function waitForAppServerChildrenToExit(): Promise { - const childPids = spawnedAppServerPids(); - const deadline = Date.now() + 15_000; - while (childPids.some(processIsAlive)) { - if (Date.now() > deadline) { - throw new Error( - `Timed out waiting for app-server children to exit: ${JSON.stringify(childPids.filter(processIsAlive))}`, - ); - } - await new Promise((resolveTick) => setTimeout(resolveTick, 20)); - } -} - async function resumeThread(): Promise { harness.sendRequest(1, "thread/resume", { threadId: THREAD_ID, @@ -197,11 +166,11 @@ it("keeps the thread resumable when a settings-change rebuild hits an externally it("keeps the thread resumable when the rebuild after the child died hits an externally archived rollout", async () => { await resumeThread(); - const spawnLine = readFileSync(processLogPath, "utf8") - .split("\n") - .find((line) => line.startsWith("spawn:")); - const childPid = Number(spawnLine?.split(":")[1]); - expect(Number.isInteger(childPid)).toBe(true); + const [childPid] = spawnedAppServerPids(processLogPath); + expect(childPid).toBeDefined(); + if (childPid === undefined) { + throw new Error("Expected the fake app-server child to have spawned"); + } process.kill(childPid, "SIGKILL"); const deadline = Date.now() + 15_000; while ( diff --git a/plugins/provider-codex/src/bridge/bridge.ts b/plugins/provider-codex/src/bridge/bridge.ts index 5914b4f727..3b1fdbcd7c 100644 --- a/plugins/provider-codex/src/bridge/bridge.ts +++ b/plugins/provider-codex/src/bridge/bridge.ts @@ -276,6 +276,9 @@ const CHILD_REQUEST_TIMEOUT_MS = 60_000; const INTERRUPT_SETTLEMENT_TIMEOUT_MS = 5_000; const CODEX_ARCHIVED_SESSION_ERROR_PATTERN = /\b(?:session|thread)\s+\S+\s+is archived\b/i; +const CODEX_ACTIVE_WRITER_ERROR_PATTERN = + /\bthread\s+\S+\s+already has an active writer\b/i; +const CODEX_ACTIVE_WRITER_RETRY_DELAYS_MS = [100, 400, 1_000] as const; const CODEX_ALREADY_ARCHIVED_ERROR_PATTERN = /\bno rollout found for thread id\b/i; const CODEX_NOT_ARCHIVED_ERROR_PATTERN = @@ -318,6 +321,12 @@ function archivedSessionHint(message: string): ProviderRecoveryHint | null { : null; } +function withActiveWriterGuidance(message: string): string { + return CODEX_ACTIVE_WRITER_ERROR_PATTERN.test(message) + ? `${message}. Another Codex process still owns this thread. Close any other Codex session using it; if none is open, wait for a previous Codex process to finish shutting down or stop the leftover codex app-server process, then retry.` + : message; +} + async function delay(ms: number): Promise { await new Promise((resolve) => { setTimeout(resolve, ms); @@ -372,6 +381,7 @@ interface CodexBridgeSession { pendingPreIdentityDeltas: ThreadDelta[]; rebuildBeforeNextTurnReason: string | null; closing: boolean; + previousChildExit: Promise | null; } const sessionsByBbThreadId = new Map(); @@ -399,13 +409,18 @@ function currentSession( return session; } -function releaseSession(session: CodexBridgeSession): void { +function releaseSession(session: CodexBridgeSession): Promise { session.closing = true; if (sessionsByBbThreadId.get(session.bbThreadId) === session) { sessionsByBbThreadId.delete(session.bbThreadId); } - session.connection?.kill(); + const previousChildExit = session.previousChildExit; + session.previousChildExit = null; + const currentChildExit = session.connection?.kill() ?? Promise.resolve(); session.connection = null; + return previousChildExit === null + ? currentChildExit + : Promise.all([previousChildExit, currentChildExit]).then(() => undefined); } const codexProviderOptionsSchema = z @@ -816,6 +831,40 @@ const codexThreadIdentityResultSchema = z .object({ thread: z.object({ id: z.string().min(1) }).passthrough() }) .passthrough(); +async function requestThreadConstructionWithWriterRetry( + connection: CodexAppServerConnection, + method: string, + params: BbThreadStartParams | ThreadResumeParams | BbThreadForkParams, +): Promise> { + const sendOnce = (): Promise< + z.infer + > => + connection.request({ + method, + params, + resultSchema: codexThreadIdentityResultSchema, + timeoutMs: CHILD_REQUEST_TIMEOUT_MS, + }); + for (const [ + retryIndex, + retryDelayMs, + ] of CODEX_ACTIVE_WRITER_RETRY_DELAYS_MS.entries()) { + try { + return await sendOnce(); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (!CODEX_ACTIVE_WRITER_ERROR_PATTERN.test(message)) { + throw error; + } + process.stderr.write( + `codex ${method} found an active rollout writer; retrying in ${retryDelayMs}ms (${retryIndex + 1}/${CODEX_ACTIVE_WRITER_RETRY_DELAYS_MS.length}).\n`, + ); + await delay(retryDelayMs); + } + } + return await sendOnce(); +} + type CodexSessionConstructionRequest = | { kind: "start" } | { kind: "resume"; providerThreadId: string } @@ -843,10 +892,6 @@ async function constructThreadSession( args: ConstructThreadSessionArgs, ): Promise { const existing = sessionsByBbThreadId.get(args.threadId); - if (existing) { - releaseSession(existing); - } - const decoded = decodeCodexOptions(args.options); sessionSerialCounter += 1; const serial = sessionSerialCounter; @@ -884,8 +929,24 @@ async function constructThreadSession( pendingPreIdentityDeltas: [], rebuildBeforeNextTurnReason: null, closing: false, + previousChildExit: null, }; sessionsByBbThreadId.set(args.threadId, session); + if (existing) { + const previousChildExit = releaseSession(existing); + session.previousChildExit = previousChildExit; + await previousChildExit; + if (session.previousChildExit === previousChildExit) { + session.previousChildExit = null; + } + if (session.closing) { + throw new CodexSessionReleasedError( + new Error( + "codex session was released while waiting for the previous app-server to exit", + ), + ); + } + } if (args.request.kind === "resume") { announceSessionIdentity(session, args.request.providerThreadId); } @@ -968,12 +1029,11 @@ async function constructThreadSession( } } - const result = await connection.request({ + const result = await requestThreadConstructionWithWriterRetry( + connection, method, params, - resultSchema: codexThreadIdentityResultSchema, - timeoutMs: CHILD_REQUEST_TIMEOUT_MS, - }); + ); const codexThreadId = result.thread.id; session.codexThreadId = codexThreadId; translator.activateThreadGitWritableRoots({ @@ -1023,6 +1083,7 @@ function registerResumableSession(session: CodexBridgeSession): void { pendingPreIdentityDeltas: [], rebuildBeforeNextTurnReason: null, closing: false, + previousChildExit: null, }); } @@ -1212,8 +1273,9 @@ function sendConstructionError( error: unknown, resumable: boolean, ): void { - const message = describeCodexLaunchError(error); - const recovery = archivedSessionHint(message); + const providerMessage = describeCodexLaunchError(error); + const recovery = archivedSessionHint(providerMessage); + const message = withActiveWriterGuidance(providerMessage); sendError( id, resumable && recovery !== null @@ -1453,7 +1515,7 @@ async function handleThreadStop( if (params.intent === "release") { if (session) { - releaseSession(session); + await releaseSession(session); } sendResult(id, { ok: true }); return; @@ -1509,7 +1571,7 @@ async function handleThreadStop( providerThreadId: session.codexThreadId, }), ); - releaseSession(session); + await releaseSession(session); sendResult(id, { ok: true }); } @@ -1557,11 +1619,11 @@ async function handleThreadMaintenance( alreadyInRequestedState?: RegExp; }, ): Promise { - const settle = (): void => { + const settle = async (): Promise => { if (options?.releaseAfter) { const session = sessionsByBbThreadId.get(params.threadId); if (session) { - releaseSession(session); + await releaseSession(session); } } sendResult(id, { ok: true }); @@ -1570,13 +1632,13 @@ async function handleThreadMaintenance( await withChildForThread(params.threadId, (connection) => sendMaintenanceRequestWithRetries(connection, request), ); - settle(); + await settle(); } catch (error) { if ( error instanceof Error && options?.alreadyInRequestedState?.test(error.message) === true ) { - settle(); + await settle(); return; } rejectWithCodexError(id, error); @@ -1584,8 +1646,9 @@ async function handleThreadMaintenance( } function rejectWithCodexError(id: string | number, error: unknown): void { - const message = describeCodexLaunchError(error); - const recovery = archivedSessionHint(message); + const providerMessage = describeCodexLaunchError(error); + const recovery = archivedSessionHint(providerMessage); + const message = withActiveWriterGuidance(providerMessage); if (recovery !== null) { sendError(id, BRIDGE_JSON_RPC_ERRORS.BRIDGE_ERROR, message, { recovery }); return; diff --git a/plugins/provider-codex/src/bridge/bridge.writer-lock.test.ts b/plugins/provider-codex/src/bridge/bridge.writer-lock.test.ts new file mode 100644 index 0000000000..febfcd5989 --- /dev/null +++ b/plugins/provider-codex/src/bridge/bridge.writer-lock.test.ts @@ -0,0 +1,219 @@ +import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import { experimental_createBridgeJsonRpcTestHarness as createBridgeJsonRpcTestHarness } from "@get-bb/plugin-sdk/provider-bridge/testing"; +import { handleLine } from "./bridge.js"; +import { + cleanupBridgeProcessTest, + spawnedAppServerPids, + waitForAppServerProcessStep, +} from "./bridge-process.test-support.js"; + +const THREAD_ID = "thr_writer_lock_1"; +const PROVIDER_THREAD_ID = "codex-writer-lock-1"; + +const fakeAppServerPath = fileURLToPath( + new URL("./fake-codex-app-server.mjs", import.meta.url), +); + +const sessionOptions = { + permissionMode: "full", + permissionScope: "full", + approvalReviewer: null, + permissionEscalation: null, + reasoningLevel: "low", +} as const; + +const changedSessionOptions = { + ...sessionOptions, + reasoningLevel: "high", +} as const; + +let harness: ReturnType; +let workspaceDir = ""; +let processLogPath = ""; +let writerLockPath = ""; + +beforeEach(() => { + workspaceDir = mkdtempSync(join(tmpdir(), "bb-codex-writer-lock-")); + processLogPath = join(workspaceDir, "app-server-processes.log"); + writerLockPath = join(workspaceDir, "writer.lock"); + const scriptPath = join(workspaceDir, "fake-codex-script.json"); + writeFileSync( + scriptPath, + JSON.stringify({ + processLogPath, + writerLockPath, + sigtermDelayMs: 500, + }), + ); + vi.stubEnv("BB_CODEX_BRIDGE_APP_SERVER_COMMAND", process.execPath); + vi.stubEnv( + "BB_CODEX_BRIDGE_APP_SERVER_ARGS", + JSON.stringify([fakeAppServerPath, scriptPath]), + ); + harness = createBridgeJsonRpcTestHarness(handleLine); +}); + +afterEach(async () => { + await cleanupBridgeProcessTest({ + harness, + cleanupId: 995_001, + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + processLogPath, + workspaceDir, + unstubEnvs: vi.unstubAllEnvs, + }); +}); + +async function resumeThread(id: number): Promise { + harness.sendRequest(id, "thread/resume", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + cwd: workspaceDir, + instructionMode: "append", + options: sessionOptions, + }); + expect((await harness.waitForResponse(id)).error).toBeUndefined(); +} + +it("waits for the previous writer before resuming during a settings rebuild", async () => { + await resumeThread(1); + + harness.sendRequest(2, "turn/start", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + clientRequestId: "creq_abcdefghjk", + input: [{ type: "text", text: "hello", mentions: [] }], + options: changedSessionOptions, + }); + const rebuiltTurn = await harness.waitForResponse(2); + + expect(rebuiltTurn.error).toBeUndefined(); + expect(rebuiltTurn.result).toEqual({ threadId: THREAD_ID }); +}, 30_000); + +it("finishes releasing the writer before acknowledging thread stop", async () => { + await resumeThread(1); + + harness.sendRequest(2, "thread/stop", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + intent: "release", + activeTurnId: null, + }); + expect((await harness.waitForResponse(2)).error).toBeUndefined(); + expect(existsSync(writerLockPath)).toBe(false); + + await resumeThread(3); +}, 30_000); + +it("does not install a replacement after a concurrent release is acknowledged", async () => { + await resumeThread(1); + + harness.sendRequest(2, "turn/start", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + clientRequestId: "creq_abcdefghjk", + input: [{ type: "text", text: "hello", mentions: [] }], + options: changedSessionOptions, + }); + await waitForAppServerProcessStep(processLogPath, "sigterm"); + + harness.sendRequest(3, "thread/stop", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + intent: "release", + activeTurnId: null, + }); + expect((await harness.waitForResponse(3)).error).toBeUndefined(); + expect(existsSync(writerLockPath)).toBe(false); + + const rebuiltTurn = await harness.waitForResponse(2); + expect(rebuiltTurn.error).toBeDefined(); + expect( + harness.messages.filter((message) => message.method === "session/replaced"), + ).toHaveLength(0); + expect(spawnedAppServerPids(processLogPath)).toHaveLength(1); +}, 30_000); + +it("does not install a replacement after concurrent discard maintenance settles", async () => { + await resumeThread(1); + + harness.sendRequest(2, "turn/start", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + clientRequestId: "creq_abcdefghjk", + input: [{ type: "text", text: "hello", mentions: [] }], + options: changedSessionOptions, + }); + await waitForAppServerProcessStep(processLogPath, "sigterm"); + + harness.sendRequest(3, "thread/discard", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + }); + expect((await harness.waitForResponse(3)).result).toEqual({ ok: true }); + expect(existsSync(writerLockPath)).toBe(false); + + const rebuiltTurn = await harness.waitForResponse(2); + expect(rebuiltTurn.error).toBeDefined(); + expect( + harness.messages.filter((message) => message.method === "session/replaced"), + ).toHaveLength(0); + expect(spawnedAppServerPids(processLogPath)).toHaveLength(2); +}, 30_000); + +it("retries a resume while another Codex process is releasing the writer", async () => { + writeFileSync(writerLockPath, String(process.pid)); + const stderrWrite = vi + .spyOn(process.stderr, "write") + .mockImplementation(() => true); + + try { + harness.sendRequest(1, "thread/resume", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + cwd: workspaceDir, + instructionMode: "append", + options: sessionOptions, + }); + await waitForAppServerProcessStep(processLogPath, "writer-conflict"); + rmSync(writerLockPath, { force: true }); + const resumed = await harness.waitForResponse(1); + + expect(resumed.error).toBeUndefined(); + expect(resumed.result).toEqual({ + providerThreadId: PROVIDER_THREAD_ID, + sessionRestorable: true, + }); + expect(stderrWrite).toHaveBeenCalledWith( + "codex thread/resume found an active rollout writer; retrying in 100ms (1/3).\n", + ); + } finally { + stderrWrite.mockRestore(); + rmSync(writerLockPath, { force: true }); + } +}, 30_000); + +it("explains persistent writer contention and resumes after the owner closes", async () => { + writeFileSync(writerLockPath, String(process.pid)); + + harness.sendRequest(1, "thread/resume", { + threadId: THREAD_ID, + providerThreadId: PROVIDER_THREAD_ID, + cwd: workspaceDir, + instructionMode: "append", + options: sessionOptions, + }); + const blocked = await harness.waitForResponse(1); + rmSync(writerLockPath, { force: true }); + + expect(blocked.error?.message).toBe( + `thread ${PROVIDER_THREAD_ID} already has an active writer. Another Codex process still owns this thread. Close any other Codex session using it; if none is open, wait for a previous Codex process to finish shutting down or stop the leftover codex app-server process, then retry.`, + ); + await resumeThread(2); +}, 30_000); diff --git a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs index 74793f9e61..ccfbb42a2b 100644 --- a/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs +++ b/plugins/provider-codex/src/bridge/fake-codex-app-server.mjs @@ -29,6 +29,7 @@ import { existsSync, openSync, readFileSync, + unlinkSync, writeFileSync, } from "node:fs"; import { createInterface } from "node:readline"; @@ -166,6 +167,15 @@ const archivedThreadIds = new Set(); const processLogPath = script?.processLogPath ?? null; /** `startDelayMs`: answer `thread/start` only after this many milliseconds. */ const startDelayMs = script?.startDelayMs ?? 0; +/** + * `writerLockPath`: optional single-writer fixture shared by every fake child + * from one script. A resumed thread owns it until that child exits, mirroring + * Codex's process-lifetime writer lock. + */ +const writerLockPath = script?.writerLockPath ?? null; +/** `sigtermDelayMs`: keep a writer alive briefly after SIGTERM. */ +const sigtermDelayMs = script?.sigtermDelayMs ?? 0; +let ownsWriterLock = false; function logProcessStep(step) { if (processLogPath === null) { @@ -175,9 +185,64 @@ function logProcessStep(step) { } logProcessStep("spawn"); -process.on("SIGTERM", () => { + +function releaseWriterLock() { + if (!ownsWriterLock || writerLockPath === null) { + return; + } + ownsWriterLock = false; + if ( + existsSync(writerLockPath) && + readFileSync(writerLockPath, "utf8") === String(process.pid) + ) { + unlinkSync(writerLockPath); + } +} + +function acquireWriterLock() { + if (writerLockPath === null || ownsWriterLock) { + return true; + } + try { + writeFileSync(writerLockPath, String(process.pid), { flag: "wx" }); + ownsWriterLock = true; + return true; + } catch (error) { + if (!error || typeof error !== "object" || error.code !== "EEXIST") { + throw error; + } + const ownerPid = Number(readFileSync(writerLockPath, "utf8")); + try { + process.kill(ownerPid, 0); + return false; + } catch (ownerError) { + if ( + !ownerError || + typeof ownerError !== "object" || + ownerError.code !== "ESRCH" + ) { + throw ownerError; + } + unlinkSync(writerLockPath); + return acquireWriterLock(); + } + } +} + +function exitCleanly() { + releaseWriterLock(); logProcessStep("exit"); process.exit(0); +} + +process.on("exit", releaseWriterLock); +process.on("SIGTERM", () => { + logProcessStep("sigterm"); + if (sigtermDelayMs > 0) { + setTimeout(exitCleanly, sigtermDelayMs); + return; + } + exitCleanly(); }); let scriptedTurnIndex = 0; @@ -351,6 +416,15 @@ async function handleRequest(message) { return; } case "thread/resume": { + if (!acquireWriterLock()) { + logProcessStep("writer-conflict"); + respondError( + id, + -32603, + `thread ${params.threadId} already has an active writer`, + ); + return; + } // Scripted archived-session rejection: the real app-server refuses to // resume an archived thread with an error naming the session. Tests use // an `archived-` provider-thread-id prefix to trigger it. @@ -552,6 +626,5 @@ stdinLines.on("line", (line) => { } }); stdinLines.on("close", () => { - logProcessStep("exit"); - process.exit(0); + exitCleanly(); });