From 37a4cab907216fcae02b95843d2250f06b73da2b Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Tue, 8 Sep 2026 19:00:41 +0900 Subject: [PATCH 1/2] test(codex): share retained-sync child output across barrier paths --- .../codex-retained-root-serialization.test.ts | 58 ++++++++++++++----- 1 file changed, 42 insertions(+), 16 deletions(-) diff --git a/tests/codex-integration/codex-retained-root-serialization.test.ts b/tests/codex-integration/codex-retained-root-serialization.test.ts index 07f6446582..08ef18f136 100644 --- a/tests/codex-integration/codex-retained-root-serialization.test.ts +++ b/tests/codex-integration/codex-retained-root-serialization.test.ts @@ -124,6 +124,21 @@ function sandboxChildEnv(sandbox: Sandbox): Record { return { ...sandbox.env, ...sandbox.serviceManagerEnv }; } +interface ChildResult { + exitCode: number; + stdout: string; + stderr: string; +} + +/** One consumer per pipe; barrier diagnostics and final assertions share the result. */ +function captureChildResult(child: ReturnType): Promise { + return Promise.all([ + child.exited, + new Response(child.stdout).text(), + new Response(child.stderr).text(), + ]).then(([exitCode, stdout, stderr]) => ({ exitCode, stdout, stderr })); +} + /** * Wait for a child to reach its barrier, failing fast with its output if it exits * first. The exit branch is a REJECTING promise, so while the race is pending an @@ -135,16 +150,30 @@ function sandboxChildEnv(sandbox: Sandbox): Record { * no-op catch attached up front marks that late rejection handled without * changing what the race sees. */ -async function raceBarrier(child: ReturnType, barrier: Promise): Promise { - const exitedEarly = child.exited.then(async exitCode => { - const stdout = await new Response(child.stdout).text(); - const stderr = await new Response(child.stderr).text(); +async function raceBarrier(result: Promise, barrier: Promise): Promise { + const exitedEarly = result.then(({ exitCode, stdout, stderr }) => { throw new Error(`sync exited before provider barrier (${exitCode})\nstdout=${stdout}\nstderr=${stderr}`); }); exitedEarly.catch(() => undefined); await Promise.race([barrier, exitedEarly]); } +test("barrier diagnostics retain both pipes when the child exits first", async () => { + const sandbox = makeSandbox("ocx-retained-early-exit-"); + const child = Bun.spawn([process.execPath, "--eval", ` + process.stdout.write("fixture-stdout\\n"); + process.stderr.write("fixture-stderr\\n"); + process.exitCode = 7; + `], { cwd: repoRoot, env: sandboxChildEnv(sandbox), stdout: "pipe", stderr: "pipe" }); + sandbox.children.add(child); + const result = captureChildResult(child); + + await expect(raceBarrier(result, new Promise(() => {}))).rejects.toThrow( + "sync exited before provider barrier (7)\nstdout=fixture-stdout\n\nstderr=fixture-stderr\n", + ); + expect(await result).toEqual({ exitCode: 7, stdout: "fixture-stdout\n", stderr: "fixture-stderr\n" }); +}, SPAWN_BUDGET_MS); + // A `bun --eval` child on a loaded windows-latest shard takes 8-11 s just to boot and // reach its marker (runs 33590540220 and 33605898170), so a 10 s wait was the coin flip, // not the child. Every caller passes a deadline that sits inside its own test budget so @@ -370,8 +399,9 @@ for (const publisher of ["convergence", "retained"] as const) { console.log(JSON.stringify({ status: response.status, body: await response.json() })); `], sandbox.preloadPath)], { cwd: repoRoot, env: sandboxChildEnv(sandbox), stdout: "pipe", stderr: "pipe" }); sandbox.children.add(sync); + const syncResult = captureChildResult(sync); - await raceBarrier(sync, waitForPath(requested, INTERNAL_DEADLINE_MS)); + await raceBarrier(syncResult, waitForPath(requested, INTERNAL_DEADLINE_MS)); const published = await runPublisher(sandbox, publisher, config); if (published.exitCode !== 0) { throw new Error(`${publisher} publisher failed\nstdout=${published.stdout}\nstderr=${published.stderr}`); @@ -380,11 +410,9 @@ for (const publisher of ["convergence", "retained"] as const) { expect(newer).not.toBe(initial); writeFileSync(release, "release"); - const [exitCode, stdout, stderr] = await Promise.all([ - sync.exited, - new Response(sync.stdout).text(), - new Response(sync.stderr).text(), - ]); + // Exercise the losing exit branch before the successful caller reads output. + await sync.exited; + const { exitCode, stdout, stderr } = await syncResult; expect({ exitCode, stdout, stderr }).toMatchObject({ exitCode: 0 }); expect(readFileSync(catalogPath, "utf8")).toBe(newer); } finally { @@ -447,8 +475,9 @@ test("a persisted runtime selection moved by another process during the await bl console.log(JSON.stringify(await syncCatalogModels(config))); `], sandbox.preloadPath)], { cwd: repoRoot, env: sandboxChildEnv(sandbox), stdout: "pipe", stderr: "pipe" }); sandbox.children.add(sync); + const syncResult = captureChildResult(sync); - await raceBarrier(sync, waitForPath(requested, INTERNAL_DEADLINE_MS)); + await raceBarrier(syncResult, waitForPath(requested, INTERNAL_DEADLINE_MS)); // Another process selects a different Codex runtime. No catalog byte changes. writeFileSync(runtimeStatePath, `${JSON.stringify({ @@ -460,11 +489,8 @@ test("a persisted runtime selection moved by another process during the await bl }, null, 2)}\n`); writeFileSync(release, "release"); - const [exitCode, stdout, stderr] = await Promise.all([ - sync.exited, - new Response(sync.stdout).text(), - new Response(sync.stderr).text(), - ]); + await sync.exited; + const { exitCode, stdout, stderr } = await syncResult; expect({ exitCode, stderr }).toMatchObject({ exitCode: 0 }); expect(JSON.parse(stdout.trim())).toMatchObject({ catalogWritten: false }); expect(readFileSync(catalogPath, "utf8")).toBe(initial); From 4141281b14cc7dad3e3a8b06b727ae4b2ec42ac0 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Tue, 8 Sep 2026 19:56:20 +0900 Subject: [PATCH 2/2] test: bound Windows fixture lifetimes and distinguish stale catalog writes --- .../codex-retained-root-serialization.test.ts | 17 +++- .../server-xai-responses-streaming.test.ts | 82 +++++++++++++++++-- 2 files changed, 87 insertions(+), 12 deletions(-) diff --git a/tests/codex-integration/codex-retained-root-serialization.test.ts b/tests/codex-integration/codex-retained-root-serialization.test.ts index 08ef18f136..f97877ed90 100644 --- a/tests/codex-integration/codex-retained-root-serialization.test.ts +++ b/tests/codex-integration/codex-retained-root-serialization.test.ts @@ -21,6 +21,7 @@ import { removeTreeWithRetry } from "../helpers/remove-tree"; import { repoRoot as resolveRepoRoot } from "../helpers/repo-root"; import { SPAWN_BUDGET_MS } from "../helpers/test-budget"; import { INTERNAL_DEADLINE_MS } from "../helpers/test-budget"; +import { watchdogMs } from "../helpers/ci-watchdog"; const repoRoot = resolveRepoRoot(); const sandboxes: Sandbox[] = []; @@ -367,11 +368,13 @@ for (const publisher of ["convergence", "retained"] as const) { port: 0, fetch: async request => { if (!new URL(request.url).pathname.endsWith("/models")) return new Response("not found", { status: 404 }); - if (requests++ === 0) { + const first = requests++ === 0; + if (first) { writeFileSync(requested, "requested"); while (!existsSync(release)) await Bun.sleep(5); } - return Response.json({ data: [{ id: "race-model" }] }); + // Distinct snapshots make a stale publish observable in the final catalog. + return Response.json({ data: [{ id: first ? "race-model" : "newer-race-model" }] }); }, }); const config = { @@ -401,24 +404,30 @@ for (const publisher of ["convergence", "retained"] as const) { sandbox.children.add(sync); const syncResult = captureChildResult(sync); - await raceBarrier(syncResult, waitForPath(requested, INTERNAL_DEADLINE_MS)); + // This real child imports the management route before reaching /models. + // Keep the CI startup floor, then leave room for the second publisher process. + await raceBarrier(syncResult, waitForPath(requested, watchdogMs(INTERNAL_DEADLINE_MS))); const published = await runPublisher(sandbox, publisher, config); if (published.exitCode !== 0) { throw new Error(`${publisher} publisher failed\nstdout=${published.stdout}\nstderr=${published.stderr}`); } const newer = readFileSync(catalogPath, "utf8"); expect(newer).not.toBe(initial); + const newerSlugs = JSON.parse(newer).models.map((model: { slug: string }) => model.slug); + expect(newerSlugs).toContain("fixture/newer-race-model"); + expect(newerSlugs).not.toContain("fixture/race-model"); writeFileSync(release, "release"); // Exercise the losing exit branch before the successful caller reads output. await sync.exited; const { exitCode, stdout, stderr } = await syncResult; expect({ exitCode, stdout, stderr }).toMatchObject({ exitCode: 0 }); + expect(JSON.parse(stdout).status).toBe(200); expect(readFileSync(catalogPath, "utf8")).toBe(newer); } finally { provider.stop(true); } - }, SPAWN_BUDGET_MS); + }, SPAWN_BUDGET_MS * 2); } /** diff --git a/tests/server/server-xai-responses-streaming.test.ts b/tests/server/server-xai-responses-streaming.test.ts index 316ee56b91..8a2298972c 100644 --- a/tests/server/server-xai-responses-streaming.test.ts +++ b/tests/server/server-xai-responses-streaming.test.ts @@ -12,6 +12,7 @@ import { startServer } from "../../src/server"; import type { OcxConfig } from "../../src/types"; import { installIsolatedCodexHome, type IsolatedCodexHome } from "../helpers/isolated-codex-home"; import { removeTreeWithRetry } from "../helpers/remove-tree"; +import { SERVER_BUDGET_MS } from "../helpers/test-budget"; const RESPONSES_ENDPOINT = `${XAI_GROK_CLI_BASE_URL}/responses`; const encoder = new TextEncoder(); @@ -20,8 +21,34 @@ let testDir = ""; let previousHome: string | undefined; let isolatedCodexHome: IsolatedCodexHome | null = null; let originalFetch: typeof fetch; +let activeRoutedCase: { controller: AbortController; settled: Promise } | null = null; + +function runRoutedCase(body: (signal: AbortSignal) => Promise): Promise { + const controller = new AbortController(); + const result = body(controller.signal); + // Observe the entire body, including its server-stop finally, even after a test timeout. + activeRoutedCase = { controller, settled: result.then(() => {}, () => {}) }; + return result; +} + +async function drainRoutedCase(): Promise { + const active = activeRoutedCase; + if (!active) return; + active.controller.abort(new DOMException("xAI fixture cleanup", "AbortError")); + await active.settled; + if (activeRoutedCase === active) activeRoutedCase = null; +} + +function startXaiTestServer() { + return startServer(0, { + // This wire fixture does not exercise native Codex service ownership. Avoid + // unrelated Windows service queries and native-main recovery during setup. + inspectNativeCodexOwnership: () => ({ ownership: "foreign", reason: "xAI wire fixture" }), + }); +} beforeEach(async () => { + if (activeRoutedCase) throw new Error("previous routed-parent fixture has not finished cleanup"); originalFetch = globalThis.fetch; previousHome = process.env.OPENCODEX_HOME; isolatedCodexHome = installIsolatedCodexHome("ocx-xai-responses-codex-"); @@ -36,14 +63,15 @@ beforeEach(async () => { }); }); -afterEach(() => { +afterEach(async () => { + await drainRoutedCase(); globalThis.fetch = originalFetch; if (previousHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = previousHome; isolatedCodexHome?.restore(); isolatedCodexHome = null; if (testDir) removeTreeWithRetry(testDir); -}); +}, SERVER_BUDGET_MS); function config(): OcxConfig { return { @@ -72,7 +100,43 @@ function sse(payload: unknown): Uint8Array { } describe("xAI OAuth Responses streaming opt-in", () => { - test.each([true, false])("continues a routed parent after a string child result (stream=%s)", async stream => { + test("routed-case cleanup waits for the entire aborted body finally", async () => { + let markFinally!: () => void; + const enteredFinally = new Promise(resolve => { markFinally = resolve; }); + let releaseFinally!: () => void; + const finallyGate = new Promise(resolve => { releaseFinally = resolve; }); + let finallyFinished = false; + const running = runRoutedCase(async signal => { + try { + await new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + }); + } finally { + markFinally(); + await finallyGate; + finallyFinished = true; + } + }); + const outcome = running.then(() => null, (error: unknown) => error); + let drained = false; + const draining = drainRoutedCase().then(() => { drained = true; }); + try { + await enteredFinally; + await Promise.resolve(); + // Awaiting outcome first would hide a drain helper that returned too early. + expect(drained).toBe(false); + expect(finallyFinished).toBe(false); + } finally { + releaseFinally(); + await draining; + await outcome; + } + expect(await outcome).toMatchObject({ name: "AbortError" }); + expect(finallyFinished).toBe(true); + expect(activeRoutedCase).toBeNull(); + }, SERVER_BUDGET_MS); + + test.each([true, false])("continues a routed parent after a string child result (stream=%s)", stream => runRoutedCase(async signal => { const captured: Array> = []; let privateItemRejections = 0; const childText = " Synthetic worker result\nAll requested observations returned.\n "; @@ -112,9 +176,11 @@ describe("xAI OAuth Responses streaming opt-in", () => { }) as typeof fetch; saveConfig({ ...config(), multiAgentMode: "v2" }); - const server = startServer(0); + const server = startXaiTestServer(); const send = async (session: string, input: unknown[], parentSession?: string) => { + signal.throwIfAborted(); const response = await originalFetch(new URL("/v1/responses", server.url), { + signal, method: "POST", headers: { "content-type": "application/json", "session-id": session, ...(parentSession ? { "x-codex-parent-thread-id": parentSession } : {}), }, @@ -163,7 +229,7 @@ describe("xAI OAuth Responses streaming opt-in", () => { } finally { await server.stop(true); } - }, 10_000); + }), 10_000); test("uses the native Responses wire and relays the first delta before completion", async () => { let releaseCompletion!: () => void; @@ -258,7 +324,7 @@ describe("xAI OAuth Responses streaming opt-in", () => { }) as typeof fetch; saveConfig(config()); - const server = startServer(0); + const server = startXaiTestServer(); let reader: ReadableStreamDefaultReader | undefined; try { const response = await originalFetch(new URL("/v1/responses", server.url), { @@ -373,7 +439,7 @@ describe("xAI OAuth Responses streaming opt-in", () => { }) as typeof fetch; saveConfig(config()); - const server = startServer(0); + const server = startXaiTestServer(); try { const response = await originalFetch(new URL("/v1/responses", server.url), { method: "POST", @@ -468,7 +534,7 @@ describe("xAI OAuth Responses streaming opt-in", () => { }) as typeof fetch; saveConfig(config()); - const server = startServer(0); + const server = startXaiTestServer(); try { const response = await originalFetch(new URL("/v1/responses", server.url), { method: "POST",