vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-04 03:11:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

improve(codex): reuse live app-server threads safely (#115089)

Browse Source
This commit is contained in:
Peter Steinberger
2026-07-28 05:08:23 -04:00
committed by GitHub
parent 086d17dd0d
commit 92b4af2dc8
14 changed files with 1155 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
extensions/codex/media-understanding-provider.test.ts
View File

@@ -465,6 +465,9 @@ describe("codex media understanding provider", () => {
});
it("clamps oversized image understanding turn timeouts", async () => {
// The bounded timer subtracts startup time from its clamped deadline.
// Freeze the clock so the clamp assertion cannot lose a real millisecond.
const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(1_700_000_000_000);
const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
try {
const { client } = createFakeClient();
@@ -486,6 +489,7 @@ describe("codex media understanding provider", () => {
expect(result?.text).toBe("A red square.");
expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), MAX_TIMER_TIMEOUT_MS);
} finally {
dateNowSpy.mockRestore();
vi.restoreAllMocks();
vi.clearAllTimers();
vi.useRealTimers();

79
extensions/codex/src/app-server/client-runtime.test.ts
View File

@@ -15,7 +15,11 @@ vi.mock("./rate-limit-cache.js", () => ({
mergeCodexRateLimitsUpdate: mocks.mergeRateLimitUpdate,
}));
const { ensureCodexAppServerClientRuntime } = await import("./client-runtime.js");
const {
consumeCodexAppServerLiveThread,
ensureCodexAppServerClientRuntime,
retainCodexAppServerLiveThread,
} = await import("./client-runtime.js");
describe("Codex app-server client runtime", () => {
const clients: CodexAppServerClient[] = [];
@@ -99,4 +103,77 @@ describe("Codex app-server client runtime", () => {
},
});
});
it("retains and consumes only one subscribed thread per physical client", async () => {
const harness = createClientHarness();
clients.push(harness.client);
await expect(
retainCodexAppServerLiveThread(harness.client, "thread-before-runtime"),
).resolves.toBeUndefined();
ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" });
await expect(retainCodexAppServerLiveThread(harness.client, "thread-1")).resolves.toEqual({});
await expect(consumeCodexAppServerLiveThread(harness.client, "thread-2")).resolves.toBe(false);
await expect(retainCodexAppServerLiveThread(harness.client, "thread-2")).resolves.toEqual({
previousThreadId: "thread-1",
});
await expect(consumeCodexAppServerLiveThread(harness.client, "thread-1")).resolves.toBe(false);
await expect(consumeCodexAppServerLiveThread(harness.client, "thread-2")).resolves.toBe(true);
await expect(consumeCodexAppServerLiveThread(harness.client, "thread-2")).resolves.toBe(false);
});
it("waits for the old subscription to release before another thread can acquire it", async () => {
const harness = createClientHarness();
clients.push(harness.client);
ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" });
await retainCodexAppServerLiveThread(harness.client, "thread-1");
let finishRelease: (() => void) | undefined;
const previousRelease = new Promise<void>((resolve) => {
finishRelease = resolve;
});
const transition = retainCodexAppServerLiveThread(
harness.client,
"thread-2",
async () => previousRelease,
);
const oldThreadAcquisition = consumeCodexAppServerLiveThread(harness.client, "thread-1");
finishRelease?.();
await expect(transition).resolves.toEqual({ previousThreadId: "thread-1" });
await expect(oldThreadAcquisition).resolves.toBe(false);
await expect(consumeCodexAppServerLiveThread(harness.client, "thread-2")).resolves.toBe(true);
});
it("does not expose either thread after a previous subscription release fails", async () => {
const harness = createClientHarness();
clients.push(harness.client);
ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" });
await retainCodexAppServerLiveThread(harness.client, "thread-1");
await expect(
retainCodexAppServerLiveThread(harness.client, "thread-2", async () => {
throw new Error("unsubscribe unavailable");
}),
).rejects.toThrow("unsubscribe unavailable");
await expect(consumeCodexAppServerLiveThread(harness.client, "thread-1")).resolves.toBe(false);
await expect(consumeCodexAppServerLiveThread(harness.client, "thread-2")).resolves.toBe(false);
});
it("reuses a retained subscription only for its complete configuration fingerprint", async () => {
const harness = createClientHarness();
clients.push(harness.client);
ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" });
await expect(
retainCodexAppServerLiveThread(harness.client, "thread-1", undefined, "config-before"),
).resolves.toEqual({});
await expect(
consumeCodexAppServerLiveThread(harness.client, "thread-1", "config-after"),
).resolves.toBe(false);
await expect(
consumeCodexAppServerLiveThread(harness.client, "thread-1", "config-before"),
).resolves.toBe(true);
});
});

65
extensions/codex/src/app-server/client-runtime.ts
View File

@@ -12,6 +12,9 @@ type ClientRuntimeContext = Omit<CodexAppServerAuthProfileLookup, "agentDir"> &
type ClientRuntime = {
context: ClientRuntimeContext;
retainedThreadId?: string;
retainedThreadConfigFingerprint?: string;
retainedThreadRelease?: Promise<void>;
};
const configuredClients = new WeakMap<CodexAppServerClient, ClientRuntime>();
@@ -52,3 +55,65 @@ export function ensureCodexAppServerClientRuntime(
}
});
}
/** Keep at most one idle, still-subscribed thread on a physical Codex client. */
export async function retainCodexAppServerLiveThread(
client: CodexAppServerClient,
threadId: string,
releasePreviousThread?: (previousThreadId: string) => Promise<void>,
configFingerprint?: string,
): Promise<{ previousThreadId?: string } | undefined> {
const runtime = configuredClients.get(client);
if (!runtime) {
return undefined;
}
if (runtime.retainedThreadRelease) {
await runtime.retainedThreadRelease;
}
const previousThreadId = runtime.retainedThreadId;
if (previousThreadId && previousThreadId !== threadId && releasePreviousThread) {
// Keep the old owner visible until its unsubscribe settles; concurrent
// lifecycle acquisition must never resume a thread being released.
const release = releasePreviousThread(previousThreadId);
runtime.retainedThreadRelease = release;
try {
await release;
} catch (error) {
runtime.retainedThreadId = undefined;
runtime.retainedThreadConfigFingerprint = undefined;
throw error;
} finally {
if (runtime.retainedThreadRelease === release) {
runtime.retainedThreadRelease = undefined;
}
}
}
runtime.retainedThreadId = threadId;
runtime.retainedThreadConfigFingerprint = configFingerprint;
return previousThreadId ? { previousThreadId } : {};
}
/** A warm turn can skip resume only when this exact subscription was retained. */
export async function consumeCodexAppServerLiveThread(
client: CodexAppServerClient,
threadId: string,
configFingerprint?: string,
): Promise<boolean> {
const runtime = configuredClients.get(client);
if (!runtime) {
return false;
}
if (runtime.retainedThreadRelease) {
await runtime.retainedThreadRelease;
}
if (
runtime.retainedThreadId !== threadId ||
(configFingerprint !== undefined &&
runtime.retainedThreadConfigFingerprint !== configFingerprint)
) {
return false;
}
runtime.retainedThreadId = undefined;
runtime.retainedThreadConfigFingerprint = undefined;
return true;
}

41
extensions/codex/src/app-server/run-attempt-cleanup.ts
View File

@@ -6,8 +6,12 @@ import {
import { isIncognitoSessionKey } from "../incognito-session.js";
import {
CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS,
closeCodexStartupClientBestEffort,
CodexAppServerUnsafeSubscriptionError,
unsubscribeCodexThreadBestEffort,
} from "./attempt-client-cleanup.js";
import { retainCodexAppServerLiveThread } from "./client-runtime.js";
import { resolveCodexAppServerClientInstanceId } from "./client.js";
import { scheduleCodexNativeHookRelayUnregister } from "./native-hook-relay.js";
import type { CodexAttemptActiveTurn } from "./run-attempt-active-turn.js";
import type { CodexAttemptLifecycleController } from "./run-attempt-lifecycle-controller.js";
@@ -79,6 +83,39 @@ export async function cleanupCodexAttempt(
}
const retainLiveIncognitoThread =
terminalState.turnSucceeded && isIncognitoSessionKey(params.sessionKey);
// Native-preserved and supervision threads have separate ownership and can
// never enter the ordinary persistent warm-thread cache.
const retainedPersistentThread =
terminalState.turnSucceeded &&
!isIncognitoSessionKey(params.sessionKey) &&
params.cleanupBundleMcpOnRunEnd !== true &&
!params.contextEngine &&
resourceState.thread.liveThreadConfigFingerprint !== undefined &&
resourceState.thread.clientId === resolveCodexAppServerClientInstanceId(resourceState.client) &&
resourceState.thread.preserveNativeModel !== true &&
resourceState.thread.connectionScope !== "supervision" &&
!resourceState.thread.ringZeroConfigFingerprint &&
!resourceState.thread.contextEngine
? await retainCodexAppServerLiveThread(
resourceState.client,
resourceState.thread.threadId,
async (previousThreadId) => {
const released = await unsubscribeCodexThreadBestEffort(resourceState.client, {
threadId: previousThreadId,
timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS,
});
if (!released) {
await closeCodexStartupClientBestEffort(resourceState.client);
throw new CodexAppServerUnsafeSubscriptionError(
`Codex retained thread subscription could not be released: ${previousThreadId}`,
);
}
},
resourceState.thread.liveThreadConfigFingerprint,
)
: undefined;
// Replacement waits for the prior unsubscribe before publishing a new slot.
const retainLiveThread = retainLiveIncognitoThread || retainedPersistentThread !== undefined;
const bindingReleased =
isIncognitoSessionKey(params.sessionKey) && !retainLiveIncognitoThread
? await bindingStore.mutate(bindingIdentity, {
@@ -86,8 +123,8 @@ export async function cleanupCodexAttempt(
threadId: resourceState.thread.threadId,
})
: true;
// Successful incognito turns retain the live subscription for cross-turn continuity.
if (!state.timedOut && !retainLiveIncognitoThread) {
// Only explicitly retained live threads may skip the next thread/resume.
if (!state.timedOut && !retainLiveThread) {
// Clear first: if a newer owner won the binding, its live subscription must remain intact.
if (bindingReleased) {
await unsubscribeCodexThreadBestEffort(resourceState.client, {

32
extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts
View File

@@ -169,7 +169,7 @@ describe("Codex app-server main thread cleanup", () => {
await fs.rm(tempDir, { recursive: true, force: true });
});
it("unsubscribes the main Codex thread after a completed turn", async () => {
it("retains a subscribed persistent Codex thread after a completed turn", async () => {
const sessionFile = path.join(tempDir, "session.jsonl");
const workspaceDir = path.join(tempDir, "workspace");
const requests: Array<{ method: string; params: unknown }> = [];
@@ -217,16 +217,26 @@ describe("Codex app-server main thread cleanup", () => {
const result = await run;
expect(readAttemptTerminal(result).aborted).toBe(false);
expect(request).toHaveBeenCalledWith(
"thread/unsubscribe",
{ threadId: "thread-1" },
{ timeoutMs: 5_000 },
);
expect(requests.map((entry) => entry.method)).toEqual([
"thread/start",
"turn/start",
"thread/unsubscribe",
]);
const firstBinding = await readCodexAppServerBinding(sessionFile);
expect({
clientId: firstBinding?.clientId,
threadId: firstBinding?.threadId,
preserveNativeModel: firstBinding?.preserveNativeModel,
connectionScope: firstBinding?.connectionScope,
ringZeroConfigFingerprint: firstBinding?.ringZeroConfigFingerprint,
contextEngine: firstBinding?.contextEngine,
pluginAppsFingerprint: firstBinding?.pluginAppsFingerprint,
}).toEqual({
clientId: "test-client-1",
threadId: "thread-1",
preserveNativeModel: undefined,
connectionScope: undefined,
ringZeroConfigFingerprint: undefined,
contextEngine: undefined,
pluginAppsFingerprint: expect.any(String),
});
expect(requests.map((entry) => entry.method)).toEqual(["thread/start", "turn/start"]);
});
it("keeps an incognito thread subscribed for live in-process reuse", async () => {

16
extensions/codex/src/app-server/run-attempt.context-engine.test.ts
View File

@@ -18,6 +18,7 @@ import {
import { MESSAGE_TOOL_DELIVERY_HINTS } from "openclaw/plugin-sdk/message-tool-delivery-hints";
import { createMockPluginRegistry } from "openclaw/plugin-sdk/plugin-test-runtime";
import { registerSandboxBackend } from "openclaw/plugin-sdk/sandbox";
import { upsertSessionEntry } from "openclaw/plugin-sdk/session-store-runtime";
import { formatSqliteSessionFileMarker } from "openclaw/plugin-sdk/sqlite-runtime-testing";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { readAttemptTerminal } from "./attempt-terminal.test-helper.js";
@@ -123,7 +124,10 @@ function createParams(sessionFile: string, workspaceDir: string): EmbeddedRunAtt
} as EmbeddedRunAttemptParams;
}
function createSqliteParams(workspaceDir: string, storeName: string): EmbeddedRunAttemptParams {
async function createSqliteParams(
workspaceDir: string,
storeName: string,
): Promise<EmbeddedRunAttemptParams> {
const sessionId = "session-1";
const sessionKey = "agent:main:session-1";
const storePath = path.join(tempDir, `${storeName}.sqlite`);
@@ -133,6 +137,12 @@ function createSqliteParams(workspaceDir: string, storeName: string): EmbeddedRu
storePath,
});
const params = createParams(sessionFile, workspaceDir);
await upsertSessionEntry({
agentId: "main",
sessionKey,
storePath,
entry: { sessionFile, sessionId, updatedAt: Date.now() },
});
params.sessionTarget = {
agentId: "main",
sessionId,
@@ -1981,7 +1991,7 @@ describe("runCodexAppServerAttempt context-engine lifecycle", () => {
const maintain = vi.fn(async () => ({ changed: false, bytesFreed: 0, rewrittenEntries: 0 }));
const contextEngine = createContextEngine({ afterTurn, maintain, bootstrap: undefined });
const harness = createStartedThreadHarness();
const params = createSqliteParams(
const params = await createSqliteParams(
workspaceDir,
`heartbeat-${testCase.bootstrapContextRunKind}`,
);
@@ -2113,7 +2123,7 @@ describe("runCodexAppServerAttempt context-engine lifecycle", () => {
bootstrap: undefined,
});
const harness = createStartedThreadHarness();
const params = createSqliteParams(workspaceDir, "prompt-failure");
const params = await createSqliteParams(workspaceDir, "prompt-failure");
params.contextEngine = contextEngine;
const run = runCodexAppServerAttempt(params);

22
extensions/codex/src/app-server/run-attempt.test.ts
View File

@@ -220,11 +220,11 @@ async function writeExistingBinding(
});
}
function attachSqliteSessionTarget(
async function attachSqliteSessionTarget(
params: EmbeddedRunAttemptParams,
storePath: string,
sessionId: string,
): void {
): Promise<void> {
params.sessionId = sessionId;
params.sessionKey = `agent:main:${sessionId}`;
params.sessionTarget = {
@@ -233,6 +233,12 @@ function attachSqliteSessionTarget(
sessionKey: params.sessionKey,
storePath,
};
await upsertSessionEntry({
agentId: "main",
sessionKey: params.sessionKey,
storePath,
entry: { sessionFile: params.sessionFile, sessionId, updatedAt: Date.now() },
});
}
async function readTranscriptMessagesByIdentity(
@@ -1764,7 +1770,7 @@ describe("runCodexAppServerAttempt", () => {
const workspaceDir = path.join(tempDir, "workspace-early-prompt");
const harness = createStartedThreadHarness();
const params = createParams(sessionFile, workspaceDir);
attachSqliteSessionTarget(params, storePath, "session-early-prompt");
await attachSqliteSessionTarget(params, storePath, "session-early-prompt");
params.prompt = "external channel prompt";
const onUserMessagePersisted = vi.fn();
params.onUserMessagePersisted = onUserMessagePersisted;
@@ -1802,7 +1808,7 @@ describe("runCodexAppServerAttempt", () => {
const workspaceDir = path.join(tempDir, "workspace-suppressed-early-prompt");
const harness = createStartedThreadHarness();
const params = createParams(sessionFile, workspaceDir);
attachSqliteSessionTarget(params, storePath, "session-suppressed-early-prompt");
await attachSqliteSessionTarget(params, storePath, "session-suppressed-early-prompt");
params.prompt = "already persisted prompt";
params.suppressNextUserMessagePersistence = true;
const run = runCodexAppServerAttempt(params);
@@ -3538,7 +3544,7 @@ describe("runCodexAppServerAttempt", () => {
const workspaceDir = path.join(tempDir, "workspace-settled-finalization-context");
const harness = createStartedThreadHarness();
const params = createParams(sessionFile, workspaceDir);
attachSqliteSessionTarget(params, storePath, sessionId);
await attachSqliteSessionTarget(params, storePath, sessionId);
params.prompt = "Send the update to Alice.";
const run = runCodexAppServerAttempt(params);
await harness.waitForMethod("turn/start");
@@ -4260,7 +4266,11 @@ describe("runCodexAppServerAttempt", () => {
});
const elicitation = installElicitationClient(request);
const params = createRunParams();
attachSqliteSessionTarget(params, path.join(tempDir, "sessions.json"), "session-computer-use");
await attachSqliteSessionTarget(
params,
path.join(tempDir, "sessions.json"),
"session-computer-use",
);
const run = runCodexAppServerAttempt(params, {
pluginConfig: {
computerUse: {

76
extensions/codex/src/app-server/thread-fingerprints.test.ts
View File

@@ -1,5 +1,79 @@
import { describe, expect, it } from "vitest";
import { readActiveCodexTurnIdsFromResume } from "./thread-fingerprints.js";
import type { JsonObject } from "./protocol.js";
import {
fingerprintCodexThreadConfig,
readActiveCodexTurnIdsFromResume,
} from "./thread-fingerprints.js";
describe("fingerprintCodexThreadConfig", () => {
const request = {
model: "gpt-5.6-sol",
modelProvider: "openai",
approvalPolicy: "never",
approvalsReviewer: "user",
sandbox: "workspace-write",
personality: "none",
serviceTier: "fast",
developerInstructions: "Keep the current conversation private.",
config: { features: { hooks: true, plugins: false } },
};
it("stabilizes equivalent config without exposing instructions or profile", () => {
const fingerprint = fingerprintCodexThreadConfig(request, "openai:personal");
expect(fingerprint).toBe(
fingerprintCodexThreadConfig(
{ ...request, config: { features: { plugins: false, hooks: true } } },
"openai:personal",
),
);
expect(fingerprint).not.toContain("private");
expect(fingerprint).not.toContain("openai:personal");
});
it.each<{ setting: string; patch: JsonObject }>([
{ setting: "model", patch: { model: "gpt-5.6-terra" } },
{ setting: "model provider", patch: { modelProvider: "custom" } },
{ setting: "requested model provider", patch: { requestedModelProvider: "custom" } },
{ setting: "approval policy", patch: { approvalPolicy: "on-request" } },
{ setting: "approval reviewer", patch: { approvalsReviewer: "guardian" } },
{ setting: "sandbox", patch: { sandbox: "read-only" } },
{ setting: "service tier", patch: { serviceTier: "flex" } },
{ setting: "base instructions", patch: { baseInstructions: "Different base policy." } },
{ setting: "developer instructions", patch: { developerInstructions: "Different policy." } },
{ setting: "effective config", patch: { config: { features: { hooks: false } } } },
])("invalidates reuse when $setting changes", ({ patch }) => {
expect(fingerprintCodexThreadConfig({ ...request, ...patch }, "openai:personal")).not.toBe(
fingerprintCodexThreadConfig(request, "openai:personal"),
);
});
it("invalidates reuse when the selected authentication profile changes", () => {
expect(fingerprintCodexThreadConfig(request, "openai:work")).not.toBe(
fingerprintCodexThreadConfig(request, "openai:personal"),
);
});
it("invalidates reuse when the native dynamic tool catalog changes", () => {
expect(fingerprintCodexThreadConfig(request, "openai:personal", "tools-after")).not.toBe(
fingerprintCodexThreadConfig(request, "openai:personal", "tools-before"),
);
});
it("distinguishes an omitted service tier from an explicit clear", () => {
const { serviceTier: _serviceTier, ...withoutServiceTier } = request;
expect(fingerprintCodexThreadConfig(withoutServiceTier, "openai:personal")).not.toBe(
fingerprintCodexThreadConfig({ ...withoutServiceTier, serviceTier: null }, "openai:personal"),
);
});
it("preserves an explicitly omitted native model selection", () => {
expect(
fingerprintCodexThreadConfig({ ...request, requestedModel: null }, "openai:personal"),
).not.toBe(fingerprintCodexThreadConfig(request, "openai:personal"));
});
});
describe("readActiveCodexTurnIdsFromResume", () => {
it("uses the bounded initial turns page when Codex returns one", () => {

31
extensions/codex/src/app-server/thread-fingerprints.ts
View File

@@ -85,6 +85,37 @@ export function fingerprintJsonObject(value: JsonObject): string {
return JSON.stringify(stabilizeJsonValue(value));
}
/** Hash every resume-visible setting without retaining config, credentials, or instructions. */
export function fingerprintCodexThreadConfig(
request: JsonObject,
authProfileId?: string,
dynamicToolsFingerprint?: string,
): string {
return hashCodexAppServerBindingFingerprint(
fingerprintJsonObject({
authProfileId: authProfileId ?? null,
dynamicToolsFingerprint: dynamicToolsFingerprint ?? null,
model: request.model ?? null,
requestedModel:
request.requestedModel === undefined ? (request.model ?? null) : request.requestedModel,
modelProvider: request.modelProvider ?? null,
requestedModelProvider:
request.requestedModelProvider === undefined
? (request.modelProvider ?? null)
: request.requestedModelProvider,
approvalPolicy: request.approvalPolicy ?? null,
approvalsReviewer: request.approvalsReviewer ?? null,
sandbox: request.sandbox ?? null,
permissions: request.permissions ?? null,
personality: request.personality ?? null,
serviceTier: request.serviceTier === undefined ? "<omitted>" : request.serviceTier,
baseInstructions: request.baseInstructions ?? null,
developerInstructions: request.developerInstructions ?? null,
config: request.config ?? {},
}),
);
}
export function fingerprintEnvironmentSelection(
environments: CodexTurnEnvironmentParams[] | undefined,
): string | undefined {

69
extensions/codex/src/app-server/thread-lifecycle-io.ts
View File

@@ -26,7 +26,10 @@ import type {
CodexAppServerThreadBinding,
} from "./session-binding.js";
import { isCodexAppServerStartSelectionChangedError } from "./shared-client.js";
import { readActiveCodexTurnIdsFromResume } from "./thread-fingerprints.js";
import {
fingerprintCodexThreadConfig,
readActiveCodexTurnIdsFromResume,
} from "./thread-fingerprints.js";
import {
CodexAdoptedThreadActiveError,
CodexRingZeroAttestationError,
@@ -77,6 +80,10 @@ type ThreadRequestContext = {
type ResumeThreadContext = ThreadRequestContext & {
binding: CodexAppServerThreadBinding;
clearCurrentBinding: (operation: string) => Promise<void>;
prebuiltFinalConfigPatch?: {
configPatch?: JsonObject;
nativeHookRelayGeneration?: string;
};
};
type StartThreadContext = ThreadRequestContext & {
@@ -131,13 +138,14 @@ export async function resumeExistingCodexThread(
resumeBinding.connectionScope === "supervision"
? undefined
: (params.params.authProfileId ?? resumeBinding.authProfileId);
const finalConfigPatch = params.buildFinalConfigPatch?.({
action: "resume",
binding: resumeBinding,
}) ?? {
configPatch: params.finalConfigPatch,
nativeHookRelayGeneration: params.nativeHookRelayGeneration,
};
const finalConfigPatch = context.prebuiltFinalConfigPatch ??
params.buildFinalConfigPatch?.({
action: "resume",
binding: resumeBinding,
}) ?? {
configPatch: params.finalConfigPatch,
nativeHookRelayGeneration: params.nativeHookRelayGeneration,
};
// Codex rebuilds effective config on thread/resume, so replay the app
// allowlist persisted at thread/start or plugin tools disappear after one turn.
const pluginAppsConfigPatch =
@@ -289,6 +297,25 @@ export async function resumeExistingCodexThread(
...resumeBinding,
threadId: response.thread.id,
...resumePatch,
liveThreadConfigFingerprint: fingerprintCodexThreadConfig(
{
...resumeParams,
model:
resumeBinding.preserveNativeModel === true
? null
: (response.model ?? resumeParams.model ?? null),
requestedModel:
resumeBinding.preserveNativeModel === true ? null : (resumeParams.model ?? null),
modelProvider:
resumeBinding.preserveNativeModel === true ? null : (resumePatch.modelProvider ?? null),
requestedModelProvider:
resumeBinding.preserveNativeModel === true
? null
: (resumeParams.modelProvider ?? resumePatch.modelProvider ?? null),
},
authProfileId,
dynamicToolsFingerprint,
),
lifecycle: {
action: "resumed",
...(activeTurnIds.length ? { activeTurnIds } : {}),
@@ -443,6 +470,10 @@ export async function startFreshCodexThread(
agentDir: params.params.agentDir,
config: params.params.config,
});
const bindingModelProvider = normalizeBindingModelProvider(
params.params.authProfileId,
response.modelProvider ?? requestModelProvider ?? startModelProvider ?? modelProvider,
);
const nextMcpServersFingerprint =
params.mcpServersFingerprintEvaluated === true ? params.mcpServersFingerprint : undefined;
if (!preserveExistingBinding) {
@@ -457,10 +488,7 @@ export async function startFreshCodexThread(
...(rolloutPath ? { rolloutPath } : {}),
authProfileId: params.params.authProfileId,
model: response.model ?? startParams.model ?? params.params.modelId,
modelProvider: normalizeBindingModelProvider(
params.params.authProfileId,
response.modelProvider ?? requestModelProvider ?? startModelProvider ?? modelProvider,
),
modelProvider: bindingModelProvider,
dynamicToolsFingerprint,
dynamicToolsContainDeferred,
webSearchThreadConfigFingerprint,
@@ -527,6 +555,23 @@ export async function startFreshCodexThread(
pluginAppPolicyContext: pluginThreadConfig?.policyContext,
contextEngine: contextEngineBinding,
environmentSelectionFingerprint,
// Transient starts do not own the persisted binding, so their native
// subscriptions must be released instead of entering the warm cache.
...(!preserveExistingBinding
? {
liveThreadConfigFingerprint: fingerprintCodexThreadConfig(
{
...startParams,
model: response.model ?? startParams.model ?? null,
requestedModel: startParams.model ?? null,
modelProvider: bindingModelProvider ?? null,
requestedModelProvider: startParams.modelProvider ?? bindingModelProvider ?? null,
},
params.params.authProfileId,
dynamicToolsFingerprint,
),
}
: {}),
lifecycle: {
action: "started",
...(rotatedContextEngineBinding ? { rotatedContextEngineBinding } : {}),

66
extensions/codex/src/app-server/thread-lifecycle-run.ts
View File

@@ -57,6 +57,12 @@ import type {
CodexAppServerThreadLifecycleBinding,
CodexStartOrResumeThreadParams,
} from "./thread-lifecycle-types.js";
import {
releaseCodexConsumedLiveThread,
releaseCodexRetainedLiveThread,
throwIfCodexThreadLifecycleAborted,
tryReuseCodexLiveThread,
} from "./thread-lifecycle-warm.js";
import { resolveCodexAppServerThreadModelSelection } from "./thread-model-selection.js";
import {
assertCodexRingZeroHasNoManagedHooks,
@@ -163,6 +169,7 @@ export async function startOrResumeThread(
let binding = await lifecycleTiming.measure("read-binding", () =>
params.bindingStore.read(bindingIdentity),
);
const initialBoundThreadId = binding?.threadId;
const normalizeBindingModelProvider = (
authProfileId: string | undefined,
modelProvider: string | undefined,
@@ -174,22 +181,14 @@ export async function startOrResumeThread(
agentDir: params.params.agentDir,
config: params.params.config,
});
const throwIfAborted = () => {
if (!params.signal?.aborted) {
return;
}
const reason = params.signal.reason;
if (reason instanceof Error) {
throw reason;
}
const error = new Error(
typeof reason === "string" && reason.length > 0
? reason
: "codex app-server thread lifecycle aborted",
);
error.name = "AbortError";
throw error;
};
const throwIfAborted = () => throwIfCodexThreadLifecycleAborted(params.signal);
const releaseRetainedThread = (threadId: string) =>
releaseCodexRetainedLiveThread({
client: params.client,
abandonClient: params.abandonClient,
lifecycleTiming,
threadId,
});
if (!binding && bindingIdentity.kind === "session" && bindingIdentity.sessionKey) {
// Reset may rotate the OpenClaw session while this plugin is unloaded. Only
// the authoritative session store may let its successor displace that stale owner.
@@ -205,6 +204,7 @@ export async function startOrResumeThread(
}
}
if (binding?.pendingSupervisionBranch) {
await releaseRetainedThread(binding.threadId);
const pendingBinding = binding as CodexAppServerThreadBinding & {
pendingSupervisionBranch: CodexAppServerPendingSupervisionBranch;
};
@@ -630,6 +630,36 @@ export async function startOrResumeThread(
await clearCurrentBinding("rotating an unavailable ephemeral thread binding");
binding = undefined;
} else {
const warmReuse = await tryReuseCodexLiveThread({
params,
binding,
bindingIdentity,
clientId,
contextEngineBinding,
dynamicToolsFingerprint,
hostSystemAgentActive,
lifecycleTiming,
releaseConsumedThread: (threadId, cause) =>
releaseCodexConsumedLiveThread({
client: params.client,
abandonClient: params.abandonClient,
lifecycleTiming,
threadId,
cause,
}),
ringZeroActive,
ringZeroInheritedMcpServerNames,
startModelProvider,
startModelSelection,
throwIfAborted,
userMcpServersConfigPatch,
});
if (warmReuse.binding) {
return warmReuse.binding;
}
// Codex cold-resumes a changed idle thread after its last subscriber
// leaves; resume_running_thread shuts down the old cached session.
await releaseRetainedThread(binding.threadId);
const resumed = await resumeExistingCodexThread(params, {
binding,
bindingIdentity,
@@ -652,6 +682,7 @@ export async function startOrResumeThread(
normalizeBindingModelProvider,
throwIfAborted,
clearCurrentBinding,
prebuiltFinalConfigPatch: warmReuse.prebuiltFinalConfigPatch,
});
if (resumed) {
return resumed;
@@ -659,6 +690,9 @@ export async function startOrResumeThread(
}
}
if (initialBoundThreadId) {
await releaseRetainedThread(initialBoundThreadId);
}
return await startFreshCodexThread(params, {
bindingIdentity,
startModelSelection,

1
extensions/codex/src/app-server/thread-lifecycle-types.ts
View File

@@ -16,6 +16,7 @@ type CodexAppServerThreadLifecycle = {
export type CodexAppServerThreadLifecycleBinding = CodexAppServerThreadBinding & {
lifecycle: CodexAppServerThreadLifecycle;
liveThreadConfigFingerprint?: string;
};
type CodexThreadFinalConfigPatchDecision =

243
extensions/codex/src/app-server/thread-lifecycle-warm.ts Normal file
View File

@@ -0,0 +1,243 @@
import {
CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS,
closeCodexStartupClientBestEffort,
CodexAppServerUnsafeSubscriptionError,
unsubscribeCodexThreadBestEffort,
} from "./attempt-client-cleanup.js";
import { consumeCodexAppServerLiveThread } from "./client-runtime.js";
import type { CodexAppServerClient } from "./client.js";
import {
buildCodexPluginAppsConfigPatchFromPolicyContext,
mergeCodexThreadConfigs,
} from "./plugin-thread-config.js";
import type { JsonObject } from "./protocol.js";
import type {
CodexAppServerBindingIdentity,
CodexAppServerContextEngineBinding,
CodexAppServerThreadBinding,
} from "./session-binding.js";
import { fingerprintCodexThreadConfig } from "./thread-fingerprints.js";
import { CodexThreadBindingConflictError } from "./thread-lifecycle-errors.js";
import type { CodexThreadLifecycleTimingTracker } from "./thread-lifecycle-timing.js";
import type {
CodexAppServerThreadLifecycleBinding,
CodexStartOrResumeThreadParams,
} from "./thread-lifecycle-types.js";
import type { resolveCodexAppServerThreadModelSelection } from "./thread-model-selection.js";
import { buildThreadResumeParams } from "./thread-requests.js";
type CodexWarmThreadFinalConfigPatch = {
configPatch?: JsonObject;
nativeHookRelayGeneration?: string;
};
type CodexWarmThreadReuseParams = {
params: CodexStartOrResumeThreadParams;
binding: CodexAppServerThreadBinding;
bindingIdentity: CodexAppServerBindingIdentity;
clientId?: string;
contextEngineBinding?: CodexAppServerContextEngineBinding;
dynamicToolsFingerprint: string;
hostSystemAgentActive: boolean;
lifecycleTiming: CodexThreadLifecycleTimingTracker;
releaseConsumedThread: (threadId: string, cause?: unknown) => Promise<void>;
ringZeroActive: boolean;
ringZeroInheritedMcpServerNames: string[];
startModelProvider?: string;
startModelSelection: ReturnType<typeof resolveCodexAppServerThreadModelSelection>;
throwIfAborted: () => void;
userMcpServersConfigPatch?: JsonObject;
};
type CodexWarmThreadReuseResult = {
binding?: CodexAppServerThreadLifecycleBinding;
prebuiltFinalConfigPatch?: CodexWarmThreadFinalConfigPatch;
};
type CodexLiveThreadReleaseParams = {
client: CodexAppServerClient;
abandonClient?: () => Promise<void>;
lifecycleTiming: CodexThreadLifecycleTimingTracker;
threadId: string;
cause?: unknown;
};
/** Preserves the caller's abort reason across thread ownership transitions. */
export function throwIfCodexThreadLifecycleAborted(signal?: AbortSignal): void {
if (!signal?.aborted) {
return;
}
const reason = signal.reason;
if (reason instanceof Error) {
throw reason;
}
const error = new Error(
typeof reason === "string" && reason.length > 0
? reason
: "codex app-server thread lifecycle aborted",
);
error.name = "AbortError";
throw error;
}
/** Releases consumed subscription ownership or retires an unsafe client. */
export async function releaseCodexConsumedLiveThread(
options: CodexLiveThreadReleaseParams,
): Promise<void> {
const released = await options.lifecycleTiming.measure("retained-thread-unsubscribe", () =>
unsubscribeCodexThreadBestEffort(options.client, {
threadId: options.threadId,
timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS,
}),
);
if (released) {
return;
}
await (options.abandonClient ?? (() => closeCodexStartupClientBestEffort(options.client)))();
throw new CodexAppServerUnsafeSubscriptionError(
`Codex retained thread subscription could not be released: ${options.threadId}`,
options.cause !== undefined ? { cause: options.cause } : undefined,
);
}
/** Transfers retained-slot ownership before unconditionally releasing it. */
export async function releaseCodexRetainedLiveThread(
options: CodexLiveThreadReleaseParams,
): Promise<void> {
if (await consumeCodexAppServerLiveThread(options.client, options.threadId)) {
// An abort must not leave a consumed subscription orphaned on the client.
await releaseCodexConsumedLiveThread(options);
}
}
/** Reuses one safely owned, fully matching subscription on its original client. */
export async function tryReuseCodexLiveThread(
options: CodexWarmThreadReuseParams,
): Promise<CodexWarmThreadReuseResult> {
const {
params,
binding,
bindingIdentity,
clientId,
contextEngineBinding,
dynamicToolsFingerprint,
hostSystemAgentActive,
lifecycleTiming,
releaseConsumedThread,
ringZeroActive,
ringZeroInheritedMcpServerNames,
startModelProvider,
startModelSelection,
throwIfAborted,
userMcpServersConfigPatch,
} = options;
if (
!binding.clientId ||
binding.clientId !== clientId ||
binding.preserveNativeModel === true ||
binding.connectionScope === "supervision" ||
ringZeroActive ||
contextEngineBinding
) {
return {};
}
const prebuiltFinalConfigPatch = params.buildFinalConfigPatch?.({
action: "resume",
binding,
}) ?? {
configPatch: params.finalConfigPatch,
nativeHookRelayGeneration: params.nativeHookRelayGeneration,
};
const pluginAppsConfigPatch =
params.pluginThreadConfig?.enabled && binding.pluginAppPolicyContext
? buildCodexPluginAppsConfigPatchFromPolicyContext(binding.pluginAppPolicyContext)
: undefined;
const resumeAuthProfileId = params.params.authProfileId ?? binding.authProfileId;
const resumeConfig = mergeCodexThreadConfigs(
params.config,
userMcpServersConfigPatch,
pluginAppsConfigPatch,
prebuiltFinalConfigPatch.configPatch,
);
const resumeParams = lifecycleTiming.measureSync("warm-thread-resume-params", () =>
buildThreadResumeParams(params.params, {
threadId: binding.threadId,
authProfileId: resumeAuthProfileId,
model: startModelSelection.model,
modelProvider: startModelProvider,
preserveNativeModel: false,
appServer: params.appServer,
dynamicTools: params.dynamicTools,
developerInstructions: params.developerInstructions,
config: resumeConfig,
nativeCodeModeEnabled: params.nativeCodeModeEnabled,
nativeProviderWebSearchSupport: params.nativeProviderWebSearchSupport,
nativeCodeModeOnlyEnabled: params.nativeCodeModeOnlyEnabled,
webSearchAllowed: params.webSearchAllowed,
hostSystemAgentActive,
ringZeroInheritedMcpServerNames,
}),
);
const liveThreadConfigFingerprint = fingerprintCodexThreadConfig(
{
...resumeParams,
// Keep the actual loaded provider separate from caller-selected
// overrides so account or provider changes always invalidate reuse.
model: binding.model ?? resumeParams.model ?? null,
requestedModel: resumeParams.model ?? null,
modelProvider: binding.modelProvider ?? resumeParams.modelProvider ?? null,
requestedModelProvider: resumeParams.modelProvider ?? binding.modelProvider ?? null,
},
resumeAuthProfileId,
dynamicToolsFingerprint,
);
if (
!(await consumeCodexAppServerLiveThread(
params.client,
binding.threadId,
liveThreadConfigFingerprint,
))
) {
return { prebuiltFinalConfigPatch };
}
try {
const nativeHookRelayGeneration =
prebuiltFinalConfigPatch.nativeHookRelayGeneration ?? binding.nativeHookRelayGeneration;
// Validate ownership even when relay generation is unchanged; reset may
// have replaced the persisted binding since it was first read.
const committed = await lifecycleTiming.measure("warm-thread-write-binding", () =>
params.bindingStore.mutate(bindingIdentity, {
kind: "patch",
threadId: binding.threadId,
patch: { nativeHookRelayGeneration },
}),
);
if (!committed) {
throw new CodexThreadBindingConflictError(binding.threadId, "committing a reused thread");
}
throwIfAborted();
lifecycleTiming.mark("thread-ready");
lifecycleTiming.logSummary({
runId: params.params.runId,
sessionId: params.params.sessionId,
sessionKey: params.params.sessionKey,
threadId: binding.threadId,
action: "resumed",
});
return {
binding: {
...binding,
nativeHookRelayGeneration,
liveThreadConfigFingerprint,
lifecycle: { action: "resumed" },
},
prebuiltFinalConfigPatch,
};
} catch (error) {
await releaseConsumedThread(binding.threadId, error);
throw error;
}
}

462
extensions/codex/src/app-server/thread-lifecycle.binding.test.ts
View File

@@ -1,6 +1,10 @@
// Codex tests cover thread lifecycle.binding plugin behavior.
import path from "node:path";
import { describe, expect, it, vi } from "vitest";
import {
ensureCodexAppServerClientRuntime,
retainCodexAppServerLiveThread,
} from "./client-runtime.js";
import { CodexAppServerRpcError } from "./client.js";
import type { CodexDynamicToolFunctionSpec } from "./protocol.js";
import {
@@ -323,6 +327,461 @@ describe("Codex app-server thread lifecycle bindings", () => {
});
});
it("reuses only an explicitly retained subscription on the original client", async () => {
const sessionFile = path.join(tempDir, "warm-session.jsonl");
const workspaceDir = path.join(tempDir, "warm-workspace");
const params = createParams(sessionFile, workspaceDir);
const request = vi.fn(async (method: string) => {
if (method === "thread/start") {
return threadStartResult("thread-warm");
}
throw new Error(`unexpected method: ${method}`);
});
const client = {
getInstanceId: () => "client-warm",
request,
addNotificationHandler: () => () => undefined,
addRequestHandler: () => () => undefined,
} as never;
ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir });
const buildFinalConfigPatch = vi
.fn()
.mockReturnValueOnce({ nativeHookRelayGeneration: "generation-warm" })
.mockReturnValueOnce({ nativeHookRelayGeneration: "generation-warm-next" });
const common = {
client,
params,
cwd: workspaceDir,
dynamicTools: [],
appServer: createThreadLifecycleAppServerOptions(),
userMcpServersEnabled: false,
buildFinalConfigPatch,
};
const started = await startOrResumeThread(common);
await expect(
retainCodexAppServerLiveThread(
client,
started.threadId,
undefined,
started.liveThreadConfigFingerprint,
),
).resolves.toEqual({});
const reused = await startOrResumeThread(common);
expect(started).toMatchObject({
clientId: "client-warm",
threadId: "thread-warm",
nativeHookRelayGeneration: "generation-warm",
lifecycle: { action: "started" },
});
expect(reused).toMatchObject({
clientId: "client-warm",
threadId: "thread-warm",
nativeHookRelayGeneration: "generation-warm-next",
lifecycle: { action: "resumed" },
});
await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({
nativeHookRelayGeneration: "generation-warm-next",
});
expect(request.mock.calls.map(([method]) => method)).toEqual(["thread/start"]);
expect(buildFinalConfigPatch).toHaveBeenNthCalledWith(1, { action: "start" });
expect(buildFinalConfigPatch).toHaveBeenNthCalledWith(2, {
action: "resume",
binding: expect.objectContaining({ threadId: "thread-warm" }),
});
});
it("releases a retained subscription when its unchanged binding loses ownership", async () => {
const sessionFile = path.join(tempDir, "warm-conflict-session.jsonl");
const workspaceDir = path.join(tempDir, "warm-conflict-workspace");
const params = createParams(sessionFile, workspaceDir);
const request = vi.fn(async (method: string) => {
if (method === "thread/start") {
return threadStartResult("thread-warm-conflict");
}
if (method === "thread/unsubscribe") {
return { status: "unsubscribed" };
}
throw new Error(`unexpected method: ${method}`);
});
const client = {
getInstanceId: () => "client-warm-conflict",
request,
addNotificationHandler: () => () => undefined,
addRequestHandler: () => () => undefined,
} as never;
ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir });
const common = {
client,
params,
cwd: workspaceDir,
dynamicTools: [],
appServer: createThreadLifecycleAppServerOptions(),
userMcpServersEnabled: false,
};
const started = await startOrResumeThread(common);
await retainCodexAppServerLiveThread(
client,
started.threadId,
undefined,
started.liveThreadConfigFingerprint,
);
const conflictBindingStore = {
...testCodexAppServerBindingStore,
mutate: vi.fn(async (...args: Parameters<typeof testCodexAppServerBindingStore.mutate>) => {
if (args[1].kind === "patch") {
return false;
}
return await testCodexAppServerBindingStore.mutate(...args);
}),
};
await expect(
startOrResumeThreadImpl({ ...common, bindingStore: conflictBindingStore }),
).rejects.toMatchObject({ name: "CodexThreadBindingConflictError" });
expect(conflictBindingStore.mutate).toHaveBeenCalledWith(
expect.anything(),
expect.objectContaining({ kind: "patch", threadId: "thread-warm-conflict" }),
);
expect(request.mock.calls.map(([method]) => method)).toEqual([
"thread/start",
"thread/unsubscribe",
]);
});
it("releases a retained subscription before changing context-engine mode", async () => {
const sessionFile = path.join(tempDir, "warm-context-session.jsonl");
const workspaceDir = path.join(tempDir, "warm-context-workspace");
const params = createParams(sessionFile, workspaceDir);
let startCount = 0;
const request = vi.fn(async (method: string) => {
if (method === "thread/start") {
startCount += 1;
return threadStartResult(`thread-warm-context-${startCount}`);
}
if (method === "thread/unsubscribe") {
return { status: "unsubscribed" };
}
throw new Error(`unexpected method: ${method}`);
});
const client = {
getInstanceId: () => "client-warm-context",
request,
addNotificationHandler: () => () => undefined,
addRequestHandler: () => () => undefined,
} as never;
ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir });
const common = {
client,
params,
cwd: workspaceDir,
dynamicTools: [],
appServer: createThreadLifecycleAppServerOptions(),
userMcpServersEnabled: false,
};
const started = await startOrResumeThread(common);
await expect(
retainCodexAppServerLiveThread(
client,
started.threadId,
undefined,
started.liveThreadConfigFingerprint,
),
).resolves.toEqual({});
params.contextEngine = {
info: { id: "lossless-claw", name: "Lossless Claw", ownsCompaction: true },
assemble: vi.fn(),
compact: vi.fn(),
} as never;
params.contextTokenBudget = 400_000;
const rotated = await startOrResumeThread(common);
expect(request.mock.calls.map(([method]) => method)).toEqual([
"thread/start",
"thread/unsubscribe",
"thread/start",
]);
expect(rotated).toMatchObject({
threadId: "thread-warm-context-2",
contextEngine: { engineId: "lossless-claw" },
lifecycle: { action: "started", rotatedContextEngineBinding: true },
});
});
it("releases and resumes a retained thread when its effective config changes", async () => {
const sessionFile = path.join(tempDir, "warm-config-session.jsonl");
const workspaceDir = path.join(tempDir, "warm-config-workspace");
const params = createParams(sessionFile, workspaceDir);
const request = vi.fn(async (method: string) => {
if (method === "thread/start" || method === "thread/resume") {
return threadStartResult("thread-warm-config");
}
if (method === "thread/unsubscribe") {
return { status: "unsubscribed" };
}
throw new Error(`unexpected method: ${method}`);
});
const client = {
getInstanceId: () => "client-warm-config",
request,
addNotificationHandler: () => () => undefined,
addRequestHandler: () => () => undefined,
} as never;
ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir });
const common = {
client,
params,
cwd: workspaceDir,
dynamicTools: [],
appServer: createThreadLifecycleAppServerOptions(),
userMcpServersEnabled: false,
};
const started = await startOrResumeThread({
...common,
config: { test_setting: "before" },
});
await retainCodexAppServerLiveThread(
client,
started.threadId,
undefined,
started.liveThreadConfigFingerprint,
);
const resumed = await startOrResumeThread({
...common,
config: { test_setting: "after" },
});
expect(request.mock.calls.map(([method]) => method)).toEqual([
"thread/start",
"thread/unsubscribe",
"thread/resume",
]);
expect(resumed).toMatchObject({
threadId: "thread-warm-config",
lifecycle: { action: "resumed" },
});
expect(resumed.liveThreadConfigFingerprint).not.toBe(started.liveThreadConfigFingerprint);
});
it("releases and resumes a retained thread when its auth profile changes", async () => {
const sessionFile = path.join(tempDir, "warm-auth-session.jsonl");
const workspaceDir = path.join(tempDir, "warm-auth-workspace");
const params = createParams(sessionFile, workspaceDir);
params.authProfileId = "openai:before";
const request = vi.fn(async (method: string) => {
if (method === "thread/start" || method === "thread/resume") {
return threadStartResult("thread-warm-auth");
}
if (method === "thread/unsubscribe") {
return { status: "unsubscribed" };
}
throw new Error(`unexpected method: ${method}`);
});
const client = {
getInstanceId: () => "client-warm-auth",
request,
addNotificationHandler: () => () => undefined,
addRequestHandler: () => () => undefined,
} as never;
ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir });
const common = {
client,
params,
cwd: workspaceDir,
dynamicTools: [],
appServer: createThreadLifecycleAppServerOptions(),
userMcpServersEnabled: false,
};
const started = await startOrResumeThread(common);
await retainCodexAppServerLiveThread(
client,
started.threadId,
undefined,
started.liveThreadConfigFingerprint,
);
params.authProfileId = "openai:after";
const resumed = await startOrResumeThread(common);
expect(request.mock.calls.map(([method]) => method)).toEqual([
"thread/start",
"thread/unsubscribe",
"thread/resume",
]);
expect(resumed).toMatchObject({
authProfileId: "openai:after",
threadId: "thread-warm-auth",
lifecycle: { action: "resumed" },
});
expect(resumed.liveThreadConfigFingerprint).not.toBe(started.liveThreadConfigFingerprint);
});
it("releases and resumes a retained thread when its model provider changes", async () => {
const sessionFile = path.join(tempDir, "warm-provider-session.jsonl");
const workspaceDir = path.join(tempDir, "warm-provider-workspace");
const params = createParams(sessionFile, workspaceDir);
const request = vi.fn(async (method: string) => {
if (method === "thread/start" || method === "thread/resume") {
return threadStartResult("thread-warm-provider");
}
if (method === "thread/unsubscribe") {
return { status: "unsubscribed" };
}
throw new Error(`unexpected method: ${method}`);
});
const client = {
getInstanceId: () => "client-warm-provider",
request,
addNotificationHandler: () => () => undefined,
addRequestHandler: () => () => undefined,
} as never;
ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir });
const common = {
client,
params,
cwd: workspaceDir,
dynamicTools: [],
appServer: createThreadLifecycleAppServerOptions(),
userMcpServersEnabled: false,
};
const started = await startOrResumeThread(common);
await retainCodexAppServerLiveThread(
client,
started.threadId,
undefined,
started.liveThreadConfigFingerprint,
);
params.provider = "custom-provider";
const resumed = await startOrResumeThread(common);
expect(request.mock.calls.map(([method]) => method)).toEqual([
"thread/start",
"thread/unsubscribe",
"thread/resume",
]);
expect(request).toHaveBeenCalledWith(
"thread/resume",
expect.objectContaining({ modelProvider: "custom-provider" }),
expect.anything(),
);
expect(resumed.liveThreadConfigFingerprint).not.toBe(started.liveThreadConfigFingerprint);
});
it("releases and resumes a retained thread when its approval policy changes", async () => {
const sessionFile = path.join(tempDir, "warm-policy-session.jsonl");
const workspaceDir = path.join(tempDir, "warm-policy-workspace");
const params = createParams(sessionFile, workspaceDir);
const request = vi.fn(async (method: string) => {
if (method === "thread/start" || method === "thread/resume") {
return threadStartResult("thread-warm-policy");
}
if (method === "thread/unsubscribe") {
return { status: "unsubscribed" };
}
throw new Error(`unexpected method: ${method}`);
});
const client = {
getInstanceId: () => "client-warm-policy",
request,
addNotificationHandler: () => () => undefined,
addRequestHandler: () => () => undefined,
} as never;
ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir });
const appServer = createThreadLifecycleAppServerOptions();
const common = {
client,
params,
cwd: workspaceDir,
dynamicTools: [],
appServer,
userMcpServersEnabled: false,
};
const started = await startOrResumeThread(common);
await retainCodexAppServerLiveThread(
client,
started.threadId,
undefined,
started.liveThreadConfigFingerprint,
);
appServer.approvalPolicy = "on-request";
const resumed = await startOrResumeThread(common);
expect(request.mock.calls.map(([method]) => method)).toEqual([
"thread/start",
"thread/unsubscribe",
"thread/resume",
]);
expect(request).toHaveBeenCalledWith(
"thread/resume",
expect.objectContaining({ approvalPolicy: "on-request" }),
expect.anything(),
);
expect(resumed.liveThreadConfigFingerprint).not.toBe(started.liveThreadConfigFingerprint);
});
it("fails closed when a retained mode-transition subscription cannot be released", async () => {
const sessionFile = path.join(tempDir, "unsafe-warm-session.jsonl");
const workspaceDir = path.join(tempDir, "unsafe-warm-workspace");
const params = createParams(sessionFile, workspaceDir);
const request = vi.fn(async (method: string) => {
if (method === "thread/start") {
return threadStartResult("thread-unsafe-warm");
}
if (method === "thread/unsubscribe") {
throw new Error("unsubscribe unavailable");
}
throw new Error(`unexpected method: ${method}`);
});
const client = {
getInstanceId: () => "client-unsafe-warm",
request,
addNotificationHandler: () => () => undefined,
addRequestHandler: () => () => undefined,
} as never;
ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir });
const abandonClient = vi.fn(async () => undefined);
const common = {
client,
abandonClient,
params,
cwd: workspaceDir,
dynamicTools: [],
appServer: createThreadLifecycleAppServerOptions(),
userMcpServersEnabled: false,
};
const started = await startOrResumeThread(common);
await expect(
retainCodexAppServerLiveThread(
client,
started.threadId,
undefined,
started.liveThreadConfigFingerprint,
),
).resolves.toEqual({});
params.contextEngine = {
info: { id: "lossless-claw", name: "Lossless Claw", ownsCompaction: true },
assemble: vi.fn(),
compact: vi.fn(),
} as never;
await expect(startOrResumeThread(common)).rejects.toMatchObject({
name: "CodexAppServerUnsafeSubscriptionError",
message: "Codex retained thread subscription could not be released: thread-unsafe-warm",
});
expect(abandonClient).toHaveBeenCalledTimes(1);
expect(request.mock.calls.map(([method]) => method)).toEqual([
"thread/start",
"thread/unsubscribe",
]);
});
it("reuses one live ephemeral thread across two incognito turns", async () => {
const sessionFile = path.join(tempDir, "incognito-session.jsonl");
const workspaceDir = path.join(tempDir, "incognito-workspace");
@@ -1238,6 +1697,7 @@ describe("Codex app-server thread lifecycle bindings", () => {
});
expect(restrictedBinding.threadId).toBe("thread-2");
expect(restrictedBinding).not.toHaveProperty("liveThreadConfigFingerprint");
expect(savedAfterRestriction?.threadId).toBe("thread-1");
expect(resumedBinding.threadId).toBe("thread-1");
expect(request.mock.calls.map(([method]) => method)).toEqual([
@@ -1296,6 +1756,7 @@ describe("Codex app-server thread lifecycle bindings", () => {
});
expect(restrictedBinding.threadId).toBe("thread-2");
expect(restrictedBinding).not.toHaveProperty("liveThreadConfigFingerprint");
expect(savedAfterRestriction?.threadId).toBe("thread-1");
expect(resumedBinding.threadId).toBe("thread-1");
expect(request.mock.calls.map(([method]) => method)).toEqual([
@@ -1358,6 +1819,7 @@ describe("Codex app-server thread lifecycle bindings", () => {
});
expect(transientBinding.threadId).toBe("thread-2");
expect(transientBinding).not.toHaveProperty("liveThreadConfigFingerprint");
expect(savedAfterUnknownSupport?.threadId).toBe("thread-1");
expect(resumedBinding.threadId).toBe("thread-1");
expect(request.mock.calls.map(([method]) => method)).toEqual([