vrr/openclaw
2
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-09-01 19:25:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 20

fix(codex): reap orphaned app-servers before reconnect (#132610)

Browse source 
* fix(codex): reap orphaned app-servers before reconnect

* style(codex): avoid returning the containment timer handle

* fix(codex): inspect Linux children without requiring procps

* fix(codex): tolerate processes exiting during procfs reads

* test(codex): normalize socket fixture binary frames

* fix(codex): preserve startup errors during process registration
This commit is contained in:
Peter Steinberger 2026-08-29 08:53:41 -07:00 • committed by GitHub
parent c0d853b844
commit 10ee252b56
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 1326 additions and 555 deletions
Download patch file Download diff file Expand all files Collapse all files

33
docs/plugins/codex-harness-runtime.md
View file

@ -69,6 +69,39 @@ marked `catalogMode: "direct-only"` use `openclaw_direct`, which Codex keeps
directly model-visible as `DirectModelOnly` instead of exposing it to nested
Code Mode execution.
## Recovery after a hard Gateway stop
On POSIX systems, OpenClaw checks for registered orphaned Codex app-server
processes before spawning each fresh stdio child. This runs when the connection
is needed, not necessarily at Gateway boot. OpenClaw records the parent and
child process identities in the current state directory's SQLite plugin store
before sending Codex `initialize`, so a child cannot start a native turn before
its registration is durable.
Cleanup only targets a registered child whose original OpenClaw parent is no
longer running. It checks process IDs, start times, and process groups before
terminating the orphan and its discoverable descendants. Another live OpenClaw
instance, processes registered under another state directory, and externally
managed WebSocket or Unix-socket app-servers are left alone. These portable
process checks do not provide an atomic operating-system ownership guarantee
or discover descendants that independently reparented before inspection.
Linux reads process identities directly from `/proc`, including the boot ID
and process start ticks, so Alpine/BusyBox installations do not need `procps`.
macOS uses its native `ps` with a fixed locale and timezone.
If process inspection or bounded cleanup cannot confirm that the registered
orphan is gone, the new stdio connection fails instead of spawning another
child. Follow the reported action: check `/proc` access on Linux or `ps` on
macOS, or verify and stop
the reported orphan process, then retry. If the cleanup budget expires, retry
to finish the remaining registrations.
This recovery requires a spawn-time registration. It does not discover
unregistered children left by an older OpenClaw version or scan command names
to infer ownership. Windows does not yet have equivalent orphan registration
and recovery.
## Thread bindings and model changes
When an OpenClaw session is attached to an existing Codex thread, the next

34
extensions/codex/src/app-server/attempt-startup-retry.test.ts
View file

@ -1,4 +1,6 @@
import * as childProcess from "node:child_process";
import { randomUUID } from "node:crypto";
import { once } from "node:events";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
@ -27,6 +29,11 @@ import {
releaseLeasedSharedCodexAppServerClient,
} from "./shared-client.js";
import { createCodexTestModel } from "./test-support.js";
import * as processSnapshot from "./transport-process-snapshot.js";
vi.mock("node:child_process", async (importOriginal) => ({
...(await importOriginal<typeof import("node:child_process")>()),
}));
vi.mock("./desktop-generation.js", () => ({
isCodexDesktopGenerationCurrent: () => false,
@ -172,10 +179,35 @@ describe("Codex app-server startup retry", () => {
tempRoots.clear();
});
it("retries a real app-server after transient sqlite state initialization failure", async () => {
it("retries a real app-server that fails sqlite initialization before registration completes", async (ctx) => {
const fixture = await createStartupFailureFixture("transient");
let firstChildExit: Promise<unknown> | undefined;
const spawn = childProcess.spawn;
const snapshot = processSnapshot.readCodexAppServerProcessSnapshot;
const spawnSpy = vi.spyOn(childProcess, "spawn").mockImplementation((...args) => {
const child = spawn(...args);
if (
Array.isArray(args[1]) &&
args[1].includes(path.join(fixture.root, "startup-failure.mjs"))
) {
firstChildExit ??= once(child, "exit");
}
return child;
});
const snapshotSpy = vi
.spyOn(processSnapshot, "readCodexAppServerProcessSnapshot")
.mockImplementation(async (...args) => {
// A slow inspector must not replace the child's retryable startup error.
await firstChildExit;
return await snapshot(...args);
});
ctx.onTestFinished(() => {
spawnSpy.mockRestore();
snapshotSpy.mockRestore();
});
const result = await startFixtureAttempt(fixture);
expect(firstChildExit).toBeDefined();
expect(result.thread.threadId).toBe("thread-recovered");
expect(await fs.readFile(fixture.spawnCountPath, "utf8")).toBe("2");
result.turnRoute.release();

22
extensions/codex/src/app-server/attempt-startup.test.ts
View file

@ -161,7 +161,7 @@ function startThreadWithHarness(
const paths = overrides?.paths ?? createAttemptPaths();
const startSpy = overrides?.skipStartSpy
? undefined
: vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
: vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const effectivePluginConfig = overrides?.pluginConfig ?? pluginConfig;
const run = startCodexAttemptThread({
@ -383,8 +383,8 @@ describe("startCodexAttemptThread", () => {
};
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const paths = createAttemptPaths();
let persistedComputerUse = false;
const { run } = startThreadWithHarness(10_000, new AbortController().signal, {
@ -455,8 +455,8 @@ describe("startCodexAttemptThread", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
desktopGeneration.current = { epoch: 1, fingerprint: "desktop-x" };
if (changeStage === "Computer Use readiness") {
computerUseReadinessFailure.next = Object.assign(new Error("desktop selection changed"), {
@ -511,7 +511,7 @@ describe("startCodexAttemptThread", () => {
it("retires the startup generation when context restart sees a new executable owner", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const paths = createAttemptPaths();
const { run } = startThreadWithHarness(5_000, new AbortController().signal, {
harness,
@ -547,8 +547,8 @@ describe("startCodexAttemptThread", () => {
const replacement = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(retained.client)
.mockReturnValueOnce(replacement.client);
.mockResolvedValueOnce(retained.client)
.mockResolvedValueOnce(replacement.client);
const appServer = resolveCodexAppServerRuntimeOptions({ pluginConfig });
const paths = createAttemptPaths();
@ -610,7 +610,7 @@ describe("startCodexAttemptThread", () => {
it("closes indeterminate thread startup even when another lease shares the app-server", async () => {
const retained = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(retained.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(retained.client);
const appServer = resolveCodexAppServerRuntimeOptions({ pluginConfig });
const paths = createAttemptPaths();
@ -798,7 +798,7 @@ describe("startCodexAttemptThread", () => {
const clients = [createClientHarness(), createClientHarness(), createClientHarness()];
const start = vi.spyOn(CodexAppServerClient, "start");
for (const harness of clients) {
start.mockReturnValueOnce(harness.client);
start.mockResolvedValueOnce(harness.client);
}
const environmentIds = new Set<string>();
@ -831,7 +831,7 @@ describe("startCodexAttemptThread", () => {
const runtime = createPairedAttemptRuntime();
const firstPaths = createAttemptPaths();
const secondPaths = createAttemptPaths();
const start = vi.spyOn(CodexAppServerClient, "start").mockImplementation((options) => {
const start = vi.spyOn(CodexAppServerClient, "start").mockImplementation(async (options) => {
const codexHome = options?.env?.CODEX_HOME;
if (codexHome?.startsWith(`${firstPaths.agentDir}${path.sep}`)) {
return first.client;

17
extensions/codex/src/app-server/client.ts
View file

@ -273,7 +273,10 @@ export class CodexAppServerClient {
}
/** Starts a new app-server client using resolved runtime start options. */
static start(options?: Partial<CodexAppServerStartOptions>): CodexAppServerClient {
static async start(
options?: Partial<CodexAppServerStartOptions>,
assertCurrent?: () => void,
): Promise<CodexAppServerClient> {
const defaults = resolveCodexAppServerRuntimeOptions().start;
const startOptions = {
...defaults,
@ -286,7 +289,17 @@ export class CodexAppServerClient {
if (startOptions.transport === "websocket" || startOptions.transport === "unix") {
return new CodexAppServerClient(createWebSocketTransport(startOptions));
}
return new CodexAppServerClient(createStdioTransport(startOptions));
// The spawn callback runs synchronously before registration; initialization
// stays blocked until registration finishes, without losing startup errors.
let client!: CodexAppServerClient;
try {
await createStdioTransport(startOptions, process.env, assertCurrent, (child) => {
client = new CodexAppServerClient(child);
});
return client;
} catch (error) {
throw client?.getCloseError() ?? error;
}
}
/** Builds a client around a fake transport for tests. */

8
extensions/codex/src/app-server/models.test.ts
View file

@ -166,7 +166,7 @@ describe("listCodexAppServerModels", () => {
},
])("rejects $label through the app-server JSON-RPC boundary", async ({ response }) => {
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const listPromise = listCodexAppServerModels({ timeoutMs: 1000 });
await vi.waitFor(() => expect(harness.writes.length).toBeGreaterThanOrEqual(1));
@ -187,7 +187,7 @@ describe("listCodexAppServerModels", () => {
it("lists app-server models through the typed helper", async () => {
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const listPromise = listCodexAppServerModels({ limit: 12, timeoutMs: 1000 });
await vi.waitFor(() => expect(harness.writes.length).toBeGreaterThanOrEqual(1));
@ -257,7 +257,7 @@ describe("listCodexAppServerModels", () => {
it("lists all app-server model pages through one client", async () => {
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const listPromise = listAllCodexAppServerModels({ limit: 1, timeoutMs: 1000 });
await vi.waitFor(() => expect(harness.writes.length).toBeGreaterThanOrEqual(1));
@ -337,7 +337,7 @@ describe("listCodexAppServerModels", () => {
it("marks all-model listing truncated after the page cap", async () => {
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const listPromise = listAllCodexAppServerModels({ limit: 1, timeoutMs: 1000, maxPages: 1 });
await vi.waitFor(() => expect(harness.writes.length).toBeGreaterThanOrEqual(1));

16
extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts
View file

@ -188,7 +188,7 @@ describe("Codex app-server main thread cleanup", () => {
b: path.join(tempDir, "session-b.jsonl"),
};
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
for (const [index, label] of (["a", "b", "a", "b"] as const).entries()) {
const sessionKey = `agent:main:session-${label}`;
@ -284,7 +284,7 @@ describe("Codex app-server main thread cleanup", () => {
it("preserves a quiet long-running native tool while a distinct shared-client turn completes", async () => {
const physical = createClientHarness();
const startClient = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(physical.client);
const startClient = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(physical.client);
const firstParams = createParams(
path.join(tempDir, "concurrent-first.jsonl"),
path.join(tempDir, "concurrent-first-workspace"),
@ -441,7 +441,7 @@ describe("Codex app-server main thread cleanup", () => {
const workspaceDir = path.join(tempDir, "incognito-workspace");
const sessionKey = "agent:main:dashboard:incognito-live-thread";
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const run = runCodexAppServerAttempt(createParams(sessionFile, workspaceDir, sessionKey), {
bindingStore: testCodexAppServerBindingStore,
});
@ -547,7 +547,7 @@ describe("Codex app-server main thread cleanup", () => {
const workspaceDir = path.join(tempDir, "cancelled-start-workspace");
const harness = createClientHarness();
const abort = new AbortController();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const params = createParams(sessionFile, workspaceDir);
params.abortSignal = abort.signal;
@ -655,7 +655,7 @@ describe("Codex app-server main thread cleanup", () => {
const harness = createClientHarness();
const abort = new AbortController();
const close = vi.spyOn(harness.client, "close");
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const params = createParams(sessionFile, workspaceDir, sessionKey);
params.abortSignal = abort.signal;
@ -759,7 +759,7 @@ describe("Codex app-server main thread cleanup", () => {
it("rejects late cancellation after failed finalization enters cleanup", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const abort = new AbortController();
const params = createParams(
path.join(tempDir, "failed-finalization.jsonl"),
@ -803,8 +803,8 @@ describe("Codex app-server main thread cleanup", () => {
const replacement = createClientHarness();
const startClient = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(contaminated.client)
.mockReturnValueOnce(replacement.client);
.mockResolvedValueOnce(contaminated.client)
.mockResolvedValueOnce(replacement.client);
const failedRun = runCodexAppServerAttempt(
createParams(sessionFile, workspaceDir, sessionKey),

230
extensions/codex/src/app-server/shared-client.test.ts
View file

@ -320,7 +320,7 @@ describe("shared Codex app-server client", () => {
it("closes the shared app-server when the version gate fails", async () => {
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
// Model discovery uses the shared-client path, which owns child teardown
// when initialize discovers an unsupported app-server.
@ -372,6 +372,25 @@ describe("shared Codex app-server client", () => {
expect(startSpy).not.toHaveBeenCalled();
});
it("bounds isolated transport startup and closes a client returned after its deadline", async () => {
vi.useFakeTimers();
const harness = createClientHarness();
let finishStart!: (client: CodexAppServerClient) => void;
const starting = new Promise<CodexAppServerClient>((resolve) => {
finishStart = resolve;
});
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(starting);
const acquire = createIsolatedCodexAppServerClient({ timeoutMs: 50 });
const rejected = expect(acquire).rejects.toThrow("codex app-server initialize timed out");
await vi.advanceTimersByTimeAsync(0);
expect(startSpy).toHaveBeenCalledOnce();
await vi.advanceTimersByTimeAsync(50);
await rejected;
finishStart(harness.client);
await vi.advanceTimersByTimeAsync(0);
expect(harness.stdinDestroyed).toBe(true);
});
it.each(["implicit", "explicit"] as const)(
"revalidates %s auth before reusing a warm client after account replacement",
async (selector) => {
@ -379,8 +398,8 @@ describe("shared Codex app-server client", () => {
const replacement = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(replacement.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(replacement.client);
mocks.resolveCodexAppServerAuthProfileIdForAgent.mockReturnValue("openai:work");
mocks.resolveCodexAppServerAuthProfileStore.mockReturnValue({ version: 1, profiles: {} });
const options = {
@ -462,7 +481,7 @@ describe("shared Codex app-server client", () => {
it("rejects an aborted startup acquire while another caller keeps initialization alive", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const abortController = new AbortController();
const first = getLeasedSharedCodexAppServerClient({
abandonSignal: abortController.signal,
@ -482,7 +501,7 @@ describe("shared Codex app-server client", () => {
it("retains an initialized shared client by its persisted instance id", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const acquire = getLeasedSharedCodexAppServerClient({ timeoutMs: 1_000 });
await sendInitializeResult(harness, "openclaw/0.149.0 (Linux; test)");
const client = await acquire;
@ -495,7 +514,7 @@ describe("shared Codex app-server client", () => {
it("captures configuration ownership only for a sole registered lease", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
expect(() => captureExclusiveSharedCodexAppServerClient(harness.client)).toThrow(
CodexAdoptedThreadActiveError,
);
@ -519,7 +538,7 @@ describe("shared Codex app-server client", () => {
"revokes captured configuration ownership after a completed sibling %s",
async (operation) => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const options = {
timeoutMs: 1_000,
config: {},
@ -550,7 +569,7 @@ describe("shared Codex app-server client", () => {
it("revokes configuration ownership when an unleased acquire is pending", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const acquire = getLeasedSharedCodexAppServerClient({ timeoutMs: 1_000 });
await sendInitializeResult(harness, "openclaw/0.149.0 (Linux; test)");
const client = await acquire;
@ -575,7 +594,7 @@ describe("shared Codex app-server client", () => {
it("revokes configuration ownership when its physical client is retired", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const acquire = getLeasedSharedCodexAppServerClient({ timeoutMs: 1_000 });
await sendInitializeResult(harness, "openclaw/0.149.0 (Linux; test)");
const client = await acquire;
@ -596,7 +615,7 @@ describe("shared Codex app-server client", () => {
async (replacementOutcome) => {
const harness = createClientHarness();
const replacement = createClientHarness();
const start = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const start = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const options = { timeoutMs: 1_000 };
const firstLease = getLeasedSharedCodexAppServerClient(options);
await sendInitializeResult(harness, "openclaw/0.149.0 (Linux; test)");
@ -608,7 +627,7 @@ describe("shared Codex app-server client", () => {
new Error("replacement acquisition failed"),
);
} else {
start.mockReturnValue(replacement.client);
start.mockResolvedValue(replacement.client);
}
const retry = withLeasedCodexAppServerClientStartSelectionRetry({
@ -640,7 +659,7 @@ describe("shared Codex app-server client", () => {
it("falls back before starting a desktop candidate with incomplete Computer Use artifacts", async () => {
const pluginLocal = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(pluginLocal.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(pluginLocal.client);
mocks.reconcileCodexComputerUseStartArtifacts
.mockRejectedValueOnce(
new mocks.CodexComputerUseCandidateArtifactsUnavailableError(
@ -658,6 +677,7 @@ describe("shared Codex app-server client", () => {
expect(startSpy).toHaveBeenCalledTimes(1);
expect(startSpy).toHaveBeenCalledWith(
expect.objectContaining({ command: "/cache/openclaw/codex" }),
expect.any(Function),
);
expect(mocks.reconcileCodexComputerUseStartArtifacts).toHaveBeenCalledTimes(2);
expect(mocks.reconcileCodexComputerUseStartArtifacts.mock.calls[0]?.[0]).toEqual(
@ -697,9 +717,9 @@ describe("shared Codex app-server client", () => {
const pluginLocal = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(desktop.client)
.mockReturnValueOnce(pluginLocal.client)
.mockImplementation(() => {
.mockResolvedValueOnce(desktop.client)
.mockResolvedValueOnce(pluginLocal.client)
.mockImplementation(async () => {
throw new Error("unexpected duplicate start");
});
const startOptions = configureManagedDesktopFallback();
@ -735,7 +755,7 @@ describe("shared Codex app-server client", () => {
it("keeps a supported desktop prerelease instead of falling back by version", async () => {
const desktop = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(desktop.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(desktop.client);
const startOptions = configureManagedDesktopFallback();
const acquire = getSharedCodexAppServerClient({ startOptions, timeoutMs: 1_000 });
@ -767,9 +787,9 @@ describe("shared Codex app-server client", () => {
const fallback = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(desktop.client)
.mockReturnValueOnce(fallback.client)
.mockImplementation(() => {
.mockResolvedValueOnce(desktop.client)
.mockResolvedValueOnce(fallback.client)
.mockImplementation(async () => {
throw new Error("unexpected duplicate start");
});
const options = {
@ -798,8 +818,8 @@ describe("shared Codex app-server client", () => {
const captured = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(normal.client)
.mockReturnValueOnce(captured.client);
.mockResolvedValueOnce(normal.client)
.mockResolvedValueOnce(captured.client);
const startOptions: CodexAppServerStartOptions = {
transport: "stdio",
command,
@ -844,8 +864,8 @@ describe("shared Codex app-server client", () => {
const desktop = createClientHarness();
const fallback = createClientHarness();
vi.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(desktop.client)
.mockReturnValueOnce(fallback.client);
.mockResolvedValueOnce(desktop.client)
.mockResolvedValueOnce(fallback.client);
mocks.resolveManagedCodexAppServerStartOptions.mockImplementationOnce(
async (startOptions) => ({
...startOptions,
@ -907,7 +927,7 @@ describe("shared Codex app-server client", () => {
it("detects persisted Computer Use enabled after managed client startup", async () => {
await withTempDir("openclaw-codex-managed-selection-", async (root) => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
mocks.resolveManagedCodexAppServerStartOptions.mockImplementationOnce(
async (startOptions) => ({
...startOptions,
@ -1005,7 +1025,7 @@ describe("shared Codex app-server client", () => {
const generationX = { epoch: 1, fingerprint: "desktop-x" };
mocks.desktopGeneration = generationX;
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const startOptions: CodexAppServerStartOptions = {
transport: "stdio",
homeScope: "agent",
@ -1032,7 +1052,7 @@ describe("shared Codex app-server client", () => {
async (mode) => {
await withTempDir("openclaw-codex-guarded-request-cancel-", async (root) => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
mocks.resolveManagedCodexAppServerStartOptions.mockImplementationOnce(
async (startOptions) => ({
...startOptions,
@ -1100,8 +1120,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
let markFirstStarted: () => void = () => undefined;
const firstStarted = new Promise<void>((resolve) => {
markFirstStarted = resolve;
@ -1130,7 +1150,7 @@ describe("shared Codex app-server client", () => {
it("includes redacted app-server stderr when shared initialize times out", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const models = listCodexAppServerModels({ timeoutMs: 100 });
await vi.waitFor(() => expect(harness.writes.length).toBeGreaterThanOrEqual(1));
@ -1146,7 +1166,7 @@ describe("shared Codex app-server client", () => {
it("keeps shared startup alive for a caller with a longer initialize timeout", async () => {
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const shortAcquire = getSharedCodexAppServerClient({ timeoutMs: 5 });
const longAcquire = getSharedCodexAppServerClient({ timeoutMs: 1000 });
@ -1163,7 +1183,7 @@ describe("shared Codex app-server client", () => {
it("reports a stalled shared auth phase separately from initialize", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const releaseAuth = deferNextAuthProfileApplication();
const acquire = getSharedCodexAppServerClient({ timeoutMs: 100 });
@ -1176,7 +1196,7 @@ describe("shared Codex app-server client", () => {
it("keeps shared auth alive for a caller with a longer timeout", async () => {
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const releaseAuth = deferNextAuthProfileApplication();
const shortAcquire = getSharedCodexAppServerClient({ timeoutMs: 100 });
@ -1195,7 +1215,7 @@ describe("shared Codex app-server client", () => {
it("keeps a pending shared app-server alive when another acquire still owns startup", async () => {
const harness = createClientHarness();
const abandonController = new AbortController();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const abandonedAcquire = getSharedCodexAppServerClient({
timeoutMs: 1000,
@ -1220,7 +1240,7 @@ describe("shared Codex app-server client", () => {
it("does not wait for isolated initialize after a timeout closes the client", async () => {
vi.useFakeTimers();
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
let markStarted: () => void = () => undefined;
const started = new Promise<void>((resolve) => {
markStarted = resolve;
@ -1239,7 +1259,7 @@ describe("shared Codex app-server client", () => {
it("includes redacted app-server stderr when isolated initialize times out", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const client = createIsolatedCodexAppServerClient({ timeoutMs: 100 });
await vi.waitFor(() => expect(harness.writes.length).toBeGreaterThanOrEqual(1));
@ -1253,7 +1273,7 @@ describe("shared Codex app-server client", () => {
it("includes isolated auth application in the total startup deadline", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
let finishAuth: () => void = () => undefined;
mocks.applyCodexAppServerAuthProfile.mockImplementationOnce(
async () =>
@ -1275,7 +1295,7 @@ describe("shared Codex app-server client", () => {
it("does not start isolated auth after the total startup deadline elapsed", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
let now = 0;
vi.spyOn(Date, "now").mockImplementation(() => now);
@ -1291,7 +1311,7 @@ describe("shared Codex app-server client", () => {
it("passes the selected auth profile through the bridge helper", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const listPromise = listCodexAppServerModels({
timeoutMs: 1000,
@ -1309,7 +1329,7 @@ describe("shared Codex app-server client", () => {
it("carries a scoped auth store through isolated app-server startup", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const authProfileStore = { version: 1, profiles: {} };
const preparedAuthProfileStore = {
version: 1,
@ -1374,8 +1394,8 @@ describe("shared Codex app-server client", () => {
const replacement = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(replacement.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(replacement.client);
const options = { timeoutMs: 1_000, authProfileId: "openai:work" };
const acquired = getLeasedSharedCodexAppServerClient(options);
await sendInitializeResult(first, "openclaw/0.149.0 (Linux; test)");
@ -1421,7 +1441,7 @@ describe("shared Codex app-server client", () => {
it("keeps a shared prepared auth store authoritative through startup and refresh", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const authProfileStore = {
version: 1 as const,
profiles: {
@ -1494,8 +1514,8 @@ describe("shared Codex app-server client", () => {
const secondHarness = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(firstHarness.client)
.mockReturnValueOnce(secondHarness.client);
.mockResolvedValueOnce(firstHarness.client)
.mockResolvedValueOnce(secondHarness.client);
const resolvedCacheKeys: string[] = [];
mocks.resolveCodexAppServerPreparedAuthProfileSnapshot.mockImplementation(
async (params?: {
@ -1598,7 +1618,7 @@ describe("shared Codex app-server client", () => {
it("starts a prepared API-key client without profile or ambient-store resolution", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const clientPromise = getSharedCodexAppServerClient({
timeoutMs: 1000,
@ -1627,8 +1647,8 @@ describe("shared Codex app-server client", () => {
const harness = createClientHarness();
const start = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(harness.client)
.mockImplementation(() => {
.mockResolvedValueOnce(harness.client)
.mockImplementation(async () => {
throw new Error("control resume opened a second physical client");
});
const preparedAuth: CodexAppServerPreparedAuth =
@ -1695,8 +1715,8 @@ describe("shared Codex app-server client", () => {
const secondHarness = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(firstHarness.client)
.mockReturnValueOnce(secondHarness.client);
.mockResolvedValueOnce(firstHarness.client)
.mockResolvedValueOnce(secondHarness.client);
const cacheKeys: string[] = [];
mocks.resolveCodexAppServerPreparedApiKeyCacheKey.mockImplementation((apiKey: string) => {
const cacheKey =
@ -1738,7 +1758,7 @@ describe("shared Codex app-server client", () => {
it("registers persisted profile refresh for isolated app-server startup", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const clientPromise = createIsolatedCodexAppServerClient({
timeoutMs: 1000,
@ -1779,7 +1799,7 @@ describe("shared Codex app-server client", () => {
it("skips target auth resolution when native source auth is requested", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const config = { auth: { order: { openai: ["openai:target"] } } };
const clientPromise = getSharedCodexAppServerClient({
@ -1806,7 +1826,7 @@ describe("shared Codex app-server client", () => {
it("uses native auth automatically for shared user-home clients", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const clientPromise = createIsolatedCodexAppServerClient({
timeoutMs: 1000,
@ -1829,7 +1849,7 @@ describe("shared Codex app-server client", () => {
it("resolves the configured implicit auth profile before sharing a client", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const config = { auth: { order: { openai: ["openai:work"] } } };
mocks.resolveCodexAppServerAuthProfileIdForAgent.mockReturnValue("openai:work");
@ -1857,7 +1877,7 @@ describe("shared Codex app-server client", () => {
it("uses the selected agent dir for shared app-server auth bridging", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const listPromise = listCodexAppServerModels({
timeoutMs: 1000,
@ -1881,8 +1901,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const firstList = listCodexAppServerModels({
timeoutMs: 1000,
@ -1907,7 +1927,7 @@ describe("shared Codex app-server client", () => {
it("resolves the managed binary before bridging and spawning the shared client", async () => {
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
mocks.resolveManagedCodexAppServerStartOptions.mockImplementationOnce(async (startOptions) => ({
...startOptions,
command: "/cache/openclaw/codex",
@ -1939,7 +1959,7 @@ describe("shared Codex app-server client", () => {
'[plugins."computer-use@openai-bundled"]\nenabled = true\n',
);
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValue(harness.client);
const clientPromise = createIsolatedCodexAppServerClient({
agentDir,
@ -1966,8 +1986,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const firstList = listCodexAppServerModels({
timeoutMs: 1000,
@ -2008,8 +2028,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
mocks.resolveCodexAppServerFallbackApiKeyCacheKey
.mockReturnValueOnce("api-key:first")
.mockReturnValueOnce("api-key:second");
@ -2040,8 +2060,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const firstList = listCodexAppServerModels({
timeoutMs: 1000,
@ -2082,8 +2102,8 @@ describe("shared Codex app-server client", () => {
const first = createClientHarness();
const second = createClientHarness();
vi.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const firstList = listCodexAppServerModels({
timeoutMs: 1000,
@ -2126,8 +2146,8 @@ describe("shared Codex app-server client", () => {
const first = createClientHarness();
const second = createClientHarness();
vi.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const firstList = listCodexAppServerModels({ timeoutMs: 1000 });
await sendInitializeResult(first, "openclaw/0.149.0 (macOS; test)");
@ -2152,8 +2172,8 @@ describe("shared Codex app-server client", () => {
const first = createClientHarness();
const second = createClientHarness();
vi.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const firstList = listCodexAppServerModels({ timeoutMs: 1000 });
await sendInitializeResult(first, "openclaw/0.149.0 (macOS; test)");
@ -2192,7 +2212,7 @@ describe("shared Codex app-server client", () => {
it("keeps a retired one-shot client alive until native subagent completion", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const clientPromise = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
await sendInitializeResult(harness, "openclaw/0.149.0 (Linux; test)");
@ -2286,7 +2306,7 @@ describe("shared Codex app-server client", () => {
it("leases shared app-server clients before returning concurrent acquirers", async () => {
const first = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(first.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(first.client);
const firstLease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
const secondLease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
@ -2319,8 +2339,8 @@ describe("shared Codex app-server client", () => {
const replacement = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(replacement.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(replacement.client);
const completedRunLease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
const siblingRunLease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
@ -2356,8 +2376,8 @@ describe("shared Codex app-server client", () => {
const first = createClientHarness();
const second = createClientHarness();
vi.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const firstLease = getLeasedSharedCodexAppServerClient();
const pendingLease = getLeasedSharedCodexAppServerClient();
@ -2380,7 +2400,7 @@ describe("shared Codex app-server client", () => {
it("suspect retirement closes a client that was already gracefully detached", async () => {
const first = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(first.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(first.client);
const lease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
await sendInitializeResult(first, "openclaw/0.149.0 (macOS; test)");
@ -2406,7 +2426,7 @@ describe("shared Codex app-server client", () => {
it("retires gracefully by default: leased clients close on release, not immediately", async () => {
const first = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(first.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(first.client);
const lease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
await sendInitializeResult(first, "openclaw/0.149.0 (macOS; test)");
@ -2437,7 +2457,7 @@ describe("shared Codex app-server client", () => {
"invalidates catalog observations before %s settles",
async (method) => {
const transport = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(transport.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(transport.client);
const lease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
await sendInitializeResult(transport, "openclaw/0.149.0 (test)");
const client = await lease;
@ -2463,7 +2483,7 @@ describe("shared Codex app-server client", () => {
commandSource: "resolved-managed" as const,
}));
const harness = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const config = {};
const startOptions: CodexAppServerStartOptions = {
transport: "stdio",
@ -2607,8 +2627,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const config = {};
const startOptions: CodexAppServerStartOptions = {
transport: "stdio",
@ -2684,8 +2704,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const options = {
config: {},
agentDir: "/tmp/openclaw-agent",
@ -2740,8 +2760,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const options = {
config: {},
agentDir: "/tmp/openclaw-agent",
@ -2799,8 +2819,8 @@ describe("shared Codex app-server client", () => {
const first = createClientHarness();
const second = createClientHarness();
vi.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const options = {
config: {},
agentDir: "/tmp/openclaw-agent",
@ -2845,8 +2865,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const options = {
config: {},
agentDir: "/tmp/openclaw-agent",
@ -2900,8 +2920,8 @@ describe("shared Codex app-server client", () => {
const second = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const startOptions: CodexAppServerStartOptions = {
transport: "stdio",
homeScope: "agent",
@ -2954,8 +2974,8 @@ describe("shared Codex app-server client", () => {
const packageY = createClientHarness();
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(packageX.client)
.mockReturnValueOnce(packageY.client);
.mockResolvedValueOnce(packageX.client)
.mockResolvedValueOnce(packageY.client);
const options = {
config: {},
pluginConfig: { computerUse: { enabled: true, autoInstall: true } },
@ -3005,7 +3025,9 @@ describe("shared Codex app-server client", () => {
const generationY = { epoch: 2, fingerprint: "desktop-y" };
mocks.desktopGeneration = generationX;
const packageX = createClientHarness();
const startSpy = vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(packageX.client);
const startSpy = vi
.spyOn(CodexAppServerClient, "start")
.mockResolvedValueOnce(packageX.client);
const options = {
config: {},
pluginConfig: { computerUse: { enabled: true, autoInstall: true } },
@ -3045,7 +3067,7 @@ describe("shared Codex app-server client", () => {
const generation = { epoch: 1, fingerprint: "desktop-x" };
mocks.desktopGeneration = generation;
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const clientPromise = getLeasedSharedCodexAppServerClient({
config: {},
@ -3082,10 +3104,10 @@ describe("shared Codex app-server client", () => {
const packageY = createClientHarness();
const desktopY = createClientHarness();
vi.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(packageX.client)
.mockReturnValueOnce(desktopX.client)
.mockReturnValueOnce(packageY.client)
.mockReturnValueOnce(desktopY.client);
.mockResolvedValueOnce(packageX.client)
.mockResolvedValueOnce(desktopX.client)
.mockResolvedValueOnce(packageY.client)
.mockResolvedValueOnce(desktopY.client);
const options = {
config: {},
agentDir: "/tmp/openclaw-agent",
@ -3139,7 +3161,7 @@ describe("shared Codex app-server client", () => {
commandSource: "resolved-managed" as const,
}));
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const acquire = getLeasedSharedCodexAppServerClient({
config: {},
agentDir: "/tmp/openclaw-agent",
@ -3166,7 +3188,7 @@ describe("shared Codex app-server client", () => {
it("globally disposes a gracefully detached client with an explicit retain", async () => {
const harness = createClientHarness();
vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client);
vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(harness.client);
const lease = getLeasedSharedCodexAppServerClient({ timeoutMs: 1000 });
await sendInitializeResult(harness, "openclaw/0.149.0 (Linux; test)");
@ -3193,8 +3215,8 @@ describe("shared Codex app-server client", () => {
const first = createClientHarness();
const second = createClientHarness();
vi.spyOn(CodexAppServerClient, "start")
.mockReturnValueOnce(first.client)
.mockReturnValueOnce(second.client);
.mockResolvedValueOnce(first.client)
.mockResolvedValueOnce(second.client);
const firstCloseAndWait = vi.spyOn(first.client, "closeAndWait");
const secondCloseAndWait = vi.spyOn(second.client, "closeAndWait");

21
extensions/codex/src/app-server/shared-client.ts
View file

@ -1101,7 +1101,26 @@ async function startInitializedCodexAppServerClient(params: {
throw new Error("Codex app-server runtime artifact does not match verified inference");
}
assertDesktopGenerationCurrent();
const client = CodexAppServerClient.start(startOptions);
let starting: Promise<CodexAppServerClient> | undefined;
let client: CodexAppServerClient;
try {
client = await withCodexAppServerAcquireDeadline(
resolveRemainingAcquireTimeout(timeoutMs, acquireStartedAt),
(starting = CodexAppServerClient.start(startOptions, () => {
assertDesktopGenerationCurrent();
resolveRemainingAcquireTimeout(timeoutMs, acquireStartedAt);
})),
params.abandonSignal,
);
} catch (error) {
// A timed-out registration may settle later; it cannot publish a live
// client after the acquisition owner has already released its claim.
void starting?.then(
(lateClient) => lateClient.close(),
() => {},
);
throw error;
}
const nativeCommandAtStart =
startOptions.commandSource === "resolved-managed"
? resolveManagedCodexNativeCommand(startOptions.command)

4
extensions/codex/src/app-server/thread-lifecycle.binding.test.ts
View file

@ -425,7 +425,7 @@ async function createManualResumeFixture(
close: () => harness.close(),
});
}
const start = vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(client);
const start = vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(client);
try {
await getLeasedSharedCodexAppServerClient({
startOptions: { ...createThreadLifecycleAppServerOptions().start, command: process.execPath },
@ -540,7 +540,7 @@ async function createLeasedLifecycleWireClient(
return count;
});
vi.spyOn(wire.client, "initialize").mockResolvedValue(undefined);
const start = vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(wire.client);
const start = vi.spyOn(CodexAppServerClient, "start").mockResolvedValueOnce(wire.client);
try {
await getLeasedSharedCodexAppServerClient({
startOptions: { ...createThreadLifecycleAppServerOptions().start, command: process.execPath },

94
extensions/codex/src/app-server/transport-orphan.test-helper.ts Normal file
View file

@ -0,0 +1,94 @@
import { spawn } from "node:child_process";
import { createInterface } from "node:readline";
import { fileURLToPath } from "node:url";
const fixture = fileURLToPath(import.meta.url);
if (process.argv[2] === "child") {
const descendant = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], {
detached: true,
stdio: "ignore",
});
process.stdout.write(`${JSON.stringify({ child: process.pid, descendant: descendant.pid })}\n`);
process.stdin.resume();
setInterval(() => {}, 1000);
} else {
const { createStdioTransport } = await import("./transport-stdio.js");
if (process.argv[2] === "native") {
const command = process.argv[4]!;
const cwd = process.argv[5]!;
const child = await createStdioTransport(
{
transport: "stdio",
command,
args: ["app-server", "--listen", "stdio://"],
cwd,
headers: {},
},
process.env,
);
child.stderr.pipe(process.stderr);
const send = (message: object) => child.stdin.write(`${JSON.stringify(message)}\n`);
createInterface({ input: child.stdout }).on("line", (line) => {
// SAFETY: The pinned native test binary emits Codex JSON-RPC envelopes on stdout.
const message = JSON.parse(line) as {
id?: number;
method?: string;
error?: unknown;
params: { deltaBase64: string };
};
if (message.error) {
throw new Error(JSON.stringify(message.error));
}
if (message.id === 1) {
send({ method: "initialized", params: {} });
send({
id: 2,
method: "command/exec",
params: {
command: [
process.execPath,
"-e",
"process.stdout.write(String(process.pid)+'\\n');setInterval(()=>{},1000)",
],
processId: "orphan-proof",
streamStdoutStderr: true,
disableTimeout: true,
sandboxPolicy: { type: "dangerFullAccess" },
cwd,
},
});
} else if (message.method === "command/exec/outputDelta") {
const descendant = Number(
Buffer.from(message.params.deltaBase64, "base64").toString().trim(),
);
if (Number.isSafeInteger(descendant) && descendant > 0) {
process.stdout.write(
`${JSON.stringify({ parent: process.pid, child: child.pid, descendant })}\n`,
);
}
}
});
send({
id: 1,
method: "initialize",
params: {
clientInfo: { name: "openclaw_orphan_test", version: "1.0.0" },
capabilities: { experimentalApi: true },
},
});
} else {
const child = await createStdioTransport({
transport: "stdio",
command: process.execPath,
args: ["--import", "tsx", fixture, "child"],
headers: {},
});
child.stderr.pipe(process.stderr);
createInterface({ input: child.stdout }).once("line", (line) => {
process.stdout.write(`${JSON.stringify({ parent: process.pid, ...JSON.parse(line) })}\n`);
});
child.once("error", (error) => {
throw error;
});
}
}

152
extensions/codex/src/app-server/transport-orphan.test.ts Normal file
View file

@ -0,0 +1,152 @@
import { execFileSync, spawn, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { createInterface } from "node:readline";
import { fileURLToPath } from "node:url";
import { describe, expect, it } from "vitest";
import { createCodexNativeTestState } from "./native-app-server.test-support.js";
type ProcessTree = { parent: number; child: number; descendant: number };
const fixture = fileURLToPath(new URL("./transport-orphan.test-helper.ts", import.meta.url));
function isAlive(pid: number) {
try {
process.kill(pid, 0);
return !execFileSync("ps", ["-o", "stat=", "-p", String(pid)], { encoding: "utf8" })
.trim()
.startsWith("Z");
} catch {
return false;
}
}
describe.skipIf(process.platform === "win32")("Codex stdio crash recovery", () => {
it.for(["fixture", "native"])(
"reaps the dead $0 owner before a fresh spawn and preserves live owners",
{ timeout: 60_000 },
async (mode, ctx) => {
const root = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-orphan-"));
const fakeBin = path.join(root, "bin");
await fs.mkdir(fakeBin);
await fs.writeFile(path.join(fakeBin, "ps"), "#!/bin/sh\nexit 1\n", { mode: 0o755 });
const unavailablePs = `${fakeBin}${path.delimiter}${process.env.PATH}`;
const parents: ChildProcess[] = [];
const trees: ProcessTree[] = [];
ctx.onTestFinished(async () => {
// Also capture children of a fixture that failed before reporting ready.
const rows = execFileSync("ps", ["-axo", "pid=,ppid="], { encoding: "utf8" })
.trim()
.split("\n")
.map((line) => line.trim().split(/\s+/).map(Number));
const owned = new Set(parents.flatMap((parent) => (parent.pid ? [parent.pid] : [])));
for (const pid of owned) {
for (const [child, parent] of rows) {
if (parent === pid && child) {
owned.add(child);
}
}
}
for (const pid of [...owned].toReversed()) {
try {
process.kill(pid, "SIGKILL");
} catch {
/* Already exited. */
}
}
for (const tree of trees) {
for (const pid of [tree.descendant, tree.child, tree.parent]) {
try {
process.kill(pid, "SIGKILL");
} catch {
// Test-owned processes may already have been reaped by the successor.
}
}
}
for (const parent of parents) {
parent.kill("SIGKILL");
parent.stdout?.destroy();
parent.stderr?.destroy();
}
await fs.rm(root, { recursive: true, force: true });
});
const start = async (
stateDir: string,
searchPath = process.platform === "linux" ? unavailablePs : process.env.PATH,
) => {
const native =
mode === "native"
? await createCodexNativeTestState(path.join(root, `native-${parents.length}`))
: undefined;
if (native) {
await fs.writeFile(
path.join(native.codexHome, "config.toml"),
'cli_auth_credentials_store="ephemeral"\n[features]\nrespect_system_proxy=false\nshell_snapshot=false\n[analytics]\nenabled=false\n[feedback]\nenabled=false\n',
);
}
const args = [
"--import",
"tsx",
fixture,
mode,
root,
...(native ? [native.command, native.cwd] : []),
];
const parent = spawn(process.execPath, args, {
env: {
HOME: root,
...native?.env,
PATH: searchPath,
OPENCLAW_STATE_DIR: stateDir,
NODE_ENV: "test",
},
stdio: ["ignore", "pipe", "pipe"],
});
parents.push(parent);
let stderr = "";
parent.stderr?.on("data", (chunk: Buffer) => {
stderr = (stderr + chunk.toString()).slice(-4_000);
});
const lines = createInterface({ input: parent.stdout! });
const tree = await new Promise<ProcessTree>((resolve, reject) => {
parent.once("error", reject);
parent.once("exit", () => reject(new Error(`Fixture exited: ${stderr}`)));
lines.once("line", (line) => resolve(JSON.parse(line) as ProcessTree));
});
lines.close();
trees.push(tree);
return { parent, tree };
};
const stateDir = path.join(root, "state");
const old = await start(stateDir);
const live = await start(stateDir);
const other = await start(path.join(root, "other-state"));
expect(isAlive(old.tree.child)).toBe(true);
if (mode === "native") {
// Freeze a real app-server mid-command so EOF cannot complete its cleanup.
process.kill(old.tree.child, "SIGSTOP");
}
const exited = once(old.parent, "exit");
old.parent.kill("SIGKILL");
await exited;
expect(isAlive(old.tree.child)).toBe(true);
expect(isAlive(old.tree.descendant)).toBe(true);
if (process.platform !== "linux") {
await expect(start(stateDir, unavailablePs)).rejects.toThrow(
"Cannot inspect registered Codex processes",
);
expect(isAlive(old.tree.child)).toBe(true);
}
const fresh = await start(stateDir);
expect(isAlive(old.tree.child)).toBe(false);
expect(isAlive(old.tree.descendant)).toBe(false);
for (const tree of [live.tree, other.tree, fresh.tree]) {
expect(isAlive(tree.child)).toBe(true);
expect(isAlive(tree.descendant)).toBe(true);
}
},
);
});

207
extensions/codex/src/app-server/transport-process-containment.ts
View file

@ -1,4 +1,8 @@
import { execFile } from "node:child_process";
import {
readCodexAppServerProcess,
readCodexAppServerProcessSnapshot,
type PosixProcess,
} from "./transport-process-snapshot.js";
type ContainableTransport = {
pid?: number;
@ -7,34 +11,82 @@ type ContainableTransport = {
kill?: (signal?: NodeJS.Signals) => unknown;
};
type PosixProcess = {
pid: number;
ppid: number;
pgid: number;
state: string;
startedAt: string;
};
export type CodexAppServerProcessIdentity = Pick<PosixProcess, "pid" | "pgid" | "startedAt">;
const PROCESS_COLUMNS = "pid=,ppid=,pgid=,stat=,lstart=";
const MAX_CONTAINED_PROCESSES = 512;
const MAX_PROCESS_CONTAINMENT_MS = 2_000;
const MAX_PROCESS_QUIESCE_PASSES = 16;
const PROCESS_INSPECTION_MAX_BYTES = 8 * 1024 * 1024;
export async function terminateCodexAppServerDescendants(
child: ContainableTransport,
): Promise<(() => void) | undefined> {
return (await containDescendants(child))?.resume;
}
/** A durable spawn fact, never a command-line match, selects the orphan root. */
export async function terminateCodexAppServerOrphan(
expected: CodexAppServerProcessIdentity,
): Promise<boolean> {
const deadline = Date.now() + MAX_PROCESS_CONTAINMENT_MS;
const contained = await containDescendants(
{ pid: expected.pid, kill: (signal) => signalProcess(expected.pid, signal ?? "SIGTERM") },
expected,
deadline,
);
let gone = false;
try {
if (contained) {
const current = await readCodexAppServerProcess(expected.pid, deadline);
if (current && isSameLiveRoot(current, contained.root, true)) {
// Keep the verified leader stopped until its whole group is killed;
// a QA-owned child may share its parent's group and must use its PID.
signalProcess(current.pgid === current.pid ? -current.pid : current.pid, "SIGKILL");
}
}
while (Date.now() < deadline) {
const snapshot = await readCodexAppServerProcessSnapshot(deadline);
if (!snapshot?.some((row) => row.pid === process.pid)) {
return false;
}
const current = snapshot.find((row) => row.pid === expected.pid);
if (!current || !hasSameIdentity(current, expected) || current.state.startsWith("Z")) {
gone = true;
return true;
}
if (!contained) {
return false;
}
await new Promise((resolve) => {
setTimeout(resolve, 20);
});
}
return false;
} finally {
if (contained && !gone) {
await signalSameRoot(contained.root, "SIGCONT", Date.now() + MAX_PROCESS_CONTAINMENT_MS);
}
}
}
async function containDescendants(
child: ContainableTransport,
expected?: CodexAppServerProcessIdentity,
deadline = Date.now() + MAX_PROCESS_CONTAINMENT_MS,
): Promise<{ root: PosixProcess; resume: () => void } | undefined> {
const rootPid = child.pid;
if (process.platform === "win32" || !rootPid || !child.kill || hasExited(child)) {
return undefined;
}
const deadline = Date.now() + MAX_PROCESS_CONTAINMENT_MS;
const snapshot = await readProcessSnapshot(deadline);
const snapshot = await readCodexAppServerProcessSnapshot(deadline);
if (!snapshot || Date.now() >= deadline) {
return undefined;
}
const root = snapshot.find((row) => row.pid === rootPid);
if (!root || !isSameLiveRoot(root, root)) {
if (
!root ||
!(expected ? isSameLiveProcess(root, expected) : root.ppid === process.pid) ||
root.state.startsWith("Z")
) {
return undefined;
}
@ -72,21 +124,33 @@ export async function terminateCodexAppServerDescendants(
}
resumeRootOnUnwind = false;
let resumed = false;
return () => {
if (resumed) {
return;
}
resumed = true;
resumeTransportRoot(child, root, false);
return {
root,
resume: () => {
if (resumed) {
return;
}
resumed = true;
resumeTransportRoot(child, root, false);
},
};
} finally {
if (resumeRootOnUnwind) {
// Inspection failure cannot also own release. These PIDs were signaled
// synchronously in this call and have not crossed an asynchronous boundary.
for (const descendant of stoppedDescendants.values()) {
signalProcess(descendant.pid, "SIGCONT");
if (expected) {
// Orphans have no retained child handle. Failed inspection leaves the
// registration intact; never release a reused PID while unwinding.
const releaseDeadline = Date.now() + MAX_PROCESS_CONTAINMENT_MS;
for (const descendant of stoppedDescendants.values()) {
await signalSameProcess(descendant, "SIGCONT", releaseDeadline);
}
await signalSameRoot(root, "SIGCONT", releaseDeadline);
} else {
// A live parent still owns these stopped children when inspection fails.
for (const descendant of stoppedDescendants.values()) {
signalProcess(descendant.pid, "SIGCONT");
}
resumeTransportRoot(child, root, true);
}
resumeTransportRoot(child, root, true);
}
}
}
@ -103,7 +167,7 @@ async function quiesceDescendants(
if (Date.now() >= deadline) {
return undefined;
}
const snapshot = await readProcessSnapshot(deadline);
const snapshot = await readCodexAppServerProcessSnapshot(deadline);
if (!snapshot || Date.now() >= deadline) {
return undefined;
}
@ -188,81 +252,6 @@ async function quiesceDescendants(
return undefined;
}
async function readProcessSnapshot(deadline: number): Promise<PosixProcess[] | undefined> {
return await readProcesses(["-axo", PROCESS_COLUMNS], deadline);
}
async function readProcess(pid: number, deadline: number): Promise<PosixProcess | undefined> {
return (await readProcesses(["-o", PROCESS_COLUMNS, "-p", String(pid)], deadline))?.find(
(row) => row.pid === pid,
);
}
async function readProcesses(
args: string[],
deadline: number,
): Promise<PosixProcess[] | undefined> {
const remainingMs = deadline - Date.now();
if (remainingMs <= 0) {
return undefined;
}
return await new Promise<PosixProcess[] | undefined>((resolve) => {
let settled = false;
const settle = (processes: PosixProcess[] | undefined) => {
if (settled) {
return;
}
settled = true;
clearTimeout(timer);
resolve(processes);
};
const inspector = execFile(
"ps",
args,
{ encoding: "utf8", maxBuffer: PROCESS_INSPECTION_MAX_BYTES },
(error, stdout) => {
settle(error ? undefined : parseProcesses(stdout));
},
);
const timer = setTimeout(
() => {
settle(undefined);
inspector.stdout?.destroy();
inspector.stderr?.destroy();
inspector.kill("SIGKILL");
inspector.unref();
},
Math.max(1, remainingMs),
);
timer.unref?.();
});
}
function parseProcesses(output: string): PosixProcess[] {
const rows: PosixProcess[] = [];
for (const line of output.split("\n")) {
const match = /^\s*(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(.+?)\s*$/.exec(line);
if (!match) {
continue;
}
const pid = Number(match[1] ?? "");
const ppid = Number(match[2] ?? "");
const pgid = Number(match[3] ?? "");
const startedAt = (match[5] ?? "").trim().replace(/\s+/g, " ");
if (
![pid, ppid, pgid].every(Number.isSafeInteger) ||
pid <= 0 ||
ppid < 0 ||
pgid <= 0 ||
!startedAt
) {
continue;
}
rows.push({ pid, ppid, pgid, state: match[4] ?? "", startedAt });
}
return rows;
}
function collectDescendants(snapshot: PosixProcess[], rootPids: number[]): PosixProcess[] {
const childrenByParent = new Map<number, PosixProcess[]>();
for (const row of snapshot) {
@ -298,7 +287,10 @@ function isUninterruptibleState(state: string): boolean {
return state.startsWith("D") || state.startsWith("U");
}
function isSameLiveProcess(current: PosixProcess, expected: PosixProcess): boolean {
function isSameLiveProcess(
current: PosixProcess,
expected: CodexAppServerProcessIdentity,
): boolean {
return (
current.pgid === expected.pgid &&
!current.state.startsWith("Z") &&
@ -312,7 +304,7 @@ function isSameLiveRoot(
requireStopped = false,
): boolean {
return (
current.ppid === process.pid &&
current.ppid === expected.ppid &&
(!requireStopped || isQuiescedState(current.state)) &&
isSameLiveProcess(current, expected)
);
@ -323,7 +315,7 @@ async function signalSameRoot(
signal: NodeJS.Signals,
deadline: number,
): Promise<boolean> {
const current = await readProcess(root.pid, deadline);
const current = await readCodexAppServerProcess(root.pid, deadline);
return Boolean(current && isSameLiveRoot(current, root) && signalProcess(current.pid, signal));
}
@ -356,17 +348,20 @@ async function signalSameProcess(
): Promise<boolean> {
// Portable Node POSIX signals are PID-based, so never retain numeric authority:
// take this final identity snapshot synchronously immediately before every signal.
const current = await readProcess(expected.pid, deadline);
const current = await readCodexAppServerProcess(expected.pid, deadline);
return Boolean(
current && isSameLiveProcess(current, expected) && signalProcess(current.pid, signal),
);
}
function hasSameIdentity(left: PosixProcess, right: PosixProcess): boolean {
function hasSameIdentity(
left: CodexAppServerProcessIdentity,
right: CodexAppServerProcessIdentity,
): boolean {
return identityKey(left) === identityKey(right);
}
function identityKey(row: PosixProcess): string {
function identityKey(row: CodexAppServerProcessIdentity): string {
return `${row.pid}\0${row.startedAt}`;
}

95
extensions/codex/src/app-server/transport-process-registration.ts Normal file
View file

@ -0,0 +1,95 @@
import type { ChildProcessWithoutNullStreams } from "node:child_process";
import { randomUUID } from "node:crypto";
import { once } from "node:events";
import { z } from "zod";
import { terminateCodexAppServerOrphan } from "./transport-process-containment.js";
import { readCodexAppServerProcessSnapshot } from "./transport-process-snapshot.js";
const processIdentity = z.object({
pid: z.number().int().positive().safe(),
pgid: z.number().int().positive().safe(),
startedAt: z.string().min(1).max(64),
});
const registrationSchema = z.object({ parent: processIdentity, child: processIdentity }).strict();
type ProcessRegistration = z.infer<typeof registrationSchema>;
/** Reap previous owners before spawn; commit this child's identity before initialization. */
export async function prepareCodexAppServerProcessRegistration(): Promise<
(child: ChildProcessWithoutNullStreams) => Promise<void>
> {
if (process.platform === "win32") {
return async (child) => {
await once(child, "spawn");
};
}
const { createPluginStateSyncKeyedStore } =
await import("openclaw/plugin-sdk/plugin-state-store-runtime");
const store = createPluginStateSyncKeyedStore<ProcessRegistration>("codex", {
namespace: "app-server-processes",
maxEntries: 512,
// Expiration or eviction could forget a child that still owns a native turn.
overflowPolicy: "reject-new",
});
const deadline = Date.now() + 10_000;
for (const entry of store.entries()) {
if (Date.now() >= deadline) {
throw new Error("Codex orphan cleanup exceeded its startup budget. Retry to finish cleanup.");
}
const registration = registrationSchema.parse(entry.value);
const snapshot = await readCodexAppServerProcessSnapshot();
if (!snapshot?.some((row) => row.pid === process.pid)) {
throw new Error(
"Cannot inspect registered Codex processes. Check process inspection permissions (/proc on Linux, ps on macOS), then retry.",
);
}
const parent = snapshot.find((row) => row.pid === registration.parent.pid);
if (parent?.startedAt === registration.parent.startedAt && !parent.state.startsWith("Z")) {
continue;
}
if (!(await terminateCodexAppServerOrphan(registration.child))) {
throw new Error(
`Cannot reap registered Codex process ${registration.child.pid}. Stop it before retrying.`,
);
}
store.delete(entry.key);
}
return async (child) => {
try {
await once(child, "spawn");
const snapshot = await readCodexAppServerProcessSnapshot();
const parent = snapshot?.find((row) => row.pid === process.pid);
const spawned = snapshot?.find((row) => row.pid === child.pid);
if (
!parent ||
!spawned ||
spawned.ppid !== process.pid ||
child.exitCode !== null ||
child.signalCode !== null
) {
throw new Error(
"Cannot register the Codex child process. Check process inspection permissions (/proc on Linux, ps on macOS), then retry.",
);
}
const key = randomUUID();
// Codex rejects non-initialize requests; no native turn can start before
// this synchronous commit. A failed commit closes the uninitialized child.
store.register(key, {
parent: processIdentity.parse(parent),
child: processIdentity.parse(spawned),
});
child.once("exit", () => {
try {
store.delete(key);
} catch {
// Leave the durable fact for the next connection to verify and remove.
}
});
} catch (error) {
child.kill("SIGKILL");
child.stdin.destroy();
child.stdout.destroy();
child.stderr.destroy();
throw error;
}
};
}

127
extensions/codex/src/app-server/transport-process-snapshot.test.ts Normal file
View file

@ -0,0 +1,127 @@
import { execFileSync } from "node:child_process";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { isPidAlive } from "openclaw/plugin-sdk/process-runtime";
import { withEnvAsync } from "openclaw/plugin-sdk/test-env";
import { describe, expect, it, vi } from "vitest";
import { readCodexAppServerProcessSnapshot } from "./transport-process-snapshot.js";
const procfs = vi.hoisted(() => ({
readFile: vi.fn<(file: string) => Promise<string>>(),
readdir: vi.fn<() => Promise<string[]>>(),
}));
vi.mock("node:fs/promises", async (importOriginal) => {
const original = await importOriginal<typeof import("node:fs/promises")>();
return {
...original,
readFile: (...args: Parameters<typeof original.readFile>) =>
typeof args[0] === "string" &&
args[0].startsWith("/proc/") &&
procfs.readFile.getMockImplementation()
? procfs.readFile(args[0])
: original.readFile(...args),
readdir: (...args: Parameters<typeof original.readdir>) =>
args[0] === "/proc" && procfs.readdir.getMockImplementation()
? procfs.readdir()
: original.readdir(...args),
};
});
describe.skipIf(process.platform !== "linux")("Codex procfs process inspector", () => {
it.for(["ENOENT", "ESRCH", "EACCES"] as const)(
"distinguishes a vanished neighbor from unreadable state: %s",
async (code, ctx) => {
ctx.onTestFinished(() => {
procfs.readFile.mockReset();
procfs.readdir.mockReset();
});
const bootId = "00000000-0000-0000-0000-000000000001";
const neighborPid = process.pid + 1;
procfs.readdir.mockResolvedValue([String(process.pid), String(neighborPid)]);
procfs.readFile.mockImplementation(async (file) => {
if (file === "/proc/sys/kernel/random/boot_id") {
return bootId;
}
if (file === `/proc/${process.pid}/stat`) {
// Fields 3..22 follow the final ')', even when comm contains ')' and spaces.
return `${process.pid} (codex ) worker) S ${process.ppid} ${process.pid}${" 0".repeat(16)} 12345${" 0".repeat(30)}\n`;
}
if (file === `/proc/${neighborPid}/stat`) {
throw Object.assign(new Error("neighbor process read failed"), { code });
}
throw new Error(`Unexpected procfs read: ${file}`);
});
const snapshot = await readCodexAppServerProcessSnapshot();
expect(snapshot).toEqual(
code === "EACCES"
? undefined
: [
{
pid: process.pid,
ppid: process.ppid,
pgid: process.pid,
state: "S",
startedAt: `${bootId}:12345`,
},
],
);
},
);
});
describe.skipIf(process.platform === "win32" || process.platform === "linux")(
"Codex POSIX process inspector",
() => {
it.for(["unavailable", "hung"] as const)(
"settles a %s ps inspector without leaking its process",
async (mode, ctx) => {
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "codex-ps-deadline-"));
const inspectorPath = path.join(tempDir, "ps");
const pidPath = path.join(tempDir, "inspector.pid");
let inspectorPid: number | undefined;
ctx.onTestFinished(async () => {
const pid = inspectorPid ?? Number(await fs.readFile(pidPath, "utf8").catch(() => ""));
if (pid && isPidAlive(pid)) {
const command = execFileSync("/bin/ps", ["-o", "command=", "-p", String(pid)], {
encoding: "utf8",
stdio: ["ignore", "pipe", "ignore"],
});
if (command.includes(inspectorPath)) {
process.kill(pid, "SIGKILL");
}
}
await fs.rm(tempDir, { recursive: true, force: true });
});
await fs.writeFile(
inspectorPath,
`#!/usr/bin/env node
const fs = require("node:fs");
fs.writeFileSync(process.env.CODEX_TEST_PS_PID_FILE, String(process.pid));
${mode === "unavailable" ? "process.exit(1);" : "setInterval(() => {}, 1000);"}
`,
{ mode: 0o755 },
);
await withEnvAsync(
{
PATH: `${tempDir}${path.delimiter}${process.env.PATH ?? ""}`,
CODEX_TEST_PS_PID_FILE: pidPath,
},
async () => {
const startedAt = Date.now();
const budgetMs = 1_000;
const result = await readCodexAppServerProcessSnapshot(startedAt + budgetMs);
const pid = Number(await fs.readFile(pidPath, "utf8"));
inspectorPid = pid;
expect(pid).toBeGreaterThan(0);
expect(result).toBeUndefined();
// Allow scheduler jitter, but not the inspector's unbounded event loop.
expect(Date.now() - startedAt).toBeLessThan(budgetMs + 500);
await expect.poll(() => isPidAlive(pid)).toBe(false);
},
);
},
);
},
);

180
extensions/codex/src/app-server/transport-process-snapshot.ts Normal file
View file

@ -0,0 +1,180 @@
import { execFile } from "node:child_process";
import { readdir, readFile } from "node:fs/promises";
export type PosixProcess = {
pid: number;
ppid: number;
pgid: number;
state: string;
startedAt: string;
};
const PROCESS_COLUMNS = "pid=,ppid=,pgid=,stat=,lstart=";
const MAX_PROCESS_CONTAINMENT_MS = 2_000;
const PROCESS_INSPECTION_MAX_BYTES = 8 * 1024 * 1024;
export async function readCodexAppServerProcessSnapshot(
deadline = Date.now() + MAX_PROCESS_CONTAINMENT_MS,
): Promise<PosixProcess[] | undefined> {
return process.platform === "linux"
? await readLinuxProcesses(undefined, deadline)
: await readProcesses(["-axo", PROCESS_COLUMNS], deadline);
}
export async function readCodexAppServerProcess(
pid: number,
deadline: number,
): Promise<PosixProcess | undefined> {
const rows =
process.platform === "linux"
? await readLinuxProcesses(pid, deadline)
: await readProcesses(["-o", PROCESS_COLUMNS, "-p", String(pid)], deadline);
return rows?.find((row) => row.pid === pid);
}
async function readProcesses(
args: string[],
deadline: number,
): Promise<PosixProcess[] | undefined> {
const remainingMs = deadline - Date.now();
if (remainingMs <= 0) {
return undefined;
}
return await new Promise<PosixProcess[] | undefined>((resolve) => {
let settled = false;
const settle = (processes: PosixProcess[] | undefined) => {
if (settled) {
return;
}
settled = true;
clearTimeout(timer);
resolve(processes);
};
const inspector = execFile(
"ps",
args,
{
encoding: "utf8",
maxBuffer: PROCESS_INSPECTION_MAX_BYTES,
env: { ...process.env, LC_ALL: "C", TZ: "UTC" },
},
(error, stdout) => {
settle(error ? undefined : parseProcesses(stdout));
},
);
const timer = setTimeout(
() => {
settle(undefined);
inspector.stdout?.destroy();
inspector.stderr?.destroy();
inspector.kill("SIGKILL");
inspector.unref();
},
Math.max(1, remainingMs),
);
timer.unref?.();
});
}
function parseProcesses(output: string): PosixProcess[] {
const rows: PosixProcess[] = [];
for (const line of output.split("\n")) {
const match = /^\s*(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(.+?)\s*$/.exec(line);
if (!match) {
continue;
}
const pid = Number(match[1] ?? "");
const ppid = Number(match[2] ?? "");
const pgid = Number(match[3] ?? "");
const startedAt = (match[5] ?? "").trim().replace(/\s+/g, " ");
if (
![pid, ppid, pgid].every(Number.isSafeInteger) ||
pid <= 0 ||
ppid < 0 ||
pgid <= 0 ||
!startedAt
) {
continue;
}
rows.push({ pid, ppid, pgid, state: match[4] ?? "", startedAt });
}
return rows;
}
// Linux exposes stronger start identities in procfs; BusyBox ps on supported
// Alpine installs has no lstart. Boot identity prevents reuse across reboots.
async function readLinuxProcesses(
pid: number | undefined,
deadline: number,
): Promise<PosixProcess[] | undefined> {
const remainingMs = deadline - Date.now();
if (remainingMs <= 0) {
return undefined;
}
const options = { encoding: "utf8" as const, signal: AbortSignal.timeout(remainingMs) };
try {
const bootId = (await readFile("/proc/sys/kernel/random/boot_id", options)).trim();
if (!/^[a-f0-9-]{36}$/.test(bootId)) {
return undefined;
}
const pids = pid === undefined ? await readdir("/proc") : [String(pid)];
const rows: PosixProcess[] = [];
let bytes = 0;
for (const entry of pids) {
if (!/^\d+$/.test(entry)) {
continue;
}
if (Date.now() >= deadline) {
return undefined;
}
const stat = await readFile(`/proc/${entry}/stat`, options).catch((error: unknown) => {
// A process may exit between enumeration and read. Other failures must
// not turn an unreadable process into proof that an orphan is gone.
if (
error &&
typeof error === "object" &&
"code" in error &&
(error.code === "ENOENT" || error.code === "ESRCH")
) {
return undefined;
}
throw error;
});
if (stat === undefined) {
continue;
}
bytes += stat.length;
if (bytes > PROCESS_INSPECTION_MAX_BYTES) {
return undefined;
}
// comm can contain spaces, newlines and ')'; fields 3..N follow its last ')'.
const commEnd = stat.lastIndexOf(")");
const fields = stat
.slice(commEnd + 1)
.trim()
.split(/\s+/);
const ppid = Number(fields[1]);
const pgid = Number(fields[2]);
const startTicks = fields[19];
if (
commEnd < 0 ||
![ppid, pgid].every(Number.isSafeInteger) ||
!/^\d+$/.test(startTicks ?? "")
) {
return undefined;
}
if (pgid > 0) {
rows.push({
pid: Number(entry),
ppid,
pgid,
state: fields[0]!,
startedAt: `${bootId}:${startTicks}`,
});
}
}
return rows;
} catch {
return undefined;
}
}

2
extensions/codex/src/app-server/transport-stdio.config.test.ts
View file

@ -119,7 +119,7 @@ const cases: NativeConfigCase[] = [
];
async function readNativeConfig(startOptions: CodexAppServerStartOptions, env: NodeJS.ProcessEnv) {
const child = createStdioTransport(startOptions, env);
const child = await createStdioTransport(startOptions, env);
const closed = new Promise<void>((resolve) => {
child.once("close", () => resolve());
});

2
extensions/codex/src/app-server/transport-stdio.sandbox.test.ts
View file

@ -164,7 +164,7 @@ describe.skipIf(process.platform !== "darwin")("native Codex turn sandbox", () =
"supports_websockets=false",
].join("\n");
await fs.writeFile(path.join(codexHome, "config.toml"), config);
const child = createStdioTransport(
const child = await createStdioTransport(
{ transport: "stdio", command, commandSource: "config", args, cwd, headers: {} },
{ ...env, PATH: "/usr/bin:/bin", SHELL: "/bin/sh" },
);

57
extensions/codex/src/app-server/transport-stdio.test.ts
View file

@ -4,11 +4,16 @@ import type { CodexAppServerStartOptions } from "./config.js";
import { createStdioTransport, resolveCodexAppServerSpawnEnv } from "./transport-stdio.js";
const spawnMock = vi.hoisted(() => vi.fn(() => ({ pid: 1234 })));
const prepareRegistration = vi.hoisted(() => vi.fn(async () => async () => {}));
vi.mock("node:child_process", () => ({ spawn: spawnMock }));
vi.mock("./transport-process-registration.js", () => ({
prepareCodexAppServerProcessRegistration: prepareRegistration,
}));
beforeEach(() => {
spawnMock.mockClear();
prepareRegistration.mockReset().mockResolvedValue(async () => {});
});
function startOptions(command: string): CodexAppServerStartOptions {
@ -21,8 +26,23 @@ function startOptions(command: string): CodexAppServerStartOptions {
}
describe("createStdioTransport", () => {
it("spawns a compatibility endpoint in its configured working directory", () => {
createStdioTransport({
it("rechecks authority after orphan cleanup before spawning", async () => {
let active = true;
prepareRegistration.mockImplementationOnce(async () => {
active = false;
return async () => {};
});
await expect(
createStdioTransport(startOptions("codex"), {}, () => {
if (!active) {
throw new Error("owner closed");
}
}),
).rejects.toThrow("owner closed");
expect(spawnMock).not.toHaveBeenCalled();
});
it("spawns a compatibility endpoint in its configured working directory", async () => {
await createStdioTransport({
...startOptions("codex"),
cwd: "/srv/codex-project",
});
@ -34,7 +54,7 @@ describe("createStdioTransport", () => {
);
});
it("preserves wrapper prefixes, root option values, and raw override ordering", () => {
it("preserves wrapper prefixes, root option values, and raw override ordering", async () => {
const overrides = ["-c", 'developer_instructions="app-server = literal"'];
const args = [
"/wrapper.js",
@ -46,7 +66,7 @@ describe("createStdioTransport", () => {
"stdio://",
"--config=model_reasoning_effort=high",
];
createStdioTransport({ ...startOptions("node"), args });
await createStdioTransport({ ...startOptions("node"), args });
expect(spawnMock).toHaveBeenCalledWith(
"node",
@ -65,23 +85,26 @@ describe("createStdioTransport", () => {
expect(args[1]).toBe("-c");
});
it("does not reinterpret a wrapper's positional arguments after --", () => {
it("does not reinterpret a wrapper's positional arguments after --", async () => {
const args = ["/wrapper.js", "--", "-c", "opaque", "app-server"];
createStdioTransport({ ...startOptions("node"), args });
await createStdioTransport({ ...startOptions("node"), args });
expect(spawnMock).toHaveBeenCalledWith("node", args, expect.any(Object));
});
it.each(["--ws-issuer", "--ws-audience"])("preserves a subcommand-shaped %s value", (flag) => {
createStdioTransport({
...startOptions("codex"),
args: ["app-server", flag, "app-server", "-c", "model_reasoning_effort=high"],
});
expect(spawnMock).toHaveBeenCalledWith(
"codex",
["-c", "model_reasoning_effort=high", "app-server", flag, "app-server"],
expect.any(Object),
);
});
it.each(["--ws-issuer", "--ws-audience"])(
"preserves a subcommand-shaped %s value",
async (flag) => {
await createStdioTransport({
...startOptions("codex"),
args: ["app-server", flag, "app-server", "-c", "model_reasoning_effort=high"],
});
expect(spawnMock).toHaveBeenCalledWith(
"codex",
["-c", "model_reasoning_effort=high", "app-server", flag, "app-server"],
expect.any(Object),
);
},
);
});
describe("resolveCodexAppServerSpawnEnv", () => {

21
extensions/codex/src/app-server/transport-stdio.ts
View file

@ -9,6 +9,7 @@ import {
} from "openclaw/plugin-sdk/windows-spawn";
import type { CodexAppServerStartOptions } from "./config.js";
import { normalizeCodexAppServerArgs } from "./launch-args.js";
import { prepareCodexAppServerProcessRegistration } from "./transport-process-registration.js";
const UNSAFE_ENVIRONMENT_KEYS = new Set(["__proto__", "constructor", "prototype"]);
const RUNTIME_INJECTION_ENVIRONMENT_KEYS = new Set([
@ -125,17 +126,21 @@ function copySafeEnvironmentEntries(
}
/** Spawns the Codex app-server process and returns the shared transport interface. */
export function createStdioTransport(
export async function createStdioTransport(
options: CodexAppServerStartOptions,
baseEnv: NodeJS.ProcessEnv = process.env,
): ChildProcessWithoutNullStreams {
assertCurrent?: () => void,
onSpawn?: (child: ChildProcessWithoutNullStreams) => void,
): Promise<ChildProcessWithoutNullStreams> {
const env = resolveCodexAppServerSpawnEnv(options, baseEnv);
const invocation = resolveCodexAppServerSpawnInvocation(options, {
platform: process.platform,
env,
execPath: process.execPath,
});
return spawn(invocation.command, invocation.args, {
const register = await prepareCodexAppServerProcessRegistration();
assertCurrent?.();
const child = spawn(invocation.command, invocation.args, {
// Preserve the shipped Supervisor endpoint contract: relative commands and
// config discovery may depend on the endpoint's process working directory.
...(options.cwd !== undefined ? { cwd: options.cwd } : {}),
@ -145,4 +150,14 @@ export function createStdioTransport(
stdio: ["pipe", "pipe", "pipe"],
windowsHide: invocation.windowsHide,
});
try {
// Attach lifecycle observers before inspection can yield to an early exit.
onSpawn?.(child);
await register(child);
assertCurrent?.();
return child;
} catch (error) {
child.kill("SIGKILL");
throw error;
}
}

4
extensions/codex/src/app-server/transport-websocket.test.ts
View file

@ -74,7 +74,7 @@ describe("Codex app-server websocket transport", () => {
if (!address || typeof address === "string") {
throw new Error("expected websocket test server port");
}
const client = CodexAppServerClient.start({
const client = await CodexAppServerClient.start({
transport: "websocket",
url: `ws://127.0.0.1:${address.port}`,
authToken: "secret",
@ -323,7 +323,7 @@ describe("Codex app-server websocket transport", () => {
});
});
const client = CodexAppServerClient.start({
const client = await CodexAppServerClient.start({
transport: "unix",
homeScope: "user",
url: `unix://${socketPath}`,

401
extensions/codex/src/app-server/transport.process.test.ts
View file

@ -1,8 +1,12 @@
import { execFileSync, spawn } from "node:child_process";
import { once } from "node:events";
import fs from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { describe, expect, it } from "vitest";
import { describe, expect, it, vi } from "vitest";
import { terminateCodexAppServerOrphan } from "./transport-process-containment.js";
import * as processSnapshot from "./transport-process-snapshot.js";
import type { PosixProcess } from "./transport-process-snapshot.js";
import { closeCodexAppServerTransportAndWait } from "./transport.js";
type FixtureEvent = {
@ -86,6 +90,38 @@ async function removeTaskOwnedFixtureProcesses(tempDir: string): Promise<void> {
}
describe.skipIf(process.platform === "win32")("Codex app-server process containment", () => {
it.each(["startedAt", "pgid"] as const)(
"does not signal a registered PID with a changed %s",
async (field) => {
const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], {
detached: true,
stdio: "ignore",
});
await once(child, "spawn");
try {
const identity = (await processSnapshot.readCodexAppServerProcessSnapshot())?.find(
(row) => row.pid === child.pid,
);
if (!identity) {
throw new Error("Missing test process identity");
}
const stale =
field === "startedAt"
? { ...identity, startedAt: "Mon Jan 1 00:00:00 2001" }
: { ...identity, pgid: identity.pgid + 1 };
// PID reuse retires a stale row; group drift remains ambiguous and blocks startup.
expect(await terminateCodexAppServerOrphan(stale)).toBe(field === "startedAt");
expect(child.exitCode).toBeNull();
expect(child.signalCode).toBeNull();
expect(process.kill(child.pid!, 0)).toBe(true);
} finally {
const exited = once(child, "exit");
child.kill("SIGKILL");
await exited;
}
},
);
it("reaps descendants in independent and root process groups before close returns", async () => {
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-transport-process-"));
const logPath = path.join(tempDir, "processes.jsonl");
@ -171,17 +207,21 @@ process.stdin.on("end", () => process.exit(0));
}
});
it("revalidates and retains every identity proven while quiescing", async () => {
it.each([
["reuse", true],
["late", false],
["reparented", false],
["root-resumed", false],
["traced", false],
["uninterruptible", false],
["snapshot-failure", true],
["inspection-timeout", true],
["extended", false],
] as const)("revalidates identities while quiescing: %s", async (mode, sentinelSurvived) => {
const tempDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-codex-identity-reuse-"));
const rootPath = path.join(tempDir, "root.mjs");
const sentinelPath = path.join(tempDir, "sentinel.mjs");
const sentinelPidPath = path.join(tempDir, "sentinel.pid");
const driverPath = path.join(tempDir, "driver.mts");
const fakeBin = path.join(tempDir, "bin");
const fakePsPath = path.join(fakeBin, "ps");
const fakePsCounterPath = path.join(tempDir, "ps-count");
const scenarioPath = path.join(tempDir, "ps-scenario.json");
await fs.mkdir(fakeBin);
await fs.writeFile(
sentinelPath,
`
@ -202,230 +242,135 @@ process.stdin.resume();
process.stdin.on("end", () => process.exit(0));
`,
);
await fs.writeFile(
fakePsPath,
`#!/bin/sh
count=0
if [ -f ${JSON.stringify(fakePsCounterPath)} ]; then count=$(command cat ${JSON.stringify(fakePsCounterPath)}); fi
printf '%s' "$((count + 1))" > ${JSON.stringify(fakePsCounterPath)}
last=$(command cat ${JSON.stringify(scenarioPath)})
if [ "$count" -gt "$last" ]; then count=$last; fi
rows=${JSON.stringify(scenarioPath)}.$count
payload=$(command cat "$rows")
if [ "$payload" = FAIL ]; then exit 1; fi
if [ "$payload" = HANG ]; then while :; do sleep 1; done; fi
printf '%s\\n' "$payload"
`,
{ mode: 0o755 },
);
await fs.writeFile(
driverPath,
`
import { execFileSync, spawn } from "node:child_process";
import fs from "node:fs/promises";
import path from "node:path";
import { pathToFileURL } from "node:url";
const [mode, transportPath, rootPath, sentinelPath, sentinelPidPath, fakeBin, counterPath, scenarioPath, tempDir] = process.argv.slice(2);
const { closeCodexAppServerTransportAndWait } = await import(pathToFileURL(transportPath).href);
const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const readIdentity = (pid) => {
const line = execFileSync("/bin/ps", ["-o", "pid=,ppid=,pgid=,stat=,lstart=", "-p", String(pid)], { encoding: "utf8" }).trim();
const match = /^(\\d+)\\s+(\\d+)\\s+(\\d+)\\s+(\\S+)\\s+(.+)$/.exec(line);
if (!match) throw new Error("unexpected process identity row: " + line);
return { pid: Number(match[1]), ppid: Number(match[2]), pgid: Number(match[3]), state: match[4], startedAt: match[5] };
};
const waitForFile = async (filePath) => {
const deadline = Date.now() + 5_000;
while (Date.now() < deadline) {
const contents = await fs.readFile(filePath, "utf8").catch(() => "");
if (contents) return contents;
await delay(20);
}
throw new Error("timed out waiting for sentinel PID");
};
const commandForPid = (pid) => {
try { return execFileSync("/bin/ps", ["-o", "command=", "-p", String(pid)], { encoding: "utf8" }).trim(); }
catch { return ""; }
};
const stoppedForPid = (pid) => {
try { return /^[Tt]/.test(execFileSync("/bin/ps", ["-o", "stat=", "-p", String(pid)], { encoding: "utf8" }).trim()); }
catch { return false; }
};
const killOwned = (pid) => {
if (!commandForPid(pid).includes(tempDir)) return;
try { process.kill(pid, "SIGKILL"); } catch {}
};
const originalPath = process.env.PATH;
let root;
let sentinelPid;
let result;
try {
await fs.rm(sentinelPidPath, { force: true });
root = spawn(process.execPath, [rootPath, sentinelPath, sentinelPidPath], { detached: true, stdio: ["pipe", "pipe", "pipe"] });
sentinelPid = Number((await waitForFile(sentinelPidPath)).trim());
const rootIdentity = readIdentity(root.pid);
const sentinelIdentity = readIdentity(sentinelPid);
const stoppedRoot = { ...rootIdentity, state: "T" };
const oldSentinel = { ...sentinelIdentity, ppid: root.pid, state: "S", startedAt: "Mon Jan 1 00:00:00 2001" };
const stoppedOldSentinel = { ...oldSentinel, state: "T" };
const stoppedSentinel = { ...sentinelIdentity, state: "T" };
const snapshots =
mode === "reuse"
? [
[rootIdentity, oldSentinel],
[rootIdentity, oldSentinel],
[stoppedRoot, oldSentinel],
[stoppedRoot, oldSentinel],
[stoppedRoot, stoppedOldSentinel],
[stoppedRoot, stoppedSentinel],
]
: mode === "late"
? [
[rootIdentity],
[rootIdentity],
[stoppedRoot, sentinelIdentity],
[stoppedRoot, sentinelIdentity],
[stoppedRoot, stoppedSentinel],
]
: mode === "reparented"
? [
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
[
stoppedRoot,
{ ...stoppedSentinel, ppid: 1, pgid: stoppedSentinel.pgid + 1 },
],
]
: mode === "root-resumed"
? [
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...stoppedSentinel, ppid: root.pid }],
]
: mode === "traced"
? [
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
[
stoppedRoot,
{ ...stoppedSentinel, ppid: root.pid, state: "t" },
],
]
: mode === "uninterruptible"
? [
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[
stoppedRoot,
{ ...sentinelIdentity, ppid: root.pid, state: "U" },
],
[
stoppedRoot,
{ ...sentinelIdentity, ppid: root.pid, state: "U" },
],
]
: mode === "snapshot-failure"
? [
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
null,
]
: mode === "inspection-timeout"
? [
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
"HANG",
]
: [
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
[rootIdentity, { ...sentinelIdentity, ppid: root.pid }],
...Array.from({ length: 8 }, () => [
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
[stoppedRoot, { ...sentinelIdentity, ppid: root.pid }],
]).flat(),
[stoppedRoot, { ...stoppedSentinel, ppid: root.pid }],
];
await fs.writeFile(counterPath, "0");
await fs.writeFile(scenarioPath, String(snapshots.length - 1));
await Promise.all(snapshots.map(async (rows, index) => {
const contents = rows === null
? "FAIL\\n"
: rows === "HANG"
? "HANG\\n"
: rows.map((row) => [row.pid, row.ppid, row.pgid, row.state, row.startedAt].join(" ")).join("\\n") + "\\n";
await fs.writeFile(scenarioPath + "." + index, contents);
}));
process.env.PATH = fakeBin + path.delimiter + (originalPath ?? "");
const closed = await closeCodexAppServerTransportAndWait(root, { forceKillDelayMs: 500, exitTimeoutMs: 2_000 });
process.env.PATH = originalPath;
result = {
closed,
rootExitCode: root.exitCode,
sentinelSurvived: commandForPid(sentinelPid).includes(tempDir),
sentinelStopped: stoppedForPid(sentinelPid),
};
} finally {
process.env.PATH = originalPath;
if (sentinelPid) killOwned(sentinelPid);
if (root?.pid) killOwned(root.pid);
}
process.stdout.write(JSON.stringify(result));
`,
);
const root = spawn(process.execPath, [rootPath, sentinelPath, sentinelPidPath], {
detached: true,
stdio: ["pipe", "pipe", "pipe"],
});
let restoreInspection: (() => void) | undefined;
try {
const transportPath = path.resolve("extensions/codex/src/app-server/transport.ts");
for (const [mode, sentinelSurvived, sentinelStopped] of [
["reuse", true],
["late", false],
["reparented", false],
["root-resumed", false],
["traced", false],
["uninterruptible", false],
["snapshot-failure", true, false],
["inspection-timeout", true, false],
["extended", false],
] as const) {
const output = execFileSync(
process.execPath,
[
"--import",
"tsx",
driverPath,
mode,
transportPath,
rootPath,
sentinelPath,
sentinelPidPath,
fakeBin,
fakePsCounterPath,
scenarioPath,
tempDir,
],
{ cwd: path.resolve("."), encoding: "utf8", timeout: 30_000 },
);
const result = JSON.parse(output) as {
closed: boolean;
rootExitCode: number | null;
sentinelSurvived: boolean;
sentinelStopped: boolean;
};
expect(result).toMatchObject({ closed: true, rootExitCode: 0, sentinelSurvived });
if (sentinelStopped !== undefined) {
expect(result.sentinelStopped).toBe(sentinelStopped);
let sentinelPid: number | undefined;
const deadline = Date.now() + 5_000;
while (Date.now() < deadline) {
const contents = await fs.readFile(sentinelPidPath, "utf8").catch(() => "");
if (contents) {
sentinelPid = Number(contents);
break;
}
await delay(20);
}
const initial = await processSnapshot.readCodexAppServerProcessSnapshot();
const rootIdentity = initial?.find((row) => row.pid === root.pid);
const sentinelIdentity = initial?.find((row) => row.pid === sentinelPid);
if (!rootIdentity || !sentinelIdentity) {
throw new Error("Missing root or sentinel process identity");
}
const stoppedRoot = { ...rootIdentity, state: "T" };
const oldSentinel = {
...sentinelIdentity,
ppid: rootIdentity.pid,
state: "S",
startedAt: "Mon Jan 1 00:00:00 2001",
};
const stoppedOldSentinel = { ...oldSentinel, state: "T" };
const stoppedSentinel = { ...sentinelIdentity, state: "T" };
const runningTree = [rootIdentity, sentinelIdentity];
const rootStoppedTree = [stoppedRoot, sentinelIdentity];
const stoppedTree = [stoppedRoot, stoppedSentinel];
const scenarios: Record<typeof mode, Array<PosixProcess[] | undefined | "deadline">> = {
reuse: [
[rootIdentity, oldSentinel],
[rootIdentity, oldSentinel],
[stoppedRoot, oldSentinel],
[stoppedRoot, oldSentinel],
[stoppedRoot, stoppedOldSentinel],
stoppedTree,
],
late: [[rootIdentity], [rootIdentity], rootStoppedTree, rootStoppedTree, stoppedTree],
reparented: [
runningTree,
runningTree,
rootStoppedTree,
rootStoppedTree,
[stoppedRoot, { ...stoppedSentinel, ppid: 1, pgid: stoppedSentinel.pgid + 1 }],
],
"root-resumed": [
runningTree,
runningTree,
runningTree,
runningTree,
rootStoppedTree,
rootStoppedTree,
stoppedTree,
],
traced: [
runningTree,
runningTree,
rootStoppedTree,
rootStoppedTree,
[stoppedRoot, { ...stoppedSentinel, state: "t" }],
],
uninterruptible: [
runningTree,
runningTree,
[stoppedRoot, { ...sentinelIdentity, state: "U" }],
[stoppedRoot, { ...sentinelIdentity, state: "U" }],
],
"snapshot-failure": [runningTree, runningTree, rootStoppedTree, rootStoppedTree, undefined],
"inspection-timeout": [runningTree, runningTree, "deadline"],
extended: [
runningTree,
runningTree,
...Array.from({ length: 16 }, () => rootStoppedTree),
stoppedTree,
],
};
const snapshots = scenarios[mode];
let inspection = 0;
const readSnapshot = async (
inspectionDeadline: number,
): Promise<PosixProcess[] | undefined> => {
const rows = snapshots[Math.min(inspection++, snapshots.length - 1)];
if (rows === "deadline") {
await delay(Math.max(1, inspectionDeadline - Date.now()));
return undefined;
}
return rows;
};
const snapshotSpy = vi
.spyOn(processSnapshot, "readCodexAppServerProcessSnapshot")
.mockImplementation((inspectionDeadline = Date.now() + 2_000) =>
readSnapshot(inspectionDeadline),
);
const processSpy = vi
.spyOn(processSnapshot, "readCodexAppServerProcess")
.mockImplementation(async (pid, inspectionDeadline) =>
(await readSnapshot(inspectionDeadline))?.find((row) => row.pid === pid),
);
restoreInspection = () => {
snapshotSpy.mockRestore();
processSpy.mockRestore();
};
const closed = await closeCodexAppServerTransportAndWait(root, {
forceKillDelayMs: 500,
exitTimeoutMs: 2_000,
});
restoreInspection();
expect(closed).toBe(true);
expect(root.exitCode).toBe(0);
const survived = listProcesses().some(
(row) => row.pid === sentinelPid && row.command.includes(tempDir),
);
expect(survived).toBe(sentinelSurvived);
if (mode === "snapshot-failure" || mode === "inspection-timeout") {
const sentinel = await processSnapshot.readCodexAppServerProcess(
sentinelIdentity.pid,
Date.now() + 2_000,
);
expect(sentinel).toBeDefined();
expect(sentinel?.state).not.toMatch(/^[Tt]/);
}
} finally {
restoreInspection?.();
await removeTaskOwnedFixtureProcesses(tempDir);
await fs.rm(tempDir, { recursive: true, force: true });
}

8
extensions/codex/src/node-exec-server.runtime.ts
View file

@ -209,7 +209,7 @@ export async function runCodexNodeExecServer(params: {
}
// Awaited setup is complete; policy and invocation closure win at spawn.
params.assertExecAuthorized();
const child = createStdioTransport(
const child = await createStdioTransport(
{
transport: "stdio",
command: native,
@ -225,6 +225,12 @@ export async function runCodexNodeExecServer(params: {
clearEnv: ["NODE_OPTIONS"],
},
baseEnv,
() => {
if (io.signal.aborted) {
throw nodeExecServerAbortError(io.signal);
}
params.assertExecAuthorized();
},
);
child.stdin.on("error", (error) => {
rejectDisconnected(error);

146
test/plugins/codex-model-catalog.gateway.test.ts
View file

@ -1,8 +1,12 @@
import { EventEmitter } from "node:events";
import { PassThrough, Writable } from "node:stream";
import { once } from "node:events";
import { mkdtemp, rm } from "node:fs/promises";
import { createServer } from "node:http";
import os from "node:os";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { createTestPluginApi } from "openclaw/plugin-sdk/plugin-test-api";
import { afterEach, describe, expect, it, vi } from "vitest";
import { WebSocketServer } from "ws";
import codexPlugin from "../../extensions/codex/index.js";
import { createAgentHarnessCatalogEvaluator } from "../../src/agents/harness/model-catalog-readiness.js";
import type { AgentHarness } from "../../src/agents/harness/types.js";
@ -26,7 +30,6 @@ import {
import { withEnvAsync } from "../../src/test-utils/env.js";
import { withOpenClawTestState } from "../../src/test-utils/openclaw-test-state.js";
const transport = vi.hoisted(() => ({ spawn: vi.fn() }));
vi.mock("openclaw/plugin-sdk/simple-completion-runtime", () => ({
runHostPreparedIsolatedCompletion: vi.fn(),
}));
@ -36,15 +39,11 @@ vi.mock("openclaw/plugin-sdk/agent-harness-runtime", () => ({
formatErrorMessage: String,
OPENCLAW_VERSION: "test",
}));
vi.mock("node:child_process", async (importOriginal) => ({
...(await importOriginal<typeof import("node:child_process")>()),
spawn: transport.spawn,
}));
describe("models.list native account catalog", () => {
afterEach(() => vi.restoreAllMocks());
it("makes a native user-home API-key catalog selectable without a ChatGPT route", async () => {
it("makes a native user-home API-key catalog selectable without a ChatGPT route", async (ctx) => {
await withOpenClawTestState(
{ layout: "state-only", prefix: "native-catalog-" },
async (state) => {
@ -55,58 +54,73 @@ describe("models.list native account catalog", () => {
SYNTHETIC_ABSENT_KEY: undefined,
},
async () => {
const stdout = new PassThrough();
// macOS Unix sockets have a short path limit; keep them outside the state fixture.
const socketDir = await mkdtemp(
path.join(process.platform === "win32" ? os.tmpdir() : "/tmp", "oc-catalog-"),
);
const socketPath =
process.platform === "win32"
? `\\\\.\\pipe\\${path.basename(socketDir)}`
: path.join(socketDir, "s");
const httpServer = createServer();
const server = new WebSocketServer({ server: httpServer });
ctx.onTestFinished(async () => {
for (const socket of server.clients) {
socket.terminate();
}
await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
});
await new Promise<void>((resolve) => {
httpServer.close(() => resolve());
});
await rm(socketDir, { recursive: true, force: true });
});
const requests: string[] = [];
let account: Record<string, unknown> | null = { type: "apiKey" };
const child = Object.assign(new EventEmitter(), {
stdout,
stderr: new PassThrough(),
stdin: new Writable({
write(chunk, _encoding, callback) {
const request = JSON.parse(chunk.toString()) as { id?: number; method: string };
requests.push(request.method);
if (request.id !== undefined) {
const result =
request.method === "initialize"
? { userAgent: "openclaw/0.149.1 (test)" }
: request.method === "account/read"
? { account, requiresOpenaiAuth: true }
: request.method === "model/list"
? {
data: [
{
id: "synthetic-opaque",
model: "synthetic-opaque",
displayName: "Synthetic name",
description: "Synthetic model",
supportsPersonality: false,
inputModalities: ["text"],
supportedReasoningEfforts: [
{ reasoningEffort: "low", description: "Low" },
],
defaultReasoningEffort: "low",
hidden: false,
isDefault: true,
},
],
nextCursor: null,
}
: {};
queueMicrotask(() =>
stdout.write(`${JSON.stringify({ id: request.id, result })}\n`),
);
}
callback();
},
}),
kill: vi.fn(),
killed: false,
});
child.stdin.once("finish", () => child.emit("exit", 0, null));
transport.spawn.mockImplementation((command: string) => {
expect(command).toBe("/synthetic/codex");
return child;
server.on("connection", (socket) => {
socket.on("message", (data) => {
const encoded = Array.isArray(data)
? Buffer.concat(data)
: Buffer.from(data instanceof ArrayBuffer ? new Uint8Array(data) : data);
const request = JSON.parse(encoded.toString("utf8")) as {
id?: number;
method: string;
};
requests.push(request.method);
if (request.id !== undefined) {
const result =
request.method === "initialize"
? { userAgent: "openclaw/0.149.1 (test)" }
: request.method === "account/read"
? { account, requiresOpenaiAuth: true }
: request.method === "model/list"
? {
data: [
{
id: "synthetic-opaque",
model: "synthetic-opaque",
displayName: "Synthetic name",
description: "Synthetic model",
supportsPersonality: false,
inputModalities: ["text"],
supportedReasoningEfforts: [
{ reasoningEffort: "low", description: "Low" },
],
defaultReasoningEffort: "low",
hidden: false,
isDefault: true,
},
],
nextCursor: null,
}
: {};
socket.send(JSON.stringify({ id: request.id, result }));
}
});
});
httpServer.listen(socketPath);
await once(server, "listening");
const config: OpenClawConfig = {
agents: {
defaults: {
@ -121,7 +135,8 @@ describe("models.list native account catalog", () => {
enabled: true,
config: {
appServer: {
command: "/synthetic/codex",
transport: "unix",
url: `unix://${socketPath}`,
homeScope: "user",
approvalPolicy: "on-request",
sandbox: "workspace-write",
@ -236,9 +251,12 @@ describe("models.list native account catalog", () => {
});
expect(locked.models[0]?.available).toBe(false);
stdout.write(
`${JSON.stringify({ method: "account/updated", params: { authMode: null } })}\n`,
);
for (const socket of server.clients) {
socket.send(
JSON.stringify({ method: "account/updated", params: { authMode: null } }),
);
}
await expect.poll(() => readiness()).toBeUndefined();
expect((await configured()).models[0]?.available).toBe(false);
for (const observed of [
{
@ -313,7 +331,10 @@ describe("models.list native account catalog", () => {
expect(host.models[0]?.available, `host route ${routeIndex}`).toBe(false);
}
expect(requests).not.toContain("account/login/start");
child.emit("exit", 0, null);
for (const socket of server.clients) {
socket.close();
}
await expect.poll(() => readiness()).toBeUndefined();
expect((await configured()).models[0]?.available).toBe(false);
expect(
createAgentHarnessCatalogEvaluator(scope)(rows[0]!, {
@ -339,7 +360,6 @@ describe("models.list native account catalog", () => {
expect((await configured()).models[0]?.available).toBe(false);
} finally {
await harness.dispose?.();
child.emit("exit", 0, null);
restoreActivePluginRegistrySnapshot(previous);
}
},