diff --git a/extensions/codex/media-understanding-provider.test.ts b/extensions/codex/media-understanding-provider.test.ts index 71b2a4fb4b94..50332348f1f9 100644 --- a/extensions/codex/media-understanding-provider.test.ts +++ b/extensions/codex/media-understanding-provider.test.ts @@ -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(); diff --git a/extensions/codex/src/app-server/client-runtime.test.ts b/extensions/codex/src/app-server/client-runtime.test.ts index 3ec4fe89f77d..9423355ae54d 100644 --- a/extensions/codex/src/app-server/client-runtime.test.ts +++ b/extensions/codex/src/app-server/client-runtime.test.ts @@ -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((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); + }); }); diff --git a/extensions/codex/src/app-server/client-runtime.ts b/extensions/codex/src/app-server/client-runtime.ts index 174dda9c0a63..5de3c02fc1ec 100644 --- a/extensions/codex/src/app-server/client-runtime.ts +++ b/extensions/codex/src/app-server/client-runtime.ts @@ -12,6 +12,9 @@ type ClientRuntimeContext = Omit & type ClientRuntime = { context: ClientRuntimeContext; + retainedThreadId?: string; + retainedThreadConfigFingerprint?: string; + retainedThreadRelease?: Promise; }; const configuredClients = new WeakMap(); @@ -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, + 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 { + 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; +} diff --git a/extensions/codex/src/app-server/run-attempt-cleanup.ts b/extensions/codex/src/app-server/run-attempt-cleanup.ts index 979b7055a6b6..f50a6cbc1ba7 100644 --- a/extensions/codex/src/app-server/run-attempt-cleanup.ts +++ b/extensions/codex/src/app-server/run-attempt-cleanup.ts @@ -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, { diff --git a/extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts b/extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts index dff5e394ceee..47074a312204 100644 --- a/extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts +++ b/extensions/codex/src/app-server/run-attempt-thread-cleanup.test.ts @@ -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 () => { diff --git a/extensions/codex/src/app-server/run-attempt.context-engine.test.ts b/extensions/codex/src/app-server/run-attempt.context-engine.test.ts index ccabfc217d49..167a8d0a1f6b 100644 --- a/extensions/codex/src/app-server/run-attempt.context-engine.test.ts +++ b/extensions/codex/src/app-server/run-attempt.context-engine.test.ts @@ -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 { 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); diff --git a/extensions/codex/src/app-server/run-attempt.test.ts b/extensions/codex/src/app-server/run-attempt.test.ts index 361e1729bba4..db3b515ddc46 100644 --- a/extensions/codex/src/app-server/run-attempt.test.ts +++ b/extensions/codex/src/app-server/run-attempt.test.ts @@ -220,11 +220,11 @@ async function writeExistingBinding( }); } -function attachSqliteSessionTarget( +async function attachSqliteSessionTarget( params: EmbeddedRunAttemptParams, storePath: string, sessionId: string, -): void { +): Promise { 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: { diff --git a/extensions/codex/src/app-server/thread-fingerprints.test.ts b/extensions/codex/src/app-server/thread-fingerprints.test.ts index 40615ca28866..52c1ba058225 100644 --- a/extensions/codex/src/app-server/thread-fingerprints.test.ts +++ b/extensions/codex/src/app-server/thread-fingerprints.test.ts @@ -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", () => { diff --git a/extensions/codex/src/app-server/thread-fingerprints.ts b/extensions/codex/src/app-server/thread-fingerprints.ts index 41c3c6e4b3f9..5da5d104f473 100644 --- a/extensions/codex/src/app-server/thread-fingerprints.ts +++ b/extensions/codex/src/app-server/thread-fingerprints.ts @@ -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 ? "" : request.serviceTier, + baseInstructions: request.baseInstructions ?? null, + developerInstructions: request.developerInstructions ?? null, + config: request.config ?? {}, + }), + ); +} + export function fingerprintEnvironmentSelection( environments: CodexTurnEnvironmentParams[] | undefined, ): string | undefined { diff --git a/extensions/codex/src/app-server/thread-lifecycle-io.ts b/extensions/codex/src/app-server/thread-lifecycle-io.ts index 4c8451a7e1a8..5e8293f41ab8 100644 --- a/extensions/codex/src/app-server/thread-lifecycle-io.ts +++ b/extensions/codex/src/app-server/thread-lifecycle-io.ts @@ -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; + 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 } : {}), diff --git a/extensions/codex/src/app-server/thread-lifecycle-run.ts b/extensions/codex/src/app-server/thread-lifecycle-run.ts index b5d76316494e..2a4a94b579cf 100644 --- a/extensions/codex/src/app-server/thread-lifecycle-run.ts +++ b/extensions/codex/src/app-server/thread-lifecycle-run.ts @@ -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, diff --git a/extensions/codex/src/app-server/thread-lifecycle-types.ts b/extensions/codex/src/app-server/thread-lifecycle-types.ts index e5dd95030908..f49efb31767c 100644 --- a/extensions/codex/src/app-server/thread-lifecycle-types.ts +++ b/extensions/codex/src/app-server/thread-lifecycle-types.ts @@ -16,6 +16,7 @@ type CodexAppServerThreadLifecycle = { export type CodexAppServerThreadLifecycleBinding = CodexAppServerThreadBinding & { lifecycle: CodexAppServerThreadLifecycle; + liveThreadConfigFingerprint?: string; }; type CodexThreadFinalConfigPatchDecision = diff --git a/extensions/codex/src/app-server/thread-lifecycle-warm.ts b/extensions/codex/src/app-server/thread-lifecycle-warm.ts new file mode 100644 index 000000000000..1a6bacf25361 --- /dev/null +++ b/extensions/codex/src/app-server/thread-lifecycle-warm.ts @@ -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; + ringZeroActive: boolean; + ringZeroInheritedMcpServerNames: string[]; + startModelProvider?: string; + startModelSelection: ReturnType; + throwIfAborted: () => void; + userMcpServersConfigPatch?: JsonObject; +}; + +type CodexWarmThreadReuseResult = { + binding?: CodexAppServerThreadLifecycleBinding; + prebuiltFinalConfigPatch?: CodexWarmThreadFinalConfigPatch; +}; + +type CodexLiveThreadReleaseParams = { + client: CodexAppServerClient; + abandonClient?: () => Promise; + 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 { + 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 { + 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 { + 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; + } +} diff --git a/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts b/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts index c22eb4ed017c..6f8c428fc321 100644 --- a/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts +++ b/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts @@ -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) => { + 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([