diff --git a/src/codex/prompt-text-probe.ts b/src/codex/prompt-text-probe.ts index 6db8d1e49e..6282050b7a 100644 --- a/src/codex/prompt-text-probe.ts +++ b/src/codex/prompt-text-probe.ts @@ -100,41 +100,236 @@ function resolveCodexBinary(): string | null { /** 8 MiB is far above any real prompt and far below anything that hurts the server. */ const MAX_PROBE_OUTPUT_BYTES = 8 * 1024 * 1024; -function runProbe(binary: string, cwd: string, timeoutMs: number): Promise { - return new Promise(resolve => { +interface ProbeCommand { + binary: string; + args: string[]; + cwd: string; + timeoutMs: number; + promptRevision: string | null; +} + +interface PromptProbeFlight { + key: string; + controller: AbortController; + result: Promise; + closed: Promise; + waiters: number; + joinable: boolean; + resultSettled: boolean; + settled: boolean; +} + +interface PromptProbeExecution { + result: Promise; + closed: Promise; +} + +type SharedPromptProbeOutcome = + | { kind: "output"; raw: string } + | { kind: "failed" } + | { kind: "busy" }; + +let activePromptProbe: PromptProbeFlight | null = null; +let probeCommandForTests: { binary: string; args: string[] } | null = null; +let probeSpawnAttemptsForTests = 0; +let probeCloseBarrierForTests: Promise | null = null; + +function commandKey(command: ProbeCommand): string { + return JSON.stringify([ + command.binary, + command.args, + command.cwd, + command.timeoutMs, + command.promptRevision, + ]); +} + +function completedExecution(value: string | null): PromptProbeExecution { + return { result: Promise.resolve(value), closed: Promise.resolve() }; +} + +function runProbe( + command: ProbeCommand, + signal: AbortSignal, + onStopping: () => void, +): PromptProbeExecution { + if (signal.aborted) return completedExecution(null); + let resolveResult!: (value: string | null) => void; + let resolveClosed!: () => void; + const result = new Promise(resolve => { resolveResult = resolve; }); + const closed = new Promise(resolve => { resolveClosed = resolve; }); + let resultSettled = false; + let closeSettled = false; + + const finishResult = (value: string | null) => { + if (resultSettled) return; + resultSettled = true; + resolveResult(value); + }; + const finishClosed = () => { + if (closeSettled) return; + closeSettled = true; + resolveClosed(); + }; + + try { // A probe must never hang OR balloon the management API: it is bounded in // time AND in bytes, and every failure degrades to "unavailable" rather than // an error page. - const child = spawn(binary, ["debug", "prompt-input"], { - cwd, - stdio: ["ignore", "pipe", "ignore"], - }); + let child: ReturnType; + try { + if (probeCommandForTests) probeSpawnAttemptsForTests += 1; + child = spawn(command.binary, command.args, { + cwd: command.cwd, + stdio: ["ignore", "pipe", "ignore"], + }); + } catch { + finishResult(null); + finishClosed(); + return { result, closed }; + } const chunks: Buffer[] = []; let size = 0; let settled = false; - // One settlement path: a timeout that resolved before `close` used to leave - // the child streaming into a buffer nobody would ever read. - const settle = (value: string | null) => { + let stopping = false; + let timer: ReturnType | undefined; + + const finish = (value: string | null) => { if (settled) return; settled = true; - clearTimeout(timer); + if (timer) clearTimeout(timer); + signal.removeEventListener("abort", onAbort); + finishResult(value); + finishClosed(); + }; + + // Keep the flight admitted until `close`: kill() only requests termination + // and does not prove the exact child has released its process and stdio. + const terminate = () => { + if (settled || stopping) return; + stopping = true; + onStopping(); + if (timer) clearTimeout(timer); + signal.removeEventListener("abort", onAbort); child.stdout?.destroy(); - if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL"); - resolve(value); + // The caller is bounded even if OS termination later fails. Admission is + // retained separately by `closed`, and later probes fail soft while this + // exact child remains unproven terminal. + finishResult(null); + if (child.exitCode !== null || child.signalCode !== null) return; + try { + child.kill("SIGKILL"); + } catch { + // Exact child state is ambiguous. Keep admission non-joinable until its + // own `close` proves terminal instead of targeting a reusable numeric PID. + } }; - const timer = setTimeout(() => settle(null), timeoutMs); + const onAbort = () => terminate(); + + timer = setTimeout(terminate, command.timeoutMs); + signal.addEventListener("abort", onAbort, { once: true }); child.stdout?.on("data", (chunk: Buffer) => { size += chunk.length; - if (size > MAX_PROBE_OUTPUT_BYTES) { settle(null); return; } + if (size > MAX_PROBE_OUTPUT_BYTES) { terminate(); return; } chunks.push(chunk); }); - child.on("error", () => settle(null)); + child.on("error", () => { + // No PID means spawn itself failed, so there is no live child to drain. + if (child.pid === undefined) { + finish(null); + } + else terminate(); + }); child.on("close", code => { // Decode once, at the end: `String(chunk)` per chunk corrupts any UTF-8 // character that straddles a chunk boundary. - settle(code === 0 ? Buffer.concat(chunks).toString("utf8") : null); + const recordClose = () => { + finish(!stopping && code === 0 ? Buffer.concat(chunks).toString("utf8") : null); + }; + const barrier = probeCloseBarrierForTests; + if (barrier) void barrier.then(recordClose, recordClose); + else recordClose(); }); + // Close the race between the pre-spawn check and listener registration. + if (signal.aborted) terminate(); + } catch { + finishResult(null); + finishClosed(); + } + return { result, closed }; +} + +function startPromptProbeFlight(command: ProbeCommand): PromptProbeFlight { + const controller = new AbortController(); + const flight: PromptProbeFlight = { + key: commandKey(command), + controller, + result: Promise.resolve(null), + closed: Promise.resolve(), + waiters: 0, + joinable: true, + resultSettled: false, + settled: false, + }; + const execution = runProbe(command, controller.signal, () => { + flight.joinable = false; }); + flight.result = execution.result + .catch(() => null) + .finally(() => { + flight.resultSettled = true; + }); + flight.closed = execution.closed + .finally(() => { + flight.settled = true; + if (activePromptProbe === flight) activePromptProbe = null; + }); + activePromptProbe = flight; + return flight; +} + +async function runSharedPromptProbe( + command: ProbeCommand, + signal?: AbortSignal, +): Promise { + const key = commandKey(command); + if (signal?.aborted) return { kind: "failed" }; + const active = activePromptProbe; + if (!active) { + const raw = await waitForPromptProbeFlight(startPromptProbeFlight(command), signal); + return raw === null ? { kind: "failed" } : { kind: "output", raw }; + } + if (active.key === key && active.joinable && !active.controller.signal.aborted) { + const raw = await waitForPromptProbeFlight(active, signal); + return raw === null ? { kind: "failed" } : { kind: "output", raw }; + } + // A different or terminating flight still owns the sole process slot. Never + // wait unboundedly for an unproven close and never launch beside it. + return { kind: "busy" }; +} + +async function waitForPromptProbeFlight(flight: PromptProbeFlight, signal?: AbortSignal): Promise { + if (signal?.aborted) { + if (flight.waiters === 0 && !flight.settled) flight.controller.abort(); + return null; + } + flight.waiters += 1; + let onAbort: (() => void) | undefined; + try { + if (!signal) return await flight.result; + const aborted = new Promise(resolve => { + onAbort = () => resolve(null); + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) onAbort(); + }); + return await Promise.race([flight.result, aborted]); + } finally { + if (onAbort) signal?.removeEventListener("abort", onAbort); + flight.waiters = Math.max(0, flight.waiters - 1); + if (flight.waiters === 0 && !flight.resultSettled) { + flight.controller.abort(new DOMException("All prompt probe callers cancelled", "AbortError")); + } + } } /** Pull every `...` section out of the rendered developer messages. */ @@ -182,20 +377,44 @@ export const extractSectionsForTests = extractSections; * `cwd` matters: AGENTS.md and environment context are directory-dependent, so a * probe from the wrong place would describe a prompt the user never sees. */ -export async function probePromptText(timeoutMs = 15_000): Promise { +export async function probePromptText( + timeoutMs = 15_000, + signal?: AbortSignal, + promptRevision: string | null = null, +): Promise { // The probe runs in CODEX_HOME, never in a caller-supplied directory. A `cwd` // parameter let an authenticated request read any readable folder's AGENTS.md, // and it also described a prompt that depends on where Codex happened to run. // The global home is the one context this page can honestly report on. const codexHome = resolveCodexHomeDir(); - const binary = resolveCodexBinary(); + if (signal?.aborted) { + return { ok: false, codexHome, layers: {}, detail: "prompt probe cancelled" }; + } + const binary = probeCommandForTests?.binary ?? resolveCodexBinary(); if (!binary) { return { ok: false, codexHome, layers: {}, detail: "codex binary not found" }; } - const raw = await runProbe(binary, codexHome, timeoutMs); - if (raw === null) { - return { ok: false, codexHome, layers: {}, detail: "codex debug prompt-input failed" }; + const command: ProbeCommand = { + binary, + args: probeCommandForTests?.args ?? ["debug", "prompt-input"], + cwd: codexHome, + timeoutMs, + promptRevision, + }; + const outcome = await runSharedPromptProbe(command, signal); + if (outcome.kind !== "output") { + return { + ok: false, + codexHome, + layers: {}, + detail: signal?.aborted + ? "prompt probe cancelled" + : outcome.kind === "busy" + ? "another prompt probe is still finishing; retry shortly" + : "codex debug prompt-input failed", + }; } + const raw = outcome.raw; const sections = extractSections(raw); if (sections.size === 0) { // Zero sections from a zero-exit probe means the output did not parse, which @@ -236,3 +455,35 @@ export async function probePromptText(timeoutMs = 15_000): Promise | null): void { + probeCloseBarrierForTests = barrier; +} + +/** Test-only fail-closed drain so one failed lifecycle case cannot poison another. */ +export async function resetPromptTextProbeForTests(): Promise { + const active = activePromptProbe; + if (active && !active.settled) { + active.controller.abort(new DOMException("Prompt probe test reset", "AbortError")); + const drained = await Promise.race([ + active.closed.then(() => true), + Bun.sleep(2_000).then(() => false), + ]); + if (!drained) throw new Error("prompt probe child did not close during test reset"); + } + if (activePromptProbe === active) activePromptProbe = null; + probeCommandForTests = null; + probeSpawnAttemptsForTests = 0; + probeCloseBarrierForTests = null; +} diff --git a/src/server/management/codex-prompt-routes.ts b/src/server/management/codex-prompt-routes.ts index 82a4b42540..3c8c8614e6 100644 --- a/src/server/management/codex-prompt-routes.ts +++ b/src/server/management/codex-prompt-routes.ts @@ -320,7 +320,11 @@ export async function handleCodexPromptRoutes(ctx: ManagementContext): Promise boolean, detail: string): Promise { + const deadline = Date.now() + 5_000; + while (!predicate()) { + if (Date.now() >= deadline) throw new Error(`timed out waiting for ${detail}`); + await Bun.sleep(10); + } +} + +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + /** * Sentinels must survive every verb. The decoy is installed as CODEX_HOME for the * duration of each request, so this is not a vacuous check: a regression that @@ -117,7 +139,8 @@ async function revision(fx: Fixture): Promise { return res.body.revision as string; } -afterEach(() => { +afterEach(async () => { + await resetPromptTextProbeForTests(); while (roots.length) rmSync(roots.pop()!, { recursive: true, force: true }); }); @@ -787,7 +810,7 @@ describe("020 coverage completions", () => { const probe = await Bun.file(new URL("../src/codex/prompt-text-probe.ts", import.meta.url)).text(); // The probe resolves CODEX_HOME itself; it must not accept a directory. expect(probe).toContain("resolveCodexHomeDir()"); - expect(probe).toMatch(/export async function probePromptText\(timeoutMs/); + expect(probe).toMatch(/export async function probePromptText\(\s*timeoutMs/); }); test("26. the probe is bounded in bytes as well as in time", async () => { @@ -799,6 +822,80 @@ describe("020 coverage completions", () => { // Decoding per chunk corrupts UTF-8 that straddles a chunk boundary. expect(probe).toContain("Buffer.concat(chunks).toString(\"utf8\")"); }); + + test("27. the text route forwards live request cancellation to its exact child", async () => { + const fx = fixture(""); + const pidPath = join(fx.decoyHome, "probe-pid.txt"); + setPromptTextProbeCommandForTests({ + binary: process.execPath, + args: ["-e", [ + `require("node:fs").writeFileSync(${JSON.stringify(pidPath)}, String(process.pid));`, + "setInterval(() => {}, 1_000);", + ].join("")], + }); + const controller = new AbortController(); + const url = new URL("http://127.0.0.1:10100/api/codex-prompt/text"); + const req = new Request(url, { + method: "GET", + headers: { host: "127.0.0.1:10100" }, + signal: controller.signal, + }); + const previousHome = process.env.CODEX_HOME; + process.env.CODEX_HOME = fx.decoyHome; + let res: Response | null = null; + try { + const pending = handleManagementAPI(req, url, config, { + codexPromptPaths: { configPath: fx.configPath, storePath: fx.storePath, baseVariantDir: fx.baseVariantDir }, + }, "gui-session"); + await waitUntil(() => existsSync(pidPath), "route probe child pid"); + controller.abort(); + res = await pending; + } finally { + if (previousHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = previousHome; + } + + expect(res?.status).toBe(200); + expect(await res?.json()).toMatchObject({ ok: false, detail: "prompt probe cancelled" }); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(1); + const pid = Number(readFileSync(pidPath, "utf8")); + await waitUntil(() => !isProcessAlive(pid), "route probe child exit"); + expectDecoyUntouched(fx); + }); + + test("28. a post-write text read never joins a pre-write probe", async () => { + const fx = fixture("include_apps_instructions = false\n"); + const startedPath = join(fx.decoyHome, "revision-probe-started.txt"); + const probeOutput = JSON.stringify([{ + type: "message", + role: "developer", + content: [{ type: "input_text", text: "Skill text." }], + }]); + setPromptTextProbeCommandForTests({ + binary: process.execPath, + args: ["-e", [ + `require("node:fs").writeFileSync(${JSON.stringify(startedPath)}, "started");`, + `setTimeout(() => process.stdout.write(${JSON.stringify(probeOutput)}), 200);`, + ].join("")], + }); + + const beforeWrite = call("GET", "/api/codex-prompt/text", fx); + await waitUntil(() => existsSync(startedPath), "pre-write probe start"); + writeFileSync(fx.configPath, "include_apps_instructions = true\n", "utf8"); + + const afterWrite = await call("GET", "/api/codex-prompt/text", fx); + expect(afterWrite.body).toMatchObject({ + ok: false, + detail: "another prompt probe is still finishing; retry shortly", + }); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(1); + expect((await beforeWrite).body.ok).toBe(true); + + const fresh = await call("GET", "/api/codex-prompt/text", fx); + expect(fresh.body.ok).toBe(true); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(2); + }); + test("24. every ownership state is named, not collapsed into a boolean", async () => { // developerInstructionsOwned:false covers an ABSENT key and an EXTERNAL one, and // a GUI that cannot tell them apart hides its own create affordance from every diff --git a/tests/codex-prompt-text-probe.test.ts b/tests/codex-prompt-text-probe.test.ts index 11f453a6c8..69c4036e34 100644 --- a/tests/codex-prompt-text-probe.test.ts +++ b/tests/codex-prompt-text-probe.test.ts @@ -6,13 +6,58 @@ * that a missing body is attributed to the right cause, because the dialog shows * that attribution to a user as an explanation. */ -import { describe, expect, test } from "bun:test"; -import { extractSectionsForTests } from "../src/codex/prompt-text-probe"; +import { afterEach, describe, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + extractSectionsForTests, + probePromptText, + promptTextProbeSpawnAttemptsForTests, + resetPromptTextProbeForTests, + setPromptTextProbeCloseBarrierForTests, + setPromptTextProbeCommandForTests, +} from "../src/codex/prompt-text-probe"; + +const lifecycleRoots: string[] = []; +const VALID_PROBE_OUTPUT = JSON.stringify([{ + type: "message", + role: "developer", + content: [{ type: "input_text", text: "Skill text." }], +}]); function message(text: string): string { return JSON.stringify([{ type: "message", role: "developer", content: [{ type: "input_text", text }] }]); } +async function waitUntil(predicate: () => boolean, detail: string): Promise { + const deadline = Date.now() + 5_000; + while (!predicate()) { + if (Date.now() >= deadline) throw new Error(`timed out waiting for ${detail}`); + await Bun.sleep(10); + } +} + +function isProcessAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +function root(): string { + const path = mkdtempSync(join(tmpdir(), "ocx-prompt-probe-")); + lifecycleRoots.push(path); + return path; +} + +afterEach(async () => { + await resetPromptTextProbeForTests(); + while (lifecycleRoots.length) rmSync(lifecycleRoots.pop()!, { recursive: true, force: true }); +}); + describe("section extraction", () => { test("a tag name containing a space is still matched", () => { // Codex renders ``, with a space. A [a-z_]+ pattern @@ -68,3 +113,113 @@ describe("section extraction", () => { expect(sections.get("__agents_md")).toContain("
"); }); }); + +describe("prompt probe process lifecycle", () => { + test("a pre-aborted caller starts no child", async () => { + const marker = join(root(), "started.txt"); + setPromptTextProbeCommandForTests({ + binary: process.execPath, + args: ["-e", `require("node:fs").writeFileSync(${JSON.stringify(marker)}, "started")`], + }); + const controller = new AbortController(); + controller.abort(); + + const result = await probePromptText(2_000, controller.signal); + + expect(result.ok).toBe(false); + expect(result.detail).toBe("prompt probe cancelled"); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(0); + expect(existsSync(marker)).toBe(false); + }); + + test("concurrent callers share one child and one caller may cancel", async () => { + const started = join(root(), "started.txt"); + const source = [ + `require("node:fs").appendFileSync(${JSON.stringify(started)}, "1\\n");`, + `setTimeout(() => process.stdout.write(${JSON.stringify(VALID_PROBE_OUTPUT)}), 150);`, + ].join(""); + setPromptTextProbeCommandForTests({ binary: process.execPath, args: ["-e", source] }); + const controller = new AbortController(); + + const first = probePromptText(2_000, controller.signal); + const second = probePromptText(2_000); + controller.abort(); + + expect((await first).detail).toBe("prompt probe cancelled"); + expect((await second).ok).toBe(true); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(1); + expect(readFileSync(started, "utf8").trim().split(/\r?\n/)).toHaveLength(1); + }); + + test("the last cancellation drains the exact child before another command starts", async () => { + const dir = root(); + const pidPath = join(dir, "pid.txt"); + const overlapPath = join(dir, "overlap.txt"); + const hangingSource = [ + `require("node:fs").writeFileSync(${JSON.stringify(pidPath)}, String(process.pid));`, + "setInterval(() => {}, 1_000);", + ].join(""); + setPromptTextProbeCommandForTests({ binary: process.execPath, args: ["-e", hangingSource] }); + const controller = new AbortController(); + const hanging = probePromptText(5_000, controller.signal); + await waitUntil(() => existsSync(pidPath), "hanging child pid"); + const pid = Number(readFileSync(pidPath, "utf8")); + expect(isProcessAlive(pid)).toBe(true); + + controller.abort(); + expect((await hanging).detail).toBe("prompt probe cancelled"); + + const replacementSource = [ + `const fs = require("node:fs"); const pid = Number(fs.readFileSync(${JSON.stringify(pidPath)}, "utf8"));`, + "let priorProbeAlive = true;", + "try { process.kill(pid, 0); } catch { priorProbeAlive = false; }", + `if (priorProbeAlive) fs.writeFileSync(${JSON.stringify(overlapPath)}, "overlap");`, + `process.stdout.write(${JSON.stringify(VALID_PROBE_OUTPUT)});`, + ].join(""); + setPromptTextProbeCommandForTests({ binary: process.execPath, args: ["-e", replacementSource] }); + const blockedDuringDrain = await probePromptText(2_000); + + expect(blockedDuringDrain.ok).toBe(false); + expect(blockedDuringDrain.detail).toBe("another prompt probe is still finishing; retry shortly"); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(1); + await resetPromptTextProbeForTests(); + setPromptTextProbeCommandForTests({ binary: process.execPath, args: ["-e", replacementSource] }); + const replacement = await probePromptText(2_000); + + expect(replacement.ok).toBe(true); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(1); + expect(existsSync(overlapPath)).toBe(false); + await waitUntil(() => !isProcessAlive(pid), "cancelled child exit"); + }); + + test("admission stays occupied between child exit and close handling", async () => { + const pidPath = join(root(), "exited-parent-pid.txt"); + let releaseClose!: () => void; + setPromptTextProbeCloseBarrierForTests(new Promise(resolve => { releaseClose = resolve; })); + const delayedCloseSource = [ + `const fs = require("node:fs");`, + `fs.writeFileSync(${JSON.stringify(pidPath)}, String(process.pid));`, + `process.stdout.write(${JSON.stringify(VALID_PROBE_OUTPUT)});`, + ].join(""); + setPromptTextProbeCommandForTests({ binary: process.execPath, args: ["-e", delayedCloseSource] }); + const first = probePromptText(2_000); + await waitUntil(() => existsSync(pidPath), "exit-close parent pid"); + const pid = Number(readFileSync(pidPath, "utf8")); + await waitUntil(() => !isProcessAlive(pid), "probe parent exit"); + + setPromptTextProbeCommandForTests({ + binary: process.execPath, + args: ["-e", `process.stdout.write(${JSON.stringify(VALID_PROBE_OUTPUT)})`], + }); + const blockedBeforeClose = await probePromptText(2_000); + + expect(blockedBeforeClose.ok).toBe(false); + expect(blockedBeforeClose.detail).toBe("another prompt probe is still finishing; retry shortly"); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(1); + releaseClose(); + expect((await first).ok).toBe(true); + const afterClose = await probePromptText(2_000); + expect(afterClose.ok).toBe(true); + expect(promptTextProbeSpawnAttemptsForTests()).toBe(2); + }); +});