diff --git a/docs/plugins/codex-harness.md b/docs/plugins/codex-harness.md index 93f57f59f54b..345816d2d545 100644 --- a/docs/plugins/codex-harness.md +++ b/docs/plugins/codex-harness.md @@ -1202,8 +1202,17 @@ for Gateway pressure, and inspect host or container memory for the Codex child. The bundled Codex has no heap or RSS limit and no configurable idle-unload delay. After the last client unsubscribes, an inactive thread can remain loaded -for up to 30 minutes. On constrained hosts, reduce native Codex subagent fan-out -before increasing the Gateway heap: +for up to 30 minutes. OpenClaw independently keeps up to 64 idle conversation +threads subscribed on each Codex app-server for 30 minutes after their last +activity. This preserves warm sessions and session-scoped approvals when several +conversations alternate. Active turns and parents with unfinished native +subagents are protected from idle eviction; session reset or deletion releases +its own thread immediately. Idle-limit eviction unsubscribes the least recently +used conversation, after which Codex applies its separate unloading delay and a +later resumed session can require approvals again. + +On constrained hosts, reduce native Codex subagent fan-out before increasing the +Gateway heap: ```json5 { diff --git a/extensions/codex/harness.ts b/extensions/codex/harness.ts index 8353e8229d9e..6bd87c71cb6f 100644 --- a/extensions/codex/harness.ts +++ b/extensions/codex/harness.ts @@ -243,18 +243,25 @@ export function createCodexAppServerAgentHarness(options: { }, reset: async (params) => { if (params.sessionId) { - const { reclaimCurrentCodexSessionGeneration, sessionBindingIdentity } = - await import("./src/app-server/session-binding.js"); + const [ + { reclaimCurrentCodexSessionGeneration, sessionBindingIdentity }, + { retireCodexAppServerSessionGeneration }, + ] = await Promise.all([ + import("./src/app-server/session-binding.js"), + import("./src/app-server/session-retirement.js"), + ]); const identity = sessionBindingIdentity({ agentId: params.agentId, sessionId: params.sessionId, sessionKey: params.sessionKey, }); - const resetGeneration = - params.reason === "deleted" - ? options.bindingStore.retireSessionGeneration.bind(options.bindingStore) - : options.bindingStore.resetSessionGeneration.bind(options.bindingStore); - let reset = await resetGeneration(identity); + const resetGeneration = () => + retireCodexAppServerSessionGeneration({ + bindingStore: options.bindingStore, + identity, + mode: params.reason === "deleted" ? "retire" : "reset", + }); + let reset = await resetGeneration(); if (reset === "conflict") { const reclaimed = await reclaimCurrentCodexSessionGeneration({ bindingStore: options.bindingStore, @@ -262,7 +269,7 @@ export function createCodexAppServerAgentHarness(options: { config: options.resolveConfig?.(), }); if (reclaimed) { - reset = await resetGeneration(identity); + reset = await resetGeneration(); } } if (reset === "conflict") { diff --git a/extensions/codex/index.ts b/extensions/codex/index.ts index a02e915f02f8..92a617590dc4 100644 --- a/extensions/codex/index.ts +++ b/extensions/codex/index.ts @@ -329,15 +329,21 @@ export default definePluginEntry({ return; } const config = resolveCurrentConfig(); - const { sessionBindingIdentity } = await import("./src/app-server/session-binding.js"); - await bindingStore.retireSessionGeneration( - sessionBindingIdentity({ + const [{ sessionBindingIdentity }, { retireCodexAppServerSessionGeneration }] = + await Promise.all([ + import("./src/app-server/session-binding.js"), + import("./src/app-server/session-retirement.js"), + ]); + await retireCodexAppServerSessionGeneration({ + bindingStore, + identity: sessionBindingIdentity({ sessionId: event.sessionId, ...(sessionKey ? { sessionKey } : {}), ...(ctx.agentId ? { agentId: ctx.agentId } : {}), ...(config ? { config } : {}), }), - ); + mode: "retire", + }); }); }, }); diff --git a/extensions/codex/src/app-server/client-runtime.test.ts b/extensions/codex/src/app-server/client-runtime.test.ts index 9423355ae54d..db212d0bd2c9 100644 --- a/extensions/codex/src/app-server/client-runtime.test.ts +++ b/extensions/codex/src/app-server/client-runtime.test.ts @@ -2,6 +2,9 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import type { CodexAppServerClient } from "./client.js"; import { createClientHarness } from "./test-support.js"; +const EXPECTED_LIVE_THREAD_IDLE_TIMEOUT_MS = 30 * 60_000; +const EXPECTED_MAX_IDLE_LIVE_THREADS = 64; + const mocks = vi.hoisted(() => ({ refreshAuth: vi.fn(async () => ({ accessToken: "refreshed", chatgptAccountId: "account" })), mergeRateLimitUpdate: vi.fn(), @@ -18,6 +21,8 @@ vi.mock("./rate-limit-cache.js", () => ({ const { consumeCodexAppServerLiveThread, ensureCodexAppServerClientRuntime, + protectCodexAppServerLiveThread, + releaseCodexAppServerLiveThread, retainCodexAppServerLiveThread, } = await import("./client-runtime.js"); @@ -29,6 +34,7 @@ describe("Codex app-server client runtime", () => { client.close(); } clients.length = 0; + vi.useRealTimers(); mocks.refreshAuth.mockClear(); mocks.mergeRateLimitUpdate.mockClear(); }); @@ -55,7 +61,7 @@ describe("Codex app-server client runtime", () => { expect(addNotificationHandler).toHaveBeenCalledTimes(1); expect(addRequestHandler).toHaveBeenCalledTimes(1); - expect(addCloseHandler).not.toHaveBeenCalled(); + expect(addCloseHandler).toHaveBeenCalledTimes(1); harness.send({ method: "account/rateLimits/updated", params: { rateLimits: { primary: { usedPercent: 12 } } }, @@ -104,76 +110,205 @@ describe("Codex app-server client runtime", () => { }); }); - it("retains and consumes only one subscribed thread per physical client", async () => { + it("retains independently subscribed conversations on the same physical client", async () => { const harness = createClientHarness(); clients.push(harness.client); await expect( retainCodexAppServerLiveThread(harness.client, "thread-before-runtime"), - ).resolves.toBeUndefined(); + ).resolves.toBe(false); 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); + await expect(retainCodexAppServerLiveThread(harness.client, "thread-a")).resolves.toBe(true); + await expect(retainCodexAppServerLiveThread(harness.client, "thread-b")).resolves.toBe(true); + await expect(consumeCodexAppServerLiveThread(harness.client, "thread-a")).resolves.toEqual( + expect.objectContaining({ release: expect.any(Function) }), + ); + await expect(consumeCodexAppServerLiveThread(harness.client, "thread-b")).resolves.toEqual( + expect.objectContaining({ release: expect.any(Function) }), + ); + await expect( + consumeCodexAppServerLiveThread(harness.client, "thread-a"), + ).resolves.toBeUndefined(); }); - it("waits for the old subscription to release before another thread can acquire it", async () => { + it("blocks only the exact thread whose subscription is being released", 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) => { + const pendingRelease = new Promise((resolve) => { finishRelease = resolve; }); - const transition = retainCodexAppServerLiveThread( - harness.client, - "thread-2", - async () => previousRelease, - ); - const oldThreadAcquisition = consumeCodexAppServerLiveThread(harness.client, "thread-1"); + await retainCodexAppServerLiveThread(harness.client, "thread-a", async () => pendingRelease); + await retainCodexAppServerLiveThread(harness.client, "thread-b"); + const release = releaseCodexAppServerLiveThread(harness.client, "thread-a"); + const sameThreadAcquisition = consumeCodexAppServerLiveThread(harness.client, "thread-a"); + await expect(consumeCodexAppServerLiveThread(harness.client, "thread-b")).resolves.toEqual( + expect.objectContaining({ release: expect.any(Function) }), + ); 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); + await expect(release).resolves.toBe(true); + await expect(sameThreadAcquisition).resolves.toBeUndefined(); }); - it("does not expose either thread after a previous subscription release fails", async () => { + it("does not re-expose a failed release or discard an unrelated conversation", async () => { const harness = createClientHarness(); clients.push(harness.client); ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" }); - await retainCodexAppServerLiveThread(harness.client, "thread-1"); + await retainCodexAppServerLiveThread(harness.client, "thread-a", async () => { + throw new Error("unsubscribe unavailable"); + }); + await retainCodexAppServerLiveThread(harness.client, "thread-b"); + await expect(releaseCodexAppServerLiveThread(harness.client, "thread-a")).rejects.toThrow( + "unsubscribe unavailable", + ); 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); + consumeCodexAppServerLiveThread(harness.client, "thread-a"), + ).resolves.toBeUndefined(); + await expect(consumeCodexAppServerLiveThread(harness.client, "thread-b")).resolves.toEqual( + expect.objectContaining({ release: expect.any(Function) }), + ); }); - it("reuses a retained subscription only for its complete configuration fingerprint", async () => { + it("transfers ownership only for the exact immutable thread 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({}); + ).resolves.toBe(true); await expect( consumeCodexAppServerLiveThread(harness.client, "thread-1", "config-after"), - ).resolves.toBe(false); + ).resolves.toBeUndefined(); await expect( consumeCodexAppServerLiveThread(harness.client, "thread-1", "config-before"), - ).resolves.toBe(true); + ).resolves.toEqual( + expect.objectContaining({ + configFingerprint: "config-before", + release: expect.any(Function), + }), + ); + }); + + it("evicts only the oldest idle subscription at the per-client capacity", async () => { + const harness = createClientHarness(); + clients.push(harness.client); + ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" }); + const release = vi.fn(async (_threadId: string) => undefined); + + for (let index = 0; index <= EXPECTED_MAX_IDLE_LIVE_THREADS; index += 1) { + await retainCodexAppServerLiveThread(harness.client, `thread-${index}`, release); + } + + expect(release).toHaveBeenCalledExactlyOnceWith("thread-0"); + await expect(consumeCodexAppServerLiveThread(harness.client, "thread-1")).resolves.toEqual( + expect.objectContaining({ release }), + ); + }); + + it("expires an idle subscription without keeping the process alive", async () => { + vi.useFakeTimers(); + const harness = createClientHarness(); + clients.push(harness.client); + ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" }); + const release = vi.fn(async (_threadId: string) => undefined); + await retainCodexAppServerLiveThread(harness.client, "thread-expired", release); + + await vi.advanceTimersByTimeAsync(EXPECTED_LIVE_THREAD_IDLE_TIMEOUT_MS - 1); + expect(release).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + + expect(release).toHaveBeenCalledExactlyOnceWith("thread-expired"); + }); + + it("protects native-child parents and renews their idle clock after the final child", async () => { + vi.useFakeTimers(); + const harness = createClientHarness(); + clients.push(harness.client); + ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" }); + const release = vi.fn(async (_threadId: string) => undefined); + const unprotect = protectCodexAppServerLiveThread(harness.client, "thread-parent"); + await retainCodexAppServerLiveThread(harness.client, "thread-parent", release); + + await vi.advanceTimersByTimeAsync(EXPECTED_LIVE_THREAD_IDLE_TIMEOUT_MS * 2); + expect(release).not.toHaveBeenCalled(); + unprotect(); + await vi.advanceTimersByTimeAsync(EXPECTED_LIVE_THREAD_IDLE_TIMEOUT_MS - 1); + expect(release).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + + expect(release).toHaveBeenCalledExactlyOnceWith("thread-parent"); + }); + + it("keeps protected parents outside the independent idle-conversation limit", async () => { + const harness = createClientHarness(); + clients.push(harness.client); + ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" }); + const release = vi.fn(async (_threadId: string) => undefined); + const unprotect: Array<() => void> = []; + for (let index = 0; index < EXPECTED_MAX_IDLE_LIVE_THREADS; index += 1) { + const threadId = `parent-${index}`; + unprotect.push(protectCodexAppServerLiveThread(harness.client, threadId)); + await retainCodexAppServerLiveThread(harness.client, threadId, release); + } + await retainCodexAppServerLiveThread(harness.client, "conversation-a", release); + await retainCodexAppServerLiveThread(harness.client, "conversation-b", release); + + expect(release).not.toHaveBeenCalled(); + for (const releaseProtection of unprotect) { + releaseProtection(); + } + await vi.waitFor(() => expect(release).toHaveBeenCalledTimes(2)); + expect(release).toHaveBeenNthCalledWith(1, "conversation-a"); + expect(release).toHaveBeenNthCalledWith(2, "conversation-b"); + }); + + it.each(["thread/archived", "thread/deleted", "thread/closed"])( + "discards only the exact thread after %s", + async (method) => { + const harness = createClientHarness(); + clients.push(harness.client); + ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" }); + await retainCodexAppServerLiveThread(harness.client, "thread-a"); + await retainCodexAppServerLiveThread(harness.client, "thread-b"); + const notificationObserved = new Promise((resolve) => { + harness.client.addNotificationHandler((notification) => { + if (notification.method === method) { + resolve(); + } + }); + }); + + harness.send({ method, params: { threadId: "thread-a" } }); + await notificationObserved; + await expect( + consumeCodexAppServerLiveThread(harness.client, "thread-a"), + ).resolves.toBeUndefined(); + await expect(consumeCodexAppServerLiveThread(harness.client, "thread-b")).resolves.toEqual( + expect.objectContaining({ release: expect.any(Function) }), + ); + }, + ); + + it("clears idle ownership and its timer when the physical client closes", async () => { + vi.useFakeTimers(); + const harness = createClientHarness(); + clients.push(harness.client); + ensureCodexAppServerClientRuntime(harness.client, { agentDir: "/tmp/agent" }); + const release = vi.fn(async (_threadId: string) => undefined); + await retainCodexAppServerLiveThread(harness.client, "thread-closed", release); + + harness.client.close(); + await vi.advanceTimersByTimeAsync(EXPECTED_LIVE_THREAD_IDLE_TIMEOUT_MS); + + expect(release).not.toHaveBeenCalled(); + await expect( + consumeCodexAppServerLiveThread(harness.client, "thread-closed"), + ).resolves.toBeUndefined(); }); }); diff --git a/extensions/codex/src/app-server/client-runtime.ts b/extensions/codex/src/app-server/client-runtime.ts index 5de3c02fc1ec..7b31a7e59491 100644 --- a/extensions/codex/src/app-server/client-runtime.ts +++ b/extensions/codex/src/app-server/client-runtime.ts @@ -1,7 +1,7 @@ /** Client-scoped Codex auth and account observers. */ import { refreshCodexAppServerAuthTokens } from "./auth-bridge.js"; import type { CodexAppServerClient } from "./client.js"; -import type { JsonValue } from "./protocol.js"; +import type { CodexServiceTier, JsonValue } from "./protocol.js"; import { mergeCodexRateLimitsUpdate } from "./rate-limit-cache.js"; import type { CodexAppServerAuthProfileLookup } from "./session-binding.js"; @@ -12,11 +12,30 @@ type ClientRuntimeContext = Omit & type ClientRuntime = { context: ClientRuntimeContext; - retainedThreadId?: string; - retainedThreadConfigFingerprint?: string; - retainedThreadRelease?: Promise; + retainedThreads: Map; + releasingThreads: Map>; + protectedThreads: Map; + evictionTimer?: ReturnType; }; +type RetainedLiveThread = { + configFingerprint?: string; + serviceTier?: CodexServiceTier | null; + expiresAt: number; + release: (threadId: string) => Promise; +}; + +export type CodexAppServerLiveThreadOwnership = { + configFingerprint?: string; + serviceTier?: CodexServiceTier | null; + release: (threadId: string) => Promise; +}; + +/** Match Codex's native grace window without retaining inactive conversations indefinitely. */ +const CODEX_APP_SERVER_LIVE_THREAD_IDLE_TIMEOUT_MS = 30 * 60_000; +/** Native-child parents are active ownership, so only otherwise-idle threads count against this cap. */ +const CODEX_APP_SERVER_LIVE_THREAD_MAX_IDLE = 64; + const configuredClients = new WeakMap(); /** Installs one auth-refresh handler and one rate-limit observer per physical client. */ @@ -31,8 +50,21 @@ export function ensureCodexAppServerClientRuntime( existing.context = context; return; } - const runtime: ClientRuntime = { context }; + const runtime: ClientRuntime = { + context, + retainedThreads: new Map(), + releasingThreads: new Map(), + protectedThreads: new Map(), + }; configuredClients.set(client, runtime); + client.addCloseHandler(() => { + if (runtime.evictionTimer) { + clearTimeout(runtime.evictionTimer); + runtime.evictionTimer = undefined; + } + runtime.retainedThreads.clear(); + runtime.protectedThreads.clear(); + }); client.addRequestHandler(async (request) => { if (request.method !== "account/chatgptAuthTokens/refresh") { return undefined; @@ -52,68 +84,222 @@ export function ensureCodexAppServerClientRuntime( client.addNotificationHandler((notification) => { if (notification.method === "account/rateLimits/updated") { mergeCodexRateLimitsUpdate(client, notification.params); + return; + } + if ( + notification.method === "thread/archived" || + notification.method === "thread/deleted" || + notification.method === "thread/closed" + ) { + const threadId = (notification.params as { threadId?: unknown } | undefined)?.threadId; + if (typeof threadId === "string") { + // Codex already removed server-side ownership; unsubscribing again can + // race a replacement, so only discard this exact local idle entry. + runtime.retainedThreads.delete(threadId); + scheduleRetainedThreadEviction(client, runtime); + } } }); } -/** Keep at most one idle, still-subscribed thread on a physical Codex client. */ +function scheduleRetainedThreadEviction( + client: CodexAppServerClient, + runtime: ClientRuntime, +): void { + if (runtime.evictionTimer) { + clearTimeout(runtime.evictionTimer); + runtime.evictionTimer = undefined; + } + let expiresAt = Number.POSITIVE_INFINITY; + for (const [threadId, thread] of runtime.retainedThreads) { + if (!runtime.protectedThreads.has(threadId)) { + expiresAt = Math.min(expiresAt, thread.expiresAt); + } + } + if (!Number.isFinite(expiresAt)) { + return; + } + runtime.evictionTimer = setTimeout( + () => { + runtime.evictionTimer = undefined; + void evictExpiredRetainedThreads(client, runtime).catch(() => client.close()); + }, + Math.max(0, expiresAt - Date.now()), + ); + runtime.evictionTimer.unref?.(); +} + +async function releaseRetainedThread( + client: CodexAppServerClient, + runtime: ClientRuntime, + threadId: string, +): Promise { + const pendingRelease = runtime.releasingThreads.get(threadId); + if (pendingRelease) { + await pendingRelease; + return false; + } + const retained = runtime.retainedThreads.get(threadId); + if (!retained) { + return false; + } + runtime.retainedThreads.delete(threadId); + scheduleRetainedThreadEviction(client, runtime); + // Keep release ownership addressable until unsubscribe settles. Unrelated + // conversations must stay reusable while only this thread transitions. + const release = retained.release(threadId); + runtime.releasingThreads.set(threadId, release); + try { + await release; + return true; + } finally { + if (runtime.releasingThreads.get(threadId) === release) { + runtime.releasingThreads.delete(threadId); + } + } +} + +async function evictExpiredRetainedThreads( + client: CodexAppServerClient, + runtime: ClientRuntime, +): Promise { + const now = Date.now(); + for (const [threadId, thread] of runtime.retainedThreads) { + if (thread.expiresAt <= now && !runtime.protectedThreads.has(threadId)) { + await releaseRetainedThread(client, runtime, threadId); + } + } + scheduleRetainedThreadEviction(client, runtime); +} + +async function evictExcessIdleThreads( + client: CodexAppServerClient, + runtime: ClientRuntime, +): Promise { + let idleThreadIds = [...runtime.retainedThreads.keys()].filter( + (threadId) => !runtime.protectedThreads.has(threadId), + ); + while (idleThreadIds.length > CODEX_APP_SERVER_LIVE_THREAD_MAX_IDLE) { + await releaseRetainedThread(client, runtime, idleThreadIds[0]!); + idleThreadIds = [...runtime.retainedThreads.keys()].filter( + (threadId) => !runtime.protectedThreads.has(threadId), + ); + } +} + +/** Retain separately owned Codex subscriptions; completing B must never cold-restart A. */ 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, + releaseThread?: (threadId: string) => Promise, configFingerprint?: string, + serviceTier?: CodexServiceTier | null, ): Promise { const runtime = configuredClients.get(client); if (!runtime) { return false; } - if (runtime.retainedThreadRelease) { - await runtime.retainedThreadRelease; + const pendingRelease = runtime.releasingThreads.get(threadId); + if (pendingRelease) { + await pendingRelease; } - if ( - runtime.retainedThreadId !== threadId || - (configFingerprint !== undefined && - runtime.retainedThreadConfigFingerprint !== configFingerprint) - ) { - return false; - } - runtime.retainedThreadId = undefined; - runtime.retainedThreadConfigFingerprint = undefined; + runtime.retainedThreads.delete(threadId); + runtime.retainedThreads.set(threadId, { + configFingerprint, + serviceTier, + expiresAt: Date.now() + CODEX_APP_SERVER_LIVE_THREAD_IDLE_TIMEOUT_MS, + release: + releaseThread ?? + (async (releasedThreadId) => { + await client.request( + "thread/unsubscribe", + { threadId: releasedThreadId }, + { timeoutMs: 5_000 }, + ); + }), + }); + + // Map insertion order is the LRU. Active turns are claimed out of this map, + // and detached native-child parents are pinned until their final child settles. + await evictExcessIdleThreads(client, runtime); + scheduleRetainedThreadEviction(client, runtime); return true; } + +/** Transfer one idle subscription to its next turn or compaction without touching sibling threads. */ +export async function consumeCodexAppServerLiveThread( + client: CodexAppServerClient, + threadId: string, + configFingerprint?: string, +): Promise { + const runtime = configuredClients.get(client); + if (!runtime) { + return undefined; + } + const pendingRelease = runtime.releasingThreads.get(threadId); + if (pendingRelease) { + await pendingRelease; + return undefined; + } + const retained = runtime.retainedThreads.get(threadId); + if ( + !retained || + (configFingerprint !== undefined && retained.configFingerprint !== configFingerprint) + ) { + return undefined; + } + runtime.retainedThreads.delete(threadId); + scheduleRetainedThreadEviction(client, runtime); + return { + configFingerprint: retained.configFingerprint, + serviceTier: retained.serviceTier, + release: retained.release, + }; +} + +/** Reset/end owns the exact thread; failed generation retirement must never release its successor. */ +export async function releaseCodexAppServerLiveThread( + client: CodexAppServerClient, + threadId: string, +): Promise { + const runtime = configuredClients.get(client); + return runtime ? await releaseRetainedThread(client, runtime, threadId) : false; +} + +/** Native child work pins its parent's subscription even after the foreground parent turn ends. */ +export function protectCodexAppServerLiveThread( + client: CodexAppServerClient, + threadId: string, +): () => void { + const runtime = configuredClients.get(client); + if (!runtime) { + return () => undefined; + } + runtime.protectedThreads.set(threadId, (runtime.protectedThreads.get(threadId) ?? 0) + 1); + scheduleRetainedThreadEviction(client, runtime); + let protectedThread = true; + return () => { + if (!protectedThread) { + return; + } + protectedThread = false; + const count = runtime.protectedThreads.get(threadId) ?? 0; + if (count <= 1) { + runtime.protectedThreads.delete(threadId); + const retained = runtime.retainedThreads.get(threadId); + if (retained) { + // A detached child is live activity, not parent idleness. Its terminal + // delivery starts the parent's normal warm-session retention window. + runtime.retainedThreads.delete(threadId); + runtime.retainedThreads.set(threadId, { + ...retained, + expiresAt: Date.now() + CODEX_APP_SERVER_LIVE_THREAD_IDLE_TIMEOUT_MS, + }); + } + } else { + runtime.protectedThreads.set(threadId, count - 1); + } + scheduleRetainedThreadEviction(client, runtime); + void evictExcessIdleThreads(client, runtime).catch(() => client.close()); + }; +} diff --git a/extensions/codex/src/app-server/compact.test.ts b/extensions/codex/src/app-server/compact.test.ts index cff6517f7c96..b5090d245ee4 100644 --- a/extensions/codex/src/app-server/compact.test.ts +++ b/extensions/codex/src/app-server/compact.test.ts @@ -208,7 +208,7 @@ describe("maybeCompactCodexAppServerSession", () => { expect(details.completed).toBe(true); }); - it("resubscribes an evicted session before compacting without displacing its sibling", async () => { + it("compacts a warm session without displacing its independently retained sibling", async () => { const fake = createFakeCodexClient(); setCodexAppServerClientFactoryForTest(async () => fake.client); const sessionFile = await writeTestBinding(); @@ -229,19 +229,13 @@ describe("maybeCompactCodexAppServerSession", () => { compacted: true, }); - expect(fake.request.mock.calls.map(([method]) => method)).toEqual([ - "thread/resume", - "thread/compact/start", - "thread/unsubscribe", - ]); - expect(fake.request).toHaveBeenCalledWith( - "thread/resume", - { threadId: "thread-1", excludeTurns: true }, - expect.objectContaining({ timeoutMs: expect.any(Number) }), - ); + expect(fake.request.mock.calls.map(([method]) => method)).toEqual(["thread/compact/start"]); + await expect( + consumeCodexAppServerLiveThread(fake.client, "thread-1", "config-thread-1"), + ).resolves.toEqual(expect.objectContaining({ configFingerprint: "config-thread-1" })); await expect( consumeCodexAppServerLiveThread(fake.client, "thread-2", "config-thread-2"), - ).resolves.toBe(true); + ).resolves.toEqual(expect.objectContaining({ configFingerprint: "config-thread-2" })); }); it("keeps an owned thread subscribed when a sibling finishes during compaction", async () => { @@ -258,14 +252,17 @@ describe("maybeCompactCodexAppServerSession", () => { fake.completeCompaction(); await expect(pending).resolves.toMatchObject({ ok: true, compacted: true }); - expect(fake.request).toHaveBeenCalledWith( + expect(fake.request).not.toHaveBeenCalledWith( "thread/unsubscribe", { threadId: "thread-1" }, - { timeoutMs: 5_000 }, + expect.anything(), ); + await expect( + consumeCodexAppServerLiveThread(fake.client, "thread-1", "config-thread-1"), + ).resolves.toEqual(expect.objectContaining({ configFingerprint: "config-thread-1" })); await expect( consumeCodexAppServerLiveThread(fake.client, "thread-2", "config-thread-2"), - ).resolves.toBe(true); + ).resolves.toEqual(expect.objectContaining({ configFingerprint: "config-thread-2" })); }); it("preserves an incognito thread's separately owned live subscription", async () => { diff --git a/extensions/codex/src/app-server/compact.ts b/extensions/codex/src/app-server/compact.ts index 92f9da96c290..6b03fdb0fbb6 100644 --- a/extensions/codex/src/app-server/compact.ts +++ b/extensions/codex/src/app-server/compact.ts @@ -21,7 +21,11 @@ import { } from "./attempt-client-cleanup.js"; import { readCodexNotificationItem } from "./attempt-notifications.js"; import { resolveCodexBindingAppServerConnection } from "./binding-connection.js"; -import { consumeCodexAppServerLiveThread } from "./client-runtime.js"; +import { + consumeCodexAppServerLiveThread, + retainCodexAppServerLiveThread, + type CodexAppServerLiveThreadOwnership, +} from "./client-runtime.js"; import { CodexAppServerRpcError, type CodexAppServerClient } from "./client.js"; import { readCodexNotificationThreadId, @@ -546,6 +550,8 @@ async function compactCodexNativeThread( config: params.config, }); let releaseThreadSubscription: (() => Promise) | undefined; + let retainedThreadOwnership: CodexAppServerLiveThreadOwnership | undefined; + let compactionSucceeded = false; const releaseCompactionThread = async (threadId: string) => { if ( await unsubscribeCodexThreadBestEffort(client, { @@ -608,7 +614,11 @@ async function compactCodexNativeThread( if (!isIncognitoSessionKey(params.sessionKey)) { // Remove any idle ownership first: sibling cleanup must not evict // this subscription while compaction still awaits terminal events. - if (!(await consumeCodexAppServerLiveThread(client, binding.threadId))) { + retainedThreadOwnership = await consumeCodexAppServerLiveThread( + client, + binding.threadId, + ); + if (!retainedThreadOwnership) { await resumeCodexAppServerThread({ client, abandonClient: async () => closeCodexStartupClientBestEffort(client), @@ -732,6 +742,7 @@ async function compactCodexNativeThread( sessionId: params.sessionId, threadId: binding.threadId, }); + compactionSucceeded = true; } catch (error) { if (isCodexThreadNotFoundError(error)) { return failedCodexThreadBindingCompactionResult(params, { @@ -754,7 +765,32 @@ async function compactCodexNativeThread( } finally { completionWatch.cancel(); try { - await releaseThreadSubscription?.(); + if (compactionSucceeded && retainedThreadOwnership) { + const ownership = retainedThreadOwnership; + const currentBinding = await options.bindingStore.read(bindingIdentity); + // Reset uses this same generation lease; without it compaction + // could return an obsolete subscription after its owner ended. + const retained = + currentBinding?.threadId === binding.threadId && + (await options.bindingStore.withLease(bindingIdentity, async () => { + const leasedBinding = await options.bindingStore.read(bindingIdentity); + if (leasedBinding?.threadId !== binding.threadId) { + return false; + } + return await retainCodexAppServerLiveThread( + client, + binding.threadId, + ownership.release, + ownership.configFingerprint, + ownership.serviceTier, + ); + })); + if (!retained) { + await releaseThreadSubscription?.(); + } + } else { + await releaseThreadSubscription?.(); + } } finally { if (shouldReleaseDefaultLease) { releaseLeasedSharedCodexAppServerClient(client); diff --git a/extensions/codex/src/app-server/native-subagent-monitor.test.ts b/extensions/codex/src/app-server/native-subagent-monitor.test.ts index 7e12e09d1939..c2964f26f143 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.test.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.test.ts @@ -333,6 +333,45 @@ function taskRecord(params: { } describe("CodexNativeSubagentMonitor", () => { + it("pins a parent subscription until its final independently running child settles", async () => { + const client = createClient(); + const runtime = createRuntime(); + const releaseParentThread = vi.fn(); + const retainParentThread = vi.fn(() => releaseParentThread); + const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { + retainParentThread, + }); + const parent = registerParent(monitor); + + await notifyChildStarted(client, "parent-thread", "child-a"); + await notifyChildStarted(client, "parent-thread", "child-b"); + parent.unregister(); + + expect(retainParentThread).toHaveBeenCalledExactlyOnceWith("parent-thread"); + await client.notify(nativeCompletionNotification({ agentPath: "child-a" })); + expect(releaseParentThread).not.toHaveBeenCalled(); + await client.notify(nativeCompletionNotification({ agentPath: "child-b" })); + + expect(releaseParentThread).toHaveBeenCalledOnce(); + monitor.dispose(); + expect(releaseParentThread).toHaveBeenCalledOnce(); + }); + + it("releases detached parent subscription pins when its physical client closes", async () => { + const client = createClient(); + const runtime = createRuntime(); + const releaseParentThread = vi.fn(); + const monitor = new CodexNativeSubagentMonitor(client as never, runtime, { + retainParentThread: () => releaseParentThread, + }); + registerParent(monitor); + await notifyChildStarted(client); + + client.close(); + + expect(releaseParentThread).toHaveBeenCalledOnce(); + }); + it("keeps native subagent task mirroring on the shared client", async () => { const client = createClient(); const runtime = createRuntime(); diff --git a/extensions/codex/src/app-server/native-subagent-monitor.ts b/extensions/codex/src/app-server/native-subagent-monitor.ts index c018a0f673b4..9dab292dbc29 100644 --- a/extensions/codex/src/app-server/native-subagent-monitor.ts +++ b/extensions/codex/src/app-server/native-subagent-monitor.ts @@ -108,6 +108,7 @@ type MonitorOptions = { completionDeliveryMaxRetries?: number; now?: () => number; retainClient?: () => (() => void) | undefined; + retainParentThread?: (threadId: string) => (() => void) | undefined; }; const DEFAULT_RECOVERY_POLL_DELAYS_MS = [ @@ -153,11 +154,13 @@ function registerMonitor(params: { agentId?: string; runtime?: NativeSubagentMonitorRuntime; retainClient?: () => (() => void) | undefined; + retainParentThread?: (threadId: string) => (() => void) | undefined; }): { unregister: () => void } { let monitor = monitors.get(params.client); if (!monitor) { monitor = new Monitor(params.client, params.runtime ?? defaultRuntime, { retainClient: params.retainClient, + retainParentThread: params.retainParentThread, }); monitors.set(params.client, monitor); } @@ -183,6 +186,8 @@ class Monitor { private readonly removeNotificationHandler: () => void; private readonly removeCloseHandler: () => void; private readonly retainClient?: () => (() => void) | undefined; + private readonly retainParentThread?: (threadId: string) => (() => void) | undefined; + private readonly parentThreadRetentions = new Map void>(); private releaseClientRetention?: () => void; private disposed = false; @@ -198,6 +203,7 @@ class Monitor { options.completionDeliveryMaxRetries ?? this.completionDeliveryRetryDelaysMs.length; this.now = options.now ?? Date.now; this.retainClient = options.retainClient; + this.retainParentThread = options.retainParentThread; this.removeNotificationHandler = client.addNotificationHandler(async (notification) => { if (!NATIVE_SUBAGENT_NOTIFICATION_METHODS.has(notification.method)) { return; @@ -228,6 +234,10 @@ class Monitor { this.unregisterChild(childState); } this.releaseRetainedClient(); + for (const release of this.parentThreadRetentions.values()) { + release(); + } + this.parentThreadRetentions.clear(); for (const state of this.parentStates.values()) { state.ownerCount = 0; } @@ -999,6 +1009,14 @@ class Monitor { } if (!childState) { this.releaseClientRetention ??= this.retainClient?.(); + if (!this.parentThreadRetentions.has(parentThreadId)) { + const releaseParentThread = this.retainParentThread?.(parentThreadId); + if (releaseParentThread) { + // Child completion can be announced on its parent's subscription + // after the foreground parent turn has already released ownership. + this.parentThreadRetentions.set(parentThreadId, releaseParentThread); + } + } childState = { childThreadId, parentThreadId, @@ -1062,6 +1080,15 @@ class Monitor { if (this.childStates.get(childState.childThreadId) === childState) { this.childStates.delete(childState.childThreadId); } + if ( + ![...this.childStates.values()].some( + (remainingChild) => remainingChild.parentThreadId === childState.parentThreadId, + ) + ) { + const releaseParentThread = this.parentThreadRetentions.get(childState.parentThreadId); + this.parentThreadRetentions.delete(childState.parentThreadId); + releaseParentThread?.(); + } const statusRevision = this.threadStatusRevisions.get(childState.childThreadId); if (statusRevision?.readers === 0) { this.threadStatusRevisions.delete(childState.childThreadId); diff --git a/extensions/codex/src/app-server/run-attempt-cleanup.ts b/extensions/codex/src/app-server/run-attempt-cleanup.ts index ceec7aad4e25..94ce6a46b0fb 100644 --- a/extensions/codex/src/app-server/run-attempt-cleanup.ts +++ b/extensions/codex/src/app-server/run-attempt-cleanup.ts @@ -96,26 +96,38 @@ export async function cleanupCodexAttempt( 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; + ? (await bindingStore.read(bindingIdentity))?.threadId === resourceState.thread.threadId && + (await bindingStore.withLease(bindingIdentity, async () => { + // Reset/end uses this same generation lease. Never publish an old + // active turn after its session binding has already been retired. + if ( + (await bindingStore.read(bindingIdentity))?.threadId !== resourceState.thread.threadId + ) { + return false; + } + return await retainCodexAppServerLiveThread( + resourceState.client, + resourceState.thread.threadId, + async (threadId) => { + const released = await unsubscribeCodexThreadBestEffort(resourceState.client, { + threadId, + timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, + }); + if (!released) { + await closeCodexStartupClientBestEffort(resourceState.client); + throw new CodexAppServerUnsafeSubscriptionError( + `Codex retained thread subscription could not be released: ${threadId}`, + ); + } + }, + resourceState.thread.liveThreadConfigFingerprint, + connection.mutable.pluginAppServer.serviceTier, + ); + })) + : false; + // Codex keeps approvals in its native session; independent conversations + // must retain their own subscriptions instead of evicting one another. + const retainLiveThread = retainLiveIncognitoThread || retainedPersistentThread; const bindingReleased = isIncognitoSessionKey(params.sessionKey) && !retainLiveIncognitoThread ? await bindingStore.mutate(bindingIdentity, { diff --git a/extensions/codex/src/app-server/run-attempt-resources.ts b/extensions/codex/src/app-server/run-attempt-resources.ts index 310852118b20..6f2a0b31a3de 100644 --- a/extensions/codex/src/app-server/run-attempt-resources.ts +++ b/extensions/codex/src/app-server/run-attempt-resources.ts @@ -4,6 +4,7 @@ import { type NativeHookRelayRegistrationHandle, } from "openclaw/plugin-sdk/agent-harness-runtime"; import { resolveCodexStartupTimeoutMs } from "./attempt-timeouts.js"; +import { protectCodexAppServerLiveThread } from "./client-runtime.js"; import type { CodexAppServerClient } from "./client.js"; import { resolveCodexToolAbortTerminalReason } from "./dynamic-tool-execution.js"; import { CodexAppServerEventProjector } from "./event-projector.js"; @@ -170,6 +171,8 @@ export function prepareCodexAttemptResources(prompt: CodexAttemptPrompt) { taskRuntimeScope: params.agentHarnessTaskRuntimeScope, agentId: sessionAgentId, retainClient: () => retainSharedCodexAppServerClientIfCurrent(state.client), + retainParentThread: (protectedThreadId) => + protectCodexAppServerLiveThread(state.client, protectedThreadId), }); }; const releaseCurrentRoute = () => { 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 57d3914d7bef..d31f3361a579 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 @@ -17,8 +17,10 @@ import { } from "./run-attempt-test-harness.js"; import { readCodexAppServerBinding, + sessionBindingIdentity, testCodexAppServerBindingStore, } from "./session-binding.test-helpers.js"; +import { retireCodexAppServerSessionGeneration } from "./session-retirement.js"; import { resetSharedCodexAppServerClientForTests, retainSharedCodexAppServerClientIfCurrent, @@ -60,21 +62,33 @@ function createParams( async function waitForHarnessRequest( harness: ReturnType, method: string, -): Promise<{ id: number | string }> { - let request: { id?: number | string; method?: string } | undefined; + startIndex = 0, +): Promise<{ id: number | string; params?: unknown }> { + let request: { id?: number | string; method?: string; params?: unknown } | undefined; await vi.waitFor( () => { request = harness.writes - .map((write) => JSON.parse(write) as { id?: number | string; method?: string }) + .slice(startIndex) + .map( + (write) => + JSON.parse(write) as { id?: number | string; method?: string; params?: unknown }, + ) .find((message) => message.method === method); - expect(request?.id).toBeDefined(); + expect( + request?.id, + `expected ${method} after write ${startIndex}; observed ${JSON.stringify( + harness.writes + .slice(startIndex) + .map((write) => (JSON.parse(write) as { method: string }).method), + )}`, + ).toBeDefined(); }, { interval: 1, timeout: 5_000 }, ); if (request?.id === undefined) { throw new Error(`Codex harness did not write ${method}`); } - return { id: request.id }; + return { id: request.id, params: request.params }; } setupRunAttemptTestHooks(); @@ -166,44 +180,127 @@ describe("Codex app-server main thread cleanup", () => { expect(requests.map((entry) => entry.method)).toEqual(["thread/start", "turn/start"]); }); + it("keeps alternating conversations subscribed on their shared physical Codex client", async () => { + const workspaceDir = path.join(tempDir, "shared-workspace"); + const sessionFiles = { + a: path.join(tempDir, "session-a.jsonl"), + b: path.join(tempDir, "session-b.jsonl"), + }; + const harness = createClientHarness(); + vi.spyOn(CodexAppServerClient, "start").mockReturnValue(harness.client); + + for (const [index, label] of (["a", "b", "a", "b"] as const).entries()) { + const sessionKey = `agent:main:session-${label}`; + const params = createParams(sessionFiles[label], workspaceDir, sessionKey); + params.sessionId = `session-${label}`; + params.runId = `run-${index + 1}`; + params.provider = "openai"; + // Ordinary Codex conversations retain their native tool surface; a + // disableTools turn is intentionally isolated on a temporary thread. + params.disableTools = false; + const requestStart = harness.writes.length; + const run = runCodexAppServerAttempt(params, { + bindingStore: testCodexAppServerBindingStore, + }); + if (index === 0) { + const initialize = await waitForHarnessRequest(harness, "initialize", requestStart); + harness.send({ + id: initialize.id, + result: { userAgent: `openclaw/${CODEX_APP_SERVER_VERSION} (macOS; test)` }, + }); + } + const threadId = `thread-${label}`; + if (index < 2) { + const start = await waitForHarnessRequest(harness, "thread/start", requestStart); + harness.send({ id: start.id, result: threadStartResult(threadId, { cwd: workspaceDir }) }); + } + const turn = await waitForHarnessRequest(harness, "turn/start", requestStart); + const turnId = `turn-${index + 1}`; + harness.send({ id: turn.id, result: turnStartResult(turnId) }); + harness.send({ + method: "turn/completed", + params: { threadId, turnId, turn: { id: turnId, status: "completed" } }, + }); + expect(readAttemptTerminal(await run).aborted).toBe(false); + } + + const userRequestMethods = () => + harness.writes + .map((write) => (JSON.parse(write) as { method: string }).method) + .filter((method) => method !== "initialize" && method !== "initialized"); + expect(userRequestMethods()).toEqual([ + "thread/start", + "turn/start", + "thread/start", + "turn/start", + "turn/start", + "turn/start", + ]); + await expect(readCodexAppServerBinding(sessionFiles.a)).resolves.toMatchObject({ + threadId: "thread-a", + }); + await expect(readCodexAppServerBinding(sessionFiles.b)).resolves.toMatchObject({ + threadId: "thread-b", + }); + + const retirementStart = harness.writes.length; + const retirement = retireCodexAppServerSessionGeneration({ + bindingStore: testCodexAppServerBindingStore, + identity: sessionBindingIdentity({ + agentId: "main", + sessionId: "session-a", + sessionKey: "agent:main:session-a", + }), + mode: "reset", + }); + const unsubscribe = await waitForHarnessRequest(harness, "thread/unsubscribe", retirementStart); + expect(unsubscribe.params).toEqual({ threadId: "thread-a" }); + harness.send({ id: unsubscribe.id, result: {} }); + await expect(retirement).resolves.toBe("applied"); + + const siblingParams = createParams(sessionFiles.b, workspaceDir, "agent:main:session-b"); + siblingParams.sessionId = "session-b"; + siblingParams.runId = "run-surviving-sibling"; + siblingParams.provider = "openai"; + siblingParams.disableTools = false; + const siblingRequestStart = harness.writes.length; + const siblingRun = runCodexAppServerAttempt(siblingParams, { + bindingStore: testCodexAppServerBindingStore, + }); + const siblingTurn = await waitForHarnessRequest(harness, "turn/start", siblingRequestStart); + harness.send({ id: siblingTurn.id, result: turnStartResult("turn-5") }); + harness.send({ + method: "turn/completed", + params: { + threadId: "thread-b", + turnId: "turn-5", + turn: { id: "turn-5", status: "completed" }, + }, + }); + expect(readAttemptTerminal(await siblingRun).aborted).toBe(false); + expect(userRequestMethods().slice(-2)).toEqual(["thread/unsubscribe", "turn/start"]); + }); + it("keeps an incognito thread subscribed for live in-process reuse", async () => { const sessionFile = path.join(tempDir, "incognito-session.jsonl"); const workspaceDir = path.join(tempDir, "incognito-workspace"); const sessionKey = "agent:main:dashboard:incognito-live-thread"; - const requests: Array<{ method: string; params: unknown }> = []; - let notify: (notification: CodexServerNotification) => Promise = async () => undefined; - const request = vi.fn(async (method: string, params?: unknown) => { - requests.push({ method, params }); - if (method === "thread/start") { - return threadStartResult(); - } - if (method === "turn/start") { - return turnStartResult(); - } - return {}; - }); - const clientFactory: CodexAppServerClientFactory = multiplexedClientFactory(async () => { - return { - ...mockClientRuntimeMethods(), - request, - addNotificationHandler: (handler: typeof notify) => { - notify = handler; - return () => undefined; - }, - addRequestHandler: () => () => undefined, - addCloseHandler: () => () => undefined, - } as never; - }); - + const harness = createClientHarness(); + vi.spyOn(CodexAppServerClient, "start").mockReturnValueOnce(harness.client); const run = runCodexAppServerAttempt(createParams(sessionFile, workspaceDir, sessionKey), { bindingStore: testCodexAppServerBindingStore, - clientFactory, }); - await vi.waitFor(() => expect(requests.map((entry) => entry.method)).toContain("turn/start"), { - interval: 1, - timeout: 5_000, + const initialize = await waitForHarnessRequest(harness, "initialize"); + harness.send({ + id: initialize.id, + result: { userAgent: `openclaw/${CODEX_APP_SERVER_VERSION} (macOS; test)` }, }); - await notify({ + const start = await waitForHarnessRequest(harness, "thread/start"); + expect(start.params).toEqual(expect.objectContaining({ ephemeral: true })); + harness.send({ id: start.id, result: threadStartResult() }); + const turn = await waitForHarnessRequest(harness, "turn/start"); + harness.send({ id: turn.id, result: turnStartResult() }); + harness.send({ method: "turn/completed", params: { threadId: "thread-1", @@ -214,8 +311,21 @@ describe("Codex app-server main thread cleanup", () => { const result = await run; expect(readAttemptTerminal(result)).toMatchObject({ aborted: false, timedOut: false }); - expect(requests.map((entry) => entry.method)).toEqual(["thread/start", "turn/start"]); - expect(requests[0]?.params).toEqual(expect.objectContaining({ ephemeral: true })); + await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({ + threadId: "thread-1", + clientId: harness.client.getInstanceId(), + }); + + const requestStart = harness.writes.length; + const retirement = retireCodexAppServerSessionGeneration({ + bindingStore: testCodexAppServerBindingStore, + identity: sessionBindingIdentity({ agentId: "main", sessionId: "session-1", sessionKey }), + mode: "retire", + }); + const unsubscribe = await waitForHarnessRequest(harness, "thread/unsubscribe", requestStart); + expect(unsubscribe.params).toEqual({ threadId: "thread-1" }); + harness.send({ id: unsubscribe.id, result: {} }); + await expect(retirement).resolves.toBe("applied"); }); it.each([ diff --git a/extensions/codex/src/app-server/run-attempt-turn-request.ts b/extensions/codex/src/app-server/run-attempt-turn-request.ts index fb163c8e4584..c0115a663550 100644 --- a/extensions/codex/src/app-server/run-attempt-turn-request.ts +++ b/extensions/codex/src/app-server/run-attempt-turn-request.ts @@ -107,6 +107,7 @@ export async function prepareCodexAttemptTurnRequest( promptText: turnState.codexTurnPromptText, sandboxPolicy: resourceState.codexSandboxPolicy, environmentSelection: resourceState.codexEnvironmentSelection, + clearInheritedServiceTier: resourceState.thread.clearInheritedServiceTier, ...(usesSupervisionConnection ? {} : { model: resourceState.thread.model, modelProvider: resourceState.thread.modelProvider }), diff --git a/extensions/codex/src/app-server/session-retirement.ts b/extensions/codex/src/app-server/session-retirement.ts new file mode 100644 index 000000000000..af6c3b84fbb6 --- /dev/null +++ b/extensions/codex/src/app-server/session-retirement.ts @@ -0,0 +1,69 @@ +import { isIncognitoSessionKey } from "../incognito-session.js"; +import { + CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, + closeCodexStartupClientBestEffort, + CodexAppServerUnsafeSubscriptionError, + unsubscribeCodexThreadBestEffort, +} from "./attempt-client-cleanup.js"; +import { releaseCodexAppServerLiveThread } from "./client-runtime.js"; +import type { + CodexAppServerBindingIdentity, + CodexAppServerBindingStore, + CodexSessionGenerationRetirementResult, +} from "./session-binding.js"; +import { retainSharedCodexAppServerClientByInstanceId } from "./shared-client.js"; + +/** Retire binding and native subscription under the same generation/physical-client ownership fence. */ +export async function retireCodexAppServerSessionGeneration(params: { + bindingStore: CodexAppServerBindingStore; + identity: Extract; + mode: "reset" | "retire"; +}): Promise { + const retireGeneration = () => + params.mode === "reset" + ? params.bindingStore.resetSessionGeneration(params.identity) + : params.bindingStore.retireSessionGeneration(params.identity); + const expectedBinding = await params.bindingStore.read(params.identity); + if (!expectedBinding) { + // Leasing an absent/retired row manufactures state or rejects its fence; + // callers need the original absent/conflict result for reset reclamation. + return await retireGeneration(); + } + return await params.bindingStore.withLease(params.identity, async () => { + const binding = await params.bindingStore.read(params.identity); + if (binding?.threadId !== expectedBinding.threadId) { + return "conflict"; + } + const result = await retireGeneration(); + if (result !== "applied" || !binding?.clientId) { + return result; + } + + // Locate the original physical client only after its exact binding was + // retired; delayed reset events must never unsubscribe a newer generation. + const clientLease = retainSharedCodexAppServerClientByInstanceId(binding.clientId); + if (!clientLease) { + return result; + } + try { + const released = await releaseCodexAppServerLiveThread(clientLease.client, binding.threadId); + if (!released && isIncognitoSessionKey(params.identity.sessionKey)) { + // Ephemeral threads have no rollout to resume, so they intentionally + // bypass idle eviction but still end with their owning OpenClaw session. + const unsubscribed = await unsubscribeCodexThreadBestEffort(clientLease.client, { + threadId: binding.threadId, + timeoutMs: CODEX_APP_SERVER_UNSUBSCRIBE_TIMEOUT_MS, + }); + if (!unsubscribed) { + await closeCodexStartupClientBestEffort(clientLease.client); + throw new CodexAppServerUnsafeSubscriptionError( + `Codex retired session subscription could not be released: ${binding.threadId}`, + ); + } + } + } finally { + clientLease.release(); + } + return result; + }); +} diff --git a/extensions/codex/src/app-server/thread-binding-policy.ts b/extensions/codex/src/app-server/thread-binding-policy.ts index 55ea2fc696cc..5893e2f52285 100644 --- a/extensions/codex/src/app-server/thread-binding-policy.ts +++ b/extensions/codex/src/app-server/thread-binding-policy.ts @@ -22,7 +22,7 @@ export function shouldRotateCodexAppServerBindingForRuntime(params: { type CodexGpt56MultiAgentVersion = "v1" | "v2"; -function resolveCodexGpt56MultiAgentVersion( +export function resolveCodexGpt56MultiAgentVersion( modelRef: string | undefined, ): CodexGpt56MultiAgentVersion | undefined { let modelId = modelRef?.trim().toLowerCase(); diff --git a/extensions/codex/src/app-server/thread-fingerprints.test.ts b/extensions/codex/src/app-server/thread-fingerprints.test.ts index 68e7bc958472..968756e232d5 100644 --- a/extensions/codex/src/app-server/thread-fingerprints.test.ts +++ b/extensions/codex/src/app-server/thread-fingerprints.test.ts @@ -101,13 +101,10 @@ describe("fingerprintCodexThreadConfig", () => { }); 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: "native multi-agent generation", patch: { model: "gpt-5.6-luna" } }, + { setting: "named permissions profile", patch: { permissions: "read-only" } }, { setting: "base instructions", patch: { baseInstructions: "Different base policy." } }, { setting: "developer instructions", patch: { developerInstructions: "Different policy." } }, { setting: "effective config", patch: { config: { features: { hooks: false } } } }, @@ -129,19 +126,20 @@ describe("fingerprintCodexThreadConfig", () => { ); }); - 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.each<{ setting: string; patch: JsonObject }>([ + { setting: "model", patch: { model: "gpt-5.6-terra" } }, + { setting: "requested model", patch: { requestedModel: null } }, + { 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: "personality", patch: { personality: "friendly" } }, + { setting: "working directory", patch: { cwd: "/other/workspace" } }, + ])("preserves the native session when turn/start changes $setting", ({ patch }) => { + expect(fingerprintCodexThreadConfig({ ...request, ...patch }, "openai:personal")).toBe( + fingerprintCodexThreadConfig(request, "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", () => { diff --git a/extensions/codex/src/app-server/thread-fingerprints.ts b/extensions/codex/src/app-server/thread-fingerprints.ts index c2e0afdb9685..8bda7669d2df 100644 --- a/extensions/codex/src/app-server/thread-fingerprints.ts +++ b/extensions/codex/src/app-server/thread-fingerprints.ts @@ -7,6 +7,7 @@ import { type JsonValue, } from "./protocol.js"; import { hashCodexAppServerBindingFingerprint } from "./session-binding.js"; +import { resolveCodexGpt56MultiAgentVersion } from "./thread-binding-policy.js"; export function codexDynamicToolsFingerprint(dynamicTools: CodexDynamicToolSpec[]): string { return fingerprintDynamicTools(dynamicTools); @@ -85,7 +86,7 @@ export function fingerprintJsonObject(value: JsonObject): string { return JSON.stringify(stabilizeJsonValue(value)); } -/** Hash every resume-visible setting without retaining config, credentials, or instructions. */ +/** Hash thread-creation identity; settings already applied by turn/start must not restart Codex. */ export function fingerprintCodexThreadConfig( request: JsonObject, authProfileId?: string, @@ -95,20 +96,24 @@ export function fingerprintCodexThreadConfig( fingerprintJsonObject({ authProfileId: authProfileId ?? null, dynamicToolsFingerprint: dynamicToolsFingerprint ?? null, - model: request.model ?? null, - requestedModel: - request.requestedModel === undefined ? (request.model ?? null) : request.requestedModel, + // Codex fixes its model-selected native multi-agent generation for the + // whole session; only same-generation model changes are turn-mutable. + nativeMultiAgentVersion: + resolveCodexGpt56MultiAgentVersion( + typeof request.requestedModel === "string" + ? request.requestedModel + : typeof request.model === "string" + ? request.model + : undefined, + ) ?? null, 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, + // Named permission profiles are not currently forwarded by turn/start, + // so changing one still requires recreating the native thread. 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 ?? {}, diff --git a/extensions/codex/src/app-server/thread-lifecycle-run.ts b/extensions/codex/src/app-server/thread-lifecycle-run.ts index bee5658ac9a7..0e966b05ad5d 100644 --- a/extensions/codex/src/app-server/thread-lifecycle-run.ts +++ b/extensions/codex/src/app-server/thread-lifecycle-run.ts @@ -19,6 +19,7 @@ import { type CodexAppServerPendingSupervisionBranch, type CodexAppServerThreadBinding, } from "./session-binding.js"; +import { retainSharedCodexAppServerClientByInstanceId } from "./shared-client.js"; import { isTransientWebSearchRestriction, shouldRecheckRecoverablePluginBinding, @@ -83,6 +84,7 @@ export async function startOrResumeThread( params.bindingStore.read(bindingIdentity), ); const initialBoundThreadId = binding?.threadId; + const initialBoundClientId = binding?.clientId; const normalizeBindingModelProvider = ( authProfileId: string | undefined, modelProvider: string | undefined, @@ -95,13 +97,35 @@ export async function startOrResumeThread( config: params.params.config, }); const throwIfAborted = () => throwIfCodexThreadLifecycleAborted(params.signal); - const releaseRetainedThread = (threadId: string) => - releaseCodexRetainedLiveThread({ + const releaseRetainedThread = async ( + threadId: string, + ownerClientId = initialBoundClientId, + ) => { + if (ownerClientId && ownerClientId !== clientId) { + // Auth/runtime rotation selects a new physical client, but its map + // cannot release a subscription owned by the previous app-server. + const previousClient = retainSharedCodexAppServerClientByInstanceId(ownerClientId); + if (!previousClient) { + return; + } + try { + await releaseCodexRetainedLiveThread({ + client: previousClient.client, + lifecycleTiming, + threadId, + }); + } finally { + previousClient.release(); + } + return; + } + await 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. @@ -621,7 +645,7 @@ export async function startOrResumeThread( } } - if (initialBoundThreadId) { + if (initialBoundThreadId && !preserveExistingBinding) { await releaseRetainedThread(initialBoundThreadId); } return await startFreshCodexThread(params, { diff --git a/extensions/codex/src/app-server/thread-lifecycle-types.ts b/extensions/codex/src/app-server/thread-lifecycle-types.ts index f49efb31767c..b3d1eea59e52 100644 --- a/extensions/codex/src/app-server/thread-lifecycle-types.ts +++ b/extensions/codex/src/app-server/thread-lifecycle-types.ts @@ -17,6 +17,7 @@ type CodexAppServerThreadLifecycle = { export type CodexAppServerThreadLifecycleBinding = CodexAppServerThreadBinding & { lifecycle: CodexAppServerThreadLifecycle; liveThreadConfigFingerprint?: string; + clearInheritedServiceTier?: true; }; type CodexThreadFinalConfigPatchDecision = diff --git a/extensions/codex/src/app-server/thread-lifecycle-warm.ts b/extensions/codex/src/app-server/thread-lifecycle-warm.ts index 4d306146a2af..a2ac33d899f9 100644 --- a/extensions/codex/src/app-server/thread-lifecycle-warm.ts +++ b/extensions/codex/src/app-server/thread-lifecycle-warm.ts @@ -196,26 +196,27 @@ export async function tryReuseCodexLiveThread( resumeAuthProfileId, dynamicToolsFingerprint, ); - if ( - !(await consumeCodexAppServerLiveThread( - params.client, - binding.threadId, - liveThreadConfigFingerprint, - )) - ) { + const retainedThread = await consumeCodexAppServerLiveThread( + params.client, + binding.threadId, + liveThreadConfigFingerprint, + ); + if (!retainedThread) { return { prebuiltFinalConfigPatch }; } try { const nativeHookRelayGeneration = prebuiltFinalConfigPatch.nativeHookRelayGeneration ?? binding.nativeHookRelayGeneration; + const model = startModelSelection.model; // Validate ownership even when relay generation is unchanged; reset may - // have replaced the persisted binding since it was first read. + // have replaced the persisted binding since it was first read. Model and + // cwd are sticky turn settings, so future turns and /btw need current facts. const committed = await lifecycleTiming.measure("warm-thread-write-binding", () => params.bindingStore.mutate(bindingIdentity, { kind: "patch", threadId: binding.threadId, - patch: { nativeHookRelayGeneration }, + patch: { cwd: params.cwd, model, nativeHookRelayGeneration }, }), ); if (!committed) { @@ -233,8 +234,13 @@ export async function tryReuseCodexLiveThread( return { binding: { ...binding, + cwd: params.cwd, + model, nativeHookRelayGeneration, liveThreadConfigFingerprint, + ...(retainedThread.serviceTier && resumeParams.serviceTier === undefined + ? { clearInheritedServiceTier: true } + : {}), lifecycle: { action: "resumed" }, }, prebuiltFinalConfigPatch, 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 b8c40fc86445..b621b1f0fab3 100644 --- a/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts +++ b/extensions/codex/src/app-server/thread-lifecycle.binding.test.ts @@ -346,6 +346,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const buildFinalConfigPatch = vi @@ -370,7 +371,7 @@ describe("Codex app-server thread lifecycle bindings", () => { undefined, started.liveThreadConfigFingerprint, ), - ).resolves.toEqual({}); + ).resolves.toBe(true); const reused = await startOrResumeThread(common); expect(started).toMatchObject({ @@ -435,6 +436,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const common = { @@ -454,7 +456,7 @@ describe("Codex app-server thread lifecycle bindings", () => { undefined, started.liveThreadConfigFingerprint, ), - ).resolves.toEqual({}); + ).resolves.toBe(true); await expect(startOrResumeThread(common)).resolves.toMatchObject({ threadId: "thread-warm-isolated", lifecycle: { action: "resumed" }, @@ -470,6 +472,57 @@ describe("Codex app-server thread lifecycle bindings", () => { }); }); + it("refreshes model and workspace ownership when reusing a turn-mutable native session", async () => { + const sessionFile = path.join(tempDir, "warm-model-workspace.jsonl"); + const originalWorkspace = path.join(tempDir, "workspace-original"); + const currentWorkspace = path.join(tempDir, "workspace-current"); + const params = createParams(sessionFile, originalWorkspace); + const request = vi.fn(async (method: string) => { + if (method === "thread/start") { + return threadStartResult("thread-warm-model-workspace", { cwd: originalWorkspace }); + } + throw new Error(`unexpected method: ${method}`); + }); + const client = { + getInstanceId: () => "client-warm-model-workspace", + request, + addNotificationHandler: () => () => undefined, + addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, + } as never; + ensureCodexAppServerClientRuntime(client, { agentDir: originalWorkspace }); + const common = { + client, + params, + cwd: originalWorkspace, + dynamicTools: [], + appServer: createThreadLifecycleAppServerOptions(), + userMcpServersEnabled: false, + }; + const started = await startOrResumeThread(common); + await retainCodexAppServerLiveThread( + client, + started.threadId, + undefined, + started.liveThreadConfigFingerprint, + ); + params.modelId = "gpt-5.5"; + params.workspaceDir = currentWorkspace; + + const reused = await startOrResumeThread({ ...common, cwd: currentWorkspace }); + + expect(request.mock.calls.map(([method]) => method)).toEqual(["thread/start"]); + expect(reused).toMatchObject({ + threadId: "thread-warm-model-workspace", + cwd: currentWorkspace, + model: "gpt-5.5", + }); + await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({ + cwd: currentWorkspace, + model: "gpt-5.5", + }); + }); + 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"); @@ -488,6 +541,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const common = { @@ -549,6 +603,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const common = { @@ -567,7 +622,7 @@ describe("Codex app-server thread lifecycle bindings", () => { undefined, started.liveThreadConfigFingerprint, ), - ).resolves.toEqual({}); + ).resolves.toBe(true); params.contextEngine = { info: { id: "lossless-claw", name: "Lossless Claw", ownsCompaction: true }, @@ -607,6 +662,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const common = { @@ -664,6 +720,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const common = { @@ -716,6 +773,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const common = { @@ -750,17 +808,14 @@ describe("Codex app-server thread lifecycle bindings", () => { expect(resumed.liveThreadConfigFingerprint).not.toBe(started.liveThreadConfigFingerprint); }); - it("releases and resumes a retained thread when its approval policy changes", async () => { + it("keeps a retained thread warm when its turn-level 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") { + if (method === "thread/start") { return threadStartResult("thread-warm-policy"); } - if (method === "thread/unsubscribe") { - return { status: "unsubscribed" }; - } throw new Error(`unexpected method: ${method}`); }); const client = { @@ -768,6 +823,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const appServer = createThreadLifecycleAppServerOptions(); @@ -790,17 +846,8 @@ describe("Codex app-server thread lifecycle bindings", () => { 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); + expect(request.mock.calls.map(([method]) => method)).toEqual(["thread/start"]); + expect(resumed.liveThreadConfigFingerprint).toBe(started.liveThreadConfigFingerprint); }); it("fails closed when a retained mode-transition subscription cannot be released", async () => { @@ -821,6 +868,7 @@ describe("Codex app-server thread lifecycle bindings", () => { request, addNotificationHandler: () => () => undefined, addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, } as never; ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); const abandonClient = vi.fn(async () => undefined); @@ -841,7 +889,7 @@ describe("Codex app-server thread lifecycle bindings", () => { undefined, started.liveThreadConfigFingerprint, ), - ).resolves.toEqual({}); + ).resolves.toBe(true); params.contextEngine = { info: { id: "lossless-claw", name: "Lossless Claw", ownsCompaction: true }, @@ -1961,7 +2009,7 @@ describe("Codex app-server thread lifecycle bindings", () => { }); }); - it("uses a transient Codex thread for report-only fallback completion", async () => { + it("keeps the retained primary subscribed across a transient report-only turn", async () => { const sessionFile = path.join(tempDir, "session.jsonl"); const workspaceDir = path.join(tempDir, "workspace"); const params = createParams(sessionFile, workspaceDir); @@ -1975,19 +2023,36 @@ describe("Codex app-server thread lifecycle bindings", () => { if (method === "thread/resume") { return threadStartResult((requestParams as { threadId: string }).threadId); } + if (method === "thread/unsubscribe") { + return { status: "unsubscribed" }; + } throw new Error(`unexpected method: ${method}`); }); + const client = { + getInstanceId: () => "client-report-only", + request, + addNotificationHandler: () => () => undefined, + addRequestHandler: () => () => undefined, + addCloseHandler: () => () => undefined, + } as never; + ensureCodexAppServerClientRuntime(client, { agentDir: workspaceDir }); - await startOrResumeThread({ - client: { request } as never, + const started = await startOrResumeThread({ + client, params, cwd: workspaceDir, dynamicTools: [], appServer, }); + await retainCodexAppServerLiveThread( + client, + started.threadId, + undefined, + started.liveThreadConfigFingerprint, + ); params.delegationCapability = "report_only"; const restrictedBinding = await startOrResumeThread({ - client: { request } as never, + client, params, cwd: workspaceDir, dynamicTools: [], @@ -1996,7 +2061,7 @@ describe("Codex app-server thread lifecycle bindings", () => { const savedAfterRestriction = await readCodexAppServerBinding(sessionFile); params.delegationCapability = "full"; const resumedBinding = await startOrResumeThread({ - client: { request } as never, + client, params, cwd: workspaceDir, dynamicTools: [], @@ -2007,11 +2072,7 @@ describe("Codex app-server thread lifecycle bindings", () => { 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([ - "thread/start", - "thread/start", - "thread/resume", - ]); + expect(request.mock.calls.map(([method]) => method)).toEqual(["thread/start", "thread/start"]); expect(request.mock.calls[1]?.[1]).toMatchObject({ config: { "features.multi_agent": false, diff --git a/extensions/codex/src/app-server/thread-lifecycle.test.ts b/extensions/codex/src/app-server/thread-lifecycle.test.ts index 52ee6d093a65..9679c32756fe 100644 --- a/extensions/codex/src/app-server/thread-lifecycle.test.ts +++ b/extensions/codex/src/app-server/thread-lifecycle.test.ts @@ -1266,6 +1266,22 @@ describe("Codex app-server native code mode config", () => { expect(request.approvalsReviewer).toBe("auto_review"); }); + it("preserves omitted native tiers until a previously owned sticky tier must be cleared", () => { + const options = { + threadId: "thread-1", + cwd: "/repo", + appServer: createAppServerOptions() as never, + }; + const inherited = buildTurnStartParams(createAttemptParams({ provider: "openai" }), options); + const cleared = buildTurnStartParams(createAttemptParams({ provider: "openai" }), { + ...options, + clearInheritedServiceTier: true, + }); + + expect(inherited).not.toHaveProperty("serviceTier"); + expect(cleared.serviceTier).toBeNull(); + }); + it("allows thread config to opt into Codex code-mode-only", () => { const request = buildThreadStartParams(createAttemptParams({ provider: "openai" }), { cwd: "/repo", diff --git a/extensions/codex/src/app-server/turn-params.ts b/extensions/codex/src/app-server/turn-params.ts index a9da28bbfef7..f9adc036d150 100644 --- a/extensions/codex/src/app-server/turn-params.ts +++ b/extensions/codex/src/app-server/turn-params.ts @@ -29,6 +29,7 @@ export function buildTurnStartParams( skillsCollaborationInstructions?: string; memoryCollaborationInstructions?: string; preserveNativeTurnSettings?: boolean; + clearInheritedServiceTier?: boolean; }, ): CodexTurnStartParams { const modelSelection = options.preserveNativeTurnSettings @@ -62,9 +63,13 @@ export function buildTurnStartParams( ...(modelSelection ? { model: modelSelection.model, personality: CODEX_NATIVE_PERSONALITY_NONE } : {}), + // Codex distinguishes an omitted native default from explicitly clearing + // an OpenClaw-owned priority override left on this exact warm session. ...(options.appServer.serviceTier !== undefined ? { serviceTier: options.appServer.serviceTier } - : {}), + : options.clearInheritedServiceTier + ? { serviceTier: null } + : {}), ...(modelSelection ? { effort: resolveReasoningEffort( diff --git a/extensions/codex/src/command-handler-actions.ts b/extensions/codex/src/command-handler-actions.ts index f51bfc00a0e5..354f1145bee6 100644 --- a/extensions/codex/src/command-handler-actions.ts +++ b/extensions/codex/src/command-handler-actions.ts @@ -1,5 +1,6 @@ import { MODEL_SELECTION_LOCKED_MESSAGE } from "openclaw/plugin-sdk/model-session-runtime"; import type { PluginCommandContext } from "openclaw/plugin-sdk/plugin-entry"; +import { getSessionEntry, resolveStorePath } from "openclaw/plugin-sdk/session-store-runtime"; import { resolveCodexBindingAppServerConnection } from "./app-server/binding-connection.js"; import type { CodexComputerUseSetupParams } from "./app-server/computer-use.js"; import { isJsonObject, type JsonValue } from "./app-server/protocol.js"; @@ -331,9 +332,23 @@ export async function setConversationModel( return "Cannot set Codex model because this command did not include a stable binding identity."; } if (!normalized) { + const currentSession = + ctx.sessionId && ctx.sessionKey + ? getSessionEntry({ + storePath: resolveStorePath(ctx.config.session?.store, { agentId: target.agentId }), + sessionKey: ctx.sessionKey, + hydrateSkillPromptRefs: false, + readConsistency: "latest", + }) + : undefined; + const selectedModel = + currentSession && currentSession.sessionId === ctx.sessionId + ? currentSession.modelOverride + : undefined; const binding = await deps.bindingStore.read(target.identity); - return binding?.model - ? `Codex model: ${formatCodexDisplayText(binding.model)}` + const activeModel = selectedModel ?? binding?.model; + return activeModel + ? `Codex model: ${formatCodexDisplayText(activeModel)}` : "Usage: /codex model "; } return await deps.setCodexConversationModel({ @@ -343,6 +358,15 @@ export async function setConversationModel( model: normalized, agentDir: target.agentDir, config: ctx.config, + ...(ctx.sessionId && ctx.sessionKey + ? { + session: { + agentId: target.agentId, + sessionId: ctx.sessionId, + sessionKey: ctx.sessionKey, + }, + } + : {}), }); } diff --git a/extensions/codex/src/commands.test.ts b/extensions/codex/src/commands.test.ts index 5af9ace52108..1a297aa44e15 100644 --- a/extensions/codex/src/commands.test.ts +++ b/extensions/codex/src/commands.test.ts @@ -4974,6 +4974,38 @@ describe("codex command", () => { expect(result.text).not.toContain("[trusted](https://evil)"); }); + it("reports a conversation-bound model without an OpenClaw session identity", async () => { + await writeTestBinding( + { kind: "conversation", bindingId: "binding-data-1" }, + { threadId: "thread-conversation", cwd: "/repo", model: "bound-model" }, + ); + + const result = await handleCodexCommand( + createContext("model", undefined, { + sessionId: undefined, + sessionKey: undefined, + getCurrentConversationBinding: async () => ({ + bindingId: "binding-1", + pluginId: "codex", + pluginRoot: "/plugin", + channel: "test", + accountId: "default", + conversationId: "conversation", + boundAt: 1, + data: { + kind: "codex-app-server-session", + version: 2, + bindingId: "binding-data-1", + workspaceDir: tempDir, + }, + }), + }), + { deps: createDeps() }, + ); + + expect(result).toEqual({ text: "Codex model: bound-model" }); + }); + it("rejects malformed model commands before persisting the model", async () => { const sessionFile = path.join(tempDir, "session.jsonl"); const setCodexConversationModel = vi.fn(); diff --git a/extensions/codex/src/conversation-control.test.ts b/extensions/codex/src/conversation-control.test.ts index 125ddeca2047..6afc26abbd8e 100644 --- a/extensions/codex/src/conversation-control.test.ts +++ b/extensions/codex/src/conversation-control.test.ts @@ -5,6 +5,11 @@ import path from "node:path"; import { clearRuntimeAuthProfileStoreSnapshots } from "openclaw/plugin-sdk/agent-runtime"; import { MODEL_SELECTION_LOCKED_MESSAGE } from "openclaw/plugin-sdk/model-session-runtime"; import { upsertAuthProfile } from "openclaw/plugin-sdk/provider-auth"; +import { + getSessionEntry, + resolveStorePath, + upsertSessionEntry, +} from "openclaw/plugin-sdk/session-store-runtime"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { buildCodexSupervisionTestConnectionFingerprint, @@ -66,10 +71,6 @@ const sharedClientMocks = vi.hoisted(() => ({ getSharedCodexAppServerClient: vi.fn(), })); -function controlClient(request: ReturnType, clientId = "control-client") { - return { request, getInstanceId: () => clientId }; -} - vi.mock("./app-server/shared-client.js", () => ({ ...sharedClientMocks, getLeasedSharedCodexAppServerClient: sharedClientMocks.getSharedCodexAppServerClient, @@ -267,31 +268,19 @@ describe("codex conversation controls", () => { model: "gpt-5.4", modelProvider: "openai", }); - sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue( - controlClient( - vi.fn(async () => ({ - thread: { id: "thread-1", cwd: tempDir }, - model: "gpt-5.5", - modelProvider: "openai", - })), - ), - ); - await expect( setCodexConversationModel({ sessionFile, agentDir, model: "gpt-5.5" }), ).resolves.toBe("Codex model set to gpt-5.5."); const binding = await readCodexAppServerBinding(sessionFile); - const sharedClientParams = sharedClientMocks.getSharedCodexAppServerClient.mock.calls[0]?.[0]; - expect(sharedClientParams?.agentDir).toBe(agentDir); expect(binding?.threadId).toBe("thread-1"); expect(binding?.authProfileId).toBe("work"); expect(binding?.model).toBe("gpt-5.5"); expect(binding?.modelProvider).toBeUndefined(); - expect(binding?.clientId).toBe("control-client"); + expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled(); }); - it("keeps Guardian reviewer when switching a stale local binding to a provider-qualified OpenAI model", async () => { + it("persists provider-qualified model changes without resuming a subscribed thread", async () => { const sessionFile = path.join(tempDir, "session.jsonl"); await writeCodexAppServerBinding(sessionFile, { threadId: "thread-1", @@ -301,13 +290,6 @@ describe("codex conversation controls", () => { approvalPolicy: "on-request", sandbox: "workspace-write", }); - const request = vi.fn(async (_method: string, _requestParams?: unknown) => ({ - thread: { id: "thread-1", cwd: tempDir }, - model: "gpt-5.5", - modelProvider: "openai", - })); - sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(controlClient(request)); - await expect( setCodexConversationModel({ sessionFile, @@ -316,12 +298,10 @@ describe("codex conversation controls", () => { }), ).resolves.toBe("Codex model set to gpt-5.5."); - const resumeParams = request.mock.calls[0]?.[1] as Record | undefined; const binding = await readCodexAppServerBinding(sessionFile); - expect(resumeParams?.model).toBe("gpt-5.5"); - expect(resumeParams?.modelProvider).toBe("openai"); - expect(resumeParams?.approvalsReviewer).toBe("auto_review"); + expect(binding?.model).toBe("gpt-5.5"); expect(binding?.modelProvider).toBe("openai"); + expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled(); }); it("keeps the bound local provider when switching to another unqualified model", async () => { @@ -334,13 +314,6 @@ describe("codex conversation controls", () => { approvalPolicy: "on-request", sandbox: "workspace-write", }); - const request = vi.fn(async (_method: string, _requestParams?: unknown) => ({ - thread: { id: "thread-1", cwd: tempDir }, - model: "local-model-2", - modelProvider: "lmstudio", - })); - sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(controlClient(request)); - await expect( setCodexConversationModel({ sessionFile, @@ -349,10 +322,11 @@ describe("codex conversation controls", () => { }), ).resolves.toBe("Codex model set to local-model-2."); - const resumeParams = request.mock.calls[0]?.[1] as Record | undefined; - expect(resumeParams?.model).toBe("local-model-2"); - expect(resumeParams?.modelProvider).toBe("lmstudio"); - expect(resumeParams?.approvalsReviewer).toBe("user"); + await expect(readCodexAppServerBinding(sessionFile)).resolves.toMatchObject({ + model: "local-model-2", + modelProvider: "lmstudio", + }); + expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled(); }); it("keeps the bound local provider when reselecting a model id with a slash", async () => { @@ -365,13 +339,6 @@ describe("codex conversation controls", () => { approvalPolicy: "on-request", sandbox: "workspace-write", }); - const request = vi.fn(async (_method: string, _requestParams?: unknown) => ({ - thread: { id: "thread-1", cwd: tempDir }, - model: "openai/gpt-oss-20b", - modelProvider: "lmstudio", - })); - sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue(controlClient(request)); - await expect( setCodexConversationModel({ sessionFile, @@ -380,15 +347,61 @@ describe("codex conversation controls", () => { }), ).resolves.toBe("Codex model set to openai/gpt-oss-20b."); - const resumeParams = request.mock.calls[0]?.[1] as Record | undefined; const binding = await readCodexAppServerBinding(sessionFile); - expect(resumeParams?.model).toBe("openai/gpt-oss-20b"); - expect(resumeParams?.modelProvider).toBe("lmstudio"); - expect(resumeParams?.approvalsReviewer).toBe("user"); + expect(binding?.model).toBe("openai/gpt-oss-20b"); expect(binding?.modelProvider).toBe("lmstudio"); + expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled(); }); - it("escapes model names returned from Codex before chat display", async () => { + it("persists ordinary model selection on SessionEntry without overwriting native ownership", async () => { + const sessionKey = "agent:main:model-session"; + const sessionId = "session-model-authority"; + const identity = { kind: "session" as const, agentId: "main", sessionId, sessionKey }; + const storePath = resolveStorePath(undefined, { agentId: "main" }); + await upsertSessionEntry({ + agentId: "main", + storePath, + sessionKey, + entry: { + sessionId, + updatedAt: Date.now(), + authProfileOverride: "openai:personal", + authProfileOverrideSource: "user", + }, + }); + await testCodexAppServerBindingStore.mutate(identity, { + kind: "set", + binding: { + threadId: "thread-model-authority", + cwd: tempDir, + model: "gpt-5.4", + modelProvider: "openai", + }, + }); + + await expect( + setCodexConversationModelImpl({ + identity, + bindingStore: testCodexAppServerBindingStore, + model: "gpt-5.5", + }), + ).resolves.toBe("Codex model set to gpt-5.5."); + + expect(getSessionEntry({ storePath, sessionKey })).toMatchObject({ + providerOverride: "openai", + modelOverride: "gpt-5.5", + authProfileOverride: "openai:personal", + authProfileOverrideSource: "user", + liveModelSwitchPending: true, + }); + await expect(testCodexAppServerBindingStore.read(identity)).resolves.toMatchObject({ + threadId: "thread-model-authority", + model: "gpt-5.4", + }); + expect(sharedClientMocks.getSharedCodexAppServerClient).not.toHaveBeenCalled(); + }); + + it("escapes requested model names before chat display", async () => { const sessionFile = path.join(tempDir, "session.jsonl"); await writeCodexAppServerBinding(sessionFile, { threadId: "thread-1", @@ -396,18 +409,13 @@ describe("codex conversation controls", () => { model: "gpt-5.4", modelProvider: "openai", }); - sharedClientMocks.getSharedCodexAppServerClient.mockResolvedValue( - controlClient( - vi.fn(async () => ({ - thread: { id: "thread-1", cwd: tempDir }, - model: "gpt-5.5 <@U123> [trusted](https://evil)", - modelProvider: "openai", - })), - ), - ); - - await expect(setCodexConversationModel({ sessionFile, model: "gpt-5.5" })).resolves.toBe( - "Codex model set to gpt-5.5 <\uff20U123> \uff3btrusted\uff3d\uff08https://evil\uff09.", + await expect( + setCodexConversationModel({ + sessionFile, + model: "gpt-5.5 <@U123> [trusted](evil)", + }), + ).resolves.toBe( + "Codex model set to gpt-5.5 <\uff20U123> \uff3btrusted\uff3d\uff08evil\uff09.", ); }); }); diff --git a/extensions/codex/src/conversation-control.ts b/extensions/codex/src/conversation-control.ts index aed1f82be30f..4f4e2afe192c 100644 --- a/extensions/codex/src/conversation-control.ts +++ b/extensions/codex/src/conversation-control.ts @@ -1,16 +1,20 @@ // Codex plugin module implements conversation control behavior. -import { ModelSelectionLockedError } from "openclaw/plugin-sdk/model-session-runtime"; +import { + applyModelOverrideToSessionEntry, + ModelSelectionLockedError, +} from "openclaw/plugin-sdk/model-session-runtime"; +import { + resolveStorePath, + updateSessionStoreEntry, +} from "openclaw/plugin-sdk/session-store-runtime"; import { resolveCodexBindingAppServerConnection } from "./app-server/binding-connection.js"; -import { CODEX_CONTROL_METHODS } from "./app-server/capabilities.js"; import type { CodexAppServerClient } from "./app-server/client.js"; import { isCodexFastServiceTier, - resolveCodexModelBackedReviewerPolicyContext, - resolveCodexAppServerRuntimeOptions, type CodexAppServerApprovalPolicy, type CodexAppServerSandboxMode, } from "./app-server/config.js"; -import type { CodexServiceTier, CodexThreadResumeResponse } from "./app-server/protocol.js"; +import type { CodexServiceTier } from "./app-server/protocol.js"; import { bindingStoreKey, isCodexAppServerNativeAuthProfile, @@ -21,11 +25,7 @@ import { } from "./app-server/session-binding.js"; import { getLeasedSharedCodexAppServerClient, - releaseCodexAppServerClientLease, releaseLeasedSharedCodexAppServerClient, - withLeasedCodexAppServerClientStartSelectionRetry, - type CodexAppServerClientLease, - type CodexAppServerClientOptions, } from "./app-server/shared-client.js"; import { resolveCodexAppServerRequestModelSelection, @@ -189,6 +189,7 @@ export async function setCodexConversationModel(params: { pluginConfig?: unknown; agentDir?: string; config?: CodexAppServerBindingLookup["config"]; + session?: { agentId: string; sessionId: string; sessionKey: string }; }): Promise { const model = params.model.trim(); if (!model) { @@ -199,23 +200,6 @@ export async function setCodexConversationModel(params: { if (binding.connectionScope === "supervision") { throw new ModelSelectionLockedError(); } - const reviewerPolicyContext = resolveCodexModelBackedReviewerPolicyContext({ - provider: "codex", - model, - bindingModelProvider: binding.modelProvider, - bindingModel: binding.model, - nativeAuthProfile: isCodexAppServerNativeAuthProfile({ - authProfileId: binding.authProfileId, - ...lookup, - }), - }); - const runtime = resolveCodexAppServerRuntimeOptions({ - pluginConfig: params.pluginConfig, - modelProvider: reviewerPolicyContext.modelProvider, - model: reviewerPolicyContext.model, - config: params.config, - agentDir: params.agentDir, - }); const modelProvider = resolveConversationControlModelProvider({ authProfileId: binding.authProfileId, bindingModel: binding.model, @@ -229,35 +213,68 @@ export async function setCodexConversationModel(params: { authProfileId: binding.authProfileId, ...lookup, }); - const resumed = await resumeThreadWithOverrides({ - runtime, - threadId: binding.threadId, - authProfileId: binding.authProfileId, - ...lookup, - model: modelSelection.model, - modelProvider: modelSelection.modelProvider, - }); - const response = resumed.response; - const nextModel = response.model ?? modelSelection.model; const nextModelProvider = normalizeCodexAppServerBindingModelProvider({ authProfileId: binding.authProfileId, - modelProvider: response.modelProvider ?? modelSelection.modelProvider, + modelProvider: modelSelection.modelProvider, ...lookup, }); + const nextModel = modelSelection.model; const modelChanged = nextModel !== binding.model || nextModelProvider !== binding.modelProvider; - await patchThreadBinding(params.bindingStore, params.identity, binding.threadId, { - clientId: resumed.clientId, - cwd: response.thread.cwd ?? binding.cwd, - model: nextModel, - modelProvider: nextModelProvider, - ...(modelChanged && binding.contextEngine?.projection - ? { contextEngine: { ...binding.contextEngine, projection: undefined } } - : {}), - approvalPolicy: binding.approvalPolicy, - sandbox: binding.sandbox, - serviceTier: binding.serviceTier ?? runtime.serviceTier ?? undefined, - }); - return `Codex model set to ${formatCodexDisplayText(response.model ?? model)}.`; + const session = + params.session ?? + (params.identity.kind === "session" && params.identity.sessionKey + ? { + agentId: params.identity.agentId, + sessionId: params.identity.sessionId, + sessionKey: params.identity.sessionKey, + } + : undefined); + if (session) { + const updated = await updateSessionStoreEntry({ + storePath: resolveStorePath(params.config?.session?.store, { agentId: session.agentId }), + sessionKey: session.sessionKey, + requireWriteSuccess: true, + update: (entry) => { + if (entry.sessionId !== session.sessionId) { + return null; + } + applyModelOverrideToSessionEntry({ + entry, + selection: { provider: nextModelProvider ?? "openai", model: nextModel }, + preserveAuthProfileOverride: true, + markLiveSwitchPending: true, + }); + return entry; + }, + }); + if (!updated) { + throw new Error("Codex session changed while applying the model selection."); + } + // SessionEntry owns desired selection; the native binding remains the + // currently loaded model so generation transitions still rotate safely. + if (params.identity.kind === "conversation") { + await patchThreadBinding(params.bindingStore, params.identity, binding.threadId, { + model: nextModel, + modelProvider: nextModelProvider, + ...(modelChanged && binding.contextEngine?.projection + ? { contextEngine: { ...binding.contextEngine, projection: undefined } } + : {}), + }); + } else if (modelChanged && binding.contextEngine?.projection) { + await patchThreadBinding(params.bindingStore, params.identity, binding.threadId, { + contextEngine: { ...binding.contextEngine, projection: undefined }, + }); + } + } else { + await patchThreadBinding(params.bindingStore, params.identity, binding.threadId, { + model: nextModel, + modelProvider: nextModelProvider, + ...(modelChanged && binding.contextEngine?.projection + ? { contextEngine: { ...binding.contextEngine, projection: undefined } } + : {}), + }); + } + return `Codex model set to ${formatCodexDisplayText(nextModel)}.`; } export async function setCodexConversationFastMode(params: { @@ -360,55 +377,6 @@ async function patchThreadBinding( } } -async function resumeThreadWithOverrides(params: { - runtime: ReturnType; - threadId: string; - authProfileId?: string; - agentDir?: string; - config?: CodexAppServerBindingLookup["config"]; - model?: string; - modelProvider?: string | null; - approvalPolicy?: CodexAppServerApprovalPolicy; - sandbox?: CodexAppServerSandboxMode; - serviceTier?: CodexServiceTier; -}): Promise<{ response: CodexThreadResumeResponse; clientId: string }> { - const runtime = params.runtime; - const clientOptions = { - startOptions: runtime.start, - timeoutMs: runtime.requestTimeoutMs, - authProfileId: params.authProfileId, - ...buildBindingLookup(params), - } satisfies CodexAppServerClientOptions; - let client = await getLeasedSharedCodexAppServerClient(clientOptions); - const clientLease: CodexAppServerClientLease = { client }; - try { - const response = await withLeasedCodexAppServerClientStartSelectionRetry({ - lease: clientLease, - options: clientOptions, - run: async (requestClient, requestOptions) => - await requestClient.request( - CODEX_CONTROL_METHODS.resumeThread, - { - threadId: params.threadId, - ...(params.model ? { model: params.model } : {}), - ...(params.modelProvider ? { modelProvider: params.modelProvider } : {}), - approvalPolicy: params.approvalPolicy ?? runtime.approvalPolicy, - sandbox: params.sandbox ?? runtime.sandbox, - approvalsReviewer: runtime.approvalsReviewer, - ...(params.serviceTier ? { serviceTier: params.serviceTier } : {}), - }, - requestOptions, - ), - onClientChange: (nextClient) => { - client = nextClient; - }, - }); - return { response, clientId: client.getInstanceId() }; - } finally { - releaseCodexAppServerClientLease(clientLease); - } -} - function buildBindingLookup(params: { agentDir?: string; config?: CodexAppServerBindingLookup["config"]; diff --git a/src/gateway/gateway-cli-backend.connect.test.ts b/src/gateway/gateway-cli-backend.connect.test.ts index 1143cb768acd..b253d3862148 100644 --- a/src/gateway/gateway-cli-backend.connect.test.ts +++ b/src/gateway/gateway-cli-backend.connect.test.ts @@ -4,6 +4,7 @@ import type { AddressInfo } from "node:net"; import path from "node:path"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { WebSocketServer } from "ws"; +import { GATEWAY_CLIENT_CAPS } from "../../packages/gateway-protocol/src/client-info.js"; import { loadOrCreateDeviceIdentity } from "../infra/device-identity.js"; import { createSuiteTempRootTracker } from "../test-helpers/temp-dir.js"; import { GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES } from "../utils/message-channel.js"; @@ -94,9 +95,12 @@ describe("gateway cli backend connect", () => { await tempDirs.cleanup(); }); - it( - "connects a test gateway client through the live helper", - async () => { + it.each([ + { label: "default capabilities", caps: undefined }, + { label: "plugin approval capability", caps: [GATEWAY_CLIENT_CAPS.PLUGIN_APPROVALS] }, + ])( + "connects a test gateway client through the live helper with $label", + async ({ caps }) => { const token = `test-${Date.now()}`; const deviceIdentity = await createTempDeviceIdentity(); const server = await startMinimalGatewayServer({ token }); @@ -109,6 +113,7 @@ describe("gateway cli backend connect", () => { deviceIdentity, timeoutMs: GATEWAY_CONNECT_OPERATION_TIMEOUT_MS, maxAttemptTimeoutMs: GATEWAY_CONNECT_OPERATION_TIMEOUT_MS, + ...(caps ? { caps } : {}), requestTimeoutMs: GATEWAY_CONNECT_OPERATION_TIMEOUT_MS, waitForEventLoopReady: false, }); @@ -121,6 +126,7 @@ describe("gateway cli backend connect", () => { expect(connectClient?.displayName).toBe("vitest-live"); expect(connectClient?.version).toBe("dev"); expect(connectClient?.mode).toBe(GATEWAY_CLIENT_MODES.TEST); + expect(server.connectParams?.caps).toEqual(caps ?? []); expect(server.requests).toEqual(["connect", "health"]); } finally { await client diff --git a/src/gateway/gateway-cli-backend.live-helpers.ts b/src/gateway/gateway-cli-backend.live-helpers.ts index 86d0c10ab2fe..285b0ec7d665 100644 --- a/src/gateway/gateway-cli-backend.live-helpers.ts +++ b/src/gateway/gateway-cli-backend.live-helpers.ts @@ -402,6 +402,7 @@ export async function connectTestGatewayClient(params: { timeoutMs?: number; maxAttemptTimeoutMs?: number; clientDisplayName?: string | null; + caps?: string[]; requestTimeoutMs?: number; tickWatchTimeoutMs?: number; waitForEventLoopReady?: boolean; @@ -444,6 +445,7 @@ async function connectClientOnce(params: { timeoutMs: number; deviceIdentity?: DeviceIdentity; clientDisplayName?: string | null; + caps?: string[]; requestTimeoutMs?: number; tickWatchTimeoutMs?: number; waitForEventLoopReady?: boolean; @@ -478,6 +480,7 @@ async function connectClientOnce(params: { clientName: GATEWAY_CLIENT_NAMES.TEST, clientVersion: "dev", mode: GATEWAY_CLIENT_MODES.TEST, + ...(params.caps ? { caps: params.caps } : {}), connectChallengeTimeoutMs: params.timeoutMs, deviceIdentity: params.deviceIdentity, onHelloOk: () => finish({ client }), diff --git a/src/gateway/gateway-codex-harness.live.test.ts b/src/gateway/gateway-codex-harness.live.test.ts index 17b6b655e6ed..114a8508486d 100644 --- a/src/gateway/gateway-codex-harness.live.test.ts +++ b/src/gateway/gateway-codex-harness.live.test.ts @@ -6,6 +6,7 @@ import os from "node:os"; import path from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { describe, expect, it } from "vitest"; +import { GATEWAY_CLIENT_CAPS } from "../../packages/gateway-protocol/src/client-info.js"; import type { EventFrame } from "../../packages/gateway-protocol/src/index.js"; import { renderBitmapTextPngBase64, @@ -62,6 +63,9 @@ const CODEX_HARNESS_SUBAGENT_PROBE = isTruthyEnvValue( const CODEX_HARNESS_GUARDIAN_PROBE = isTruthyEnvValue( process.env.OPENCLAW_LIVE_CODEX_HARNESS_GUARDIAN_PROBE, ); +const CODEX_HARNESS_MULTI_SESSION_PROBE = isTruthyEnvValue( + process.env.OPENCLAW_LIVE_CODEX_HARNESS_MULTI_SESSION_PROBE, +); const CODEX_HARNESS_CODE_MODE_ONLY = isTruthyEnvValue( process.env.OPENCLAW_LIVE_CODEX_HARNESS_CODE_MODE_ONLY, ); @@ -193,7 +197,7 @@ const observedCodexThreadIds = new Map(); const observedCodexClientIds = new Map(); const observedCodexThreadActions = new Map(); -type GuardianPluginApprovalDecision = "allow-once" | "deny"; +type GuardianPluginApprovalDecision = "allow-once" | "allow-always" | "deny"; type CodexHarnessThinkingLevel = | "off" | "minimal" @@ -485,6 +489,8 @@ async function assertCodexHarnessTranscriptModelIdentity(params: { } async function writeLiveGatewayConfig(params: { + codexApprovalPolicy?: "untrusted"; + codexApprovalsReviewer?: "user"; codexAppServerMode?: "guardian" | "yolo"; codeModeOnly?: boolean; compactionMode: CodexCompactionStressMode; @@ -511,6 +517,10 @@ async function writeLiveGatewayConfig(params: { config: { appServer: { mode: params.codexAppServerMode ?? "yolo", + ...(params.codexApprovalPolicy ? { approvalPolicy: params.codexApprovalPolicy } : {}), + ...(params.codexApprovalsReviewer + ? { approvalsReviewer: params.codexApprovalsReviewer } + : {}), ...(appServerArgs ? { args: appServerArgs } : {}), ...(params.codeModeOnly === true ? { codeModeOnly: true } : {}), ...(params.loopDetectionPreToolUseRelay === false @@ -685,6 +695,99 @@ function recordCodexAttemptIdentity(params: { observedCodexThreadActions.set(params.sessionKey, action as string); } +async function verifyCodexMultiSessionApprovalPersistence(params: { + client: GatewayClient; + getResolvedPluginApprovalCount: () => number; + setPluginApprovalDecision: (decision: GuardianPluginApprovalDecision | undefined) => void; + workspace: string; +}): Promise { + const targetName = "codex-session-approval-proof.txt"; + const targetPath = path.join(params.workspace, targetName); + let previousContent = "OPENCLAW-CODEX-SESSION-INITIAL"; + await fs.writeFile(targetPath, `${previousContent}\n`, "utf8"); + const sessionKeys = { + a: "agent:dev:live-codex-harness-session-a", + b: "agent:dev:live-codex-harness-session-b", + } as const; + const firstThreadIds = new Map(); + let physicalClientId: string | undefined; + const startingApprovalCount = params.getResolvedPluginApprovalCount(); + params.setPluginApprovalDecision("allow-always"); + try { + for (const [turn, session] of (["a", "b", "a", "b"] as const).entries()) { + const sessionKey = sessionKeys[session]; + const expectedReply = `CODEX-SESSION-${session.toUpperCase()}-${turn + 1}`; + const expectedContent = `OPENCLAW-CODEX-SESSION-${session.toUpperCase()}-${turn + 1}`; + const patch = [ + "*** Begin Patch", + `*** Update File: ${targetName}`, + "@@", + `-${previousContent}`, + `+${expectedContent}`, + "*** End Patch", + ].join("\n"); + const patchCode = `const result = await tools.apply_patch(${JSON.stringify(patch)});\ntext(result);`; + const { text, events } = await requestAgentTextWithEvents({ + client: params.client, + eventPrefixes: ["codex_app_server.", "tool", "approval"], + sessionKey, + message: [ + "Use the exec tool exactly once before replying.", + "Set its JavaScript source to exactly:", + patchCode, + "Do not call apply_patch directly or use a shell.", + `After the patch succeeds, reply exactly ${expectedReply} and nothing else.`, + ].join("\n"), + }); + expect(text).toContain(expectedReply); + recordCodexAttemptIdentity({ events, sessionKey }); + expect(await fs.readFile(targetPath, "utf8")).toBe(`${expectedContent}\n`); + expect( + events.some( + (event) => + event.stream === "tool" && + event.data?.name === "apply_patch" && + event.data?.phase === "result" && + event.data?.status === "completed" && + event.data?.isError === false, + ), + `expected a completed native file change for session ${session}`, + ).toBe(true); + previousContent = expectedContent; + + const threadId = observedCodexThreadIds.get(sessionKey); + const clientId = observedCodexClientIds.get(sessionKey); + expect(threadId).toBeTruthy(); + expect(clientId).toBeTruthy(); + const initialThreadId = firstThreadIds.get(session); + if (initialThreadId) { + expect(threadId).toBe(initialThreadId); + } else { + if (session === "b") { + expect(threadId).not.toBe(firstThreadIds.get("a")); + } + firstThreadIds.set(session, threadId as string); + } + if (physicalClientId) { + expect(clientId).toBe(physicalClientId); + } else { + physicalClientId = clientId; + } + // Exec "always" creates a shared durable rule; file approvals alone + // support Codex's per-thread acceptForSession cache for this proof. + expect(params.getResolvedPluginApprovalCount()).toBe( + startingApprovalCount + Math.min(turn + 1, 2), + ); + } + expect(firstThreadIds.get("a")).not.toBe(firstThreadIds.get("b")); + console.log( + `[codex-session-proof] A=${firstThreadIds.get("a")} B=${firstThreadIds.get("b")} client=${physicalClientId} approvals=2 turns=4`, + ); + } finally { + params.setPluginApprovalDecision(undefined); + } +} + async function verifyCodexCodeModeOnlyDynamicToolProbe(params: { client: GatewayClient; sessionKey: string; @@ -1705,7 +1808,11 @@ describeLive("gateway live (Codex harness)", () => { port, token, workspace, - codexAppServerMode: CODEX_HARNESS_GUARDIAN_PROBE ? "guardian" : "yolo", + codexAppServerMode: + CODEX_HARNESS_GUARDIAN_PROBE || CODEX_HARNESS_MULTI_SESSION_PROBE ? "guardian" : "yolo", + ...(CODEX_HARNESS_MULTI_SESSION_PROBE + ? { codexApprovalPolicy: "untrusted", codexApprovalsReviewer: "user" } + : {}), codeModeOnly: CODEX_HARNESS_CODE_MODE_ONLY, compactionMode: CODEX_HARNESS_COMPACTION_MODE, ...(CODEX_HARNESS_DISABLE_LOOP_RELAY ? { loopDetectionPreToolUseRelay: false } : {}), @@ -1770,6 +1877,10 @@ describeLive("gateway live (Codex harness)", () => { timeoutMs: GATEWAY_CONNECT_TIMEOUT_MS, requestTimeoutMs: CODEX_HARNESS_REQUEST_TIMEOUT_MS, clientDisplayName: "vitest-codex-harness-live", + // Approval events require an explicit renderer capability even for admin clients. + ...(CODEX_HARNESS_MULTI_SESSION_PROBE + ? { caps: [GATEWAY_CLIENT_CAPS.PLUGIN_APPROVALS] } + : {}), onEvent: (event) => { gatewayEvents.push(event); maybeResolveGuardianPluginApproval(event); @@ -1804,6 +1915,18 @@ describeLive("gateway live (Codex harness)", () => { sessionKey, }); + if (CODEX_HARNESS_MULTI_SESSION_PROBE) { + await verifyCodexMultiSessionApprovalPersistence({ + client: activeClient, + getResolvedPluginApprovalCount: () => resolvedGuardianPluginApprovalIds.size, + setPluginApprovalDecision: (decision) => { + guardianPluginApprovalDecision = decision; + }, + workspace, + }); + break; + } + if (CODEX_HARNESS_SUBAGENT_PROBE) { logCodexLiveStep("subagent-probe:start", { sessionKey }); await verifyCodexSubagentProbe({ client: activeClient, sessionKey });