fix(agent): routed codex v2 compaction through websockets

- Reused the live Codex provider session for WebSocket-first V2 compaction.
- Fell back to SSE V2 on WebSocket transport failure before the existing V1 fallback.
- Propagated the configured WebSocket preference through manual, automatic, and advisor compaction paths.
- Added transport reuse and fallback regression coverage.

Fixes #7198

packages/agent/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed Codex V2 remote compaction bypassing the provider's live WebSocket transport before trying SSE ([#7198](https://github.com/can1357/oh-my-pi/issues/7198)).
+
 ## [17.2.2] - 2026-07-31
 
 ### Fixed

packages/agent/src/compaction/compaction-v2-streaming.ts

@@ -13,7 +13,9 @@ import { applyCodexResponsesLiteShape } from "@oh-my-pi/pi-ai/providers/openai-c
 import {
 	createOpenAICodexCompactionRequestContext,
 	createOpenAICodexCompatibilityMetadata,
+	type OpenAICodexCompactionBody,
 	type OpenAICodexCompatibilityMetadata,
+	openCodexCompactionEventStream,
 } from "@oh-my-pi/pi-ai/providers/openai-codex-responses";
 import {
 	getOpenAIPromptCacheKey,
@@ -128,6 +130,14 @@ function isOpenAiV2CompatibleModel(model: Model): boolean {
 	return api === "openai-responses" || api === "azure-openai-responses" || api === "openai-codex-responses";
 }
 
+function shouldUseCodexProviderTransport(model: Model): model is Model<"openai-codex-responses"> {
+	return (
+		model.api === "openai-codex-responses" &&
+		model.remoteCompaction?.v2Endpoint === undefined &&
+		model.remoteCompaction?.streamingEndpoint === undefined
+	);
+}
+
 function resolveOpenAiResponsesEndpoint(baseUrl: string | undefined): string {
 	const rawBase = baseUrl && baseUrl.length > 0 ? baseUrl : "https://api.openai.com/v1";
 	const normalizedBase = rawBase.replace(/\/+$/, "");
@@ -228,6 +238,7 @@ export async function requestCompactionV2Streaming(
 		retryWait?: (delayMs: number, signal?: AbortSignal) => Promise<void>;
 		providerSessionState?: Map<string, ProviderSessionState>;
 		codexCompaction?: CodexCompactionContext;
+		preferWebsockets?: boolean;
 	},
 ): Promise<CompactionV2Response> {
 	const endpoint = getCompactionV2Endpoint(model);
@@ -238,31 +249,29 @@ export async function requestCompactionV2Streaming(
 	const fetchImpl = options?.fetch ?? globalThis.fetch;
 	const retryWait = options?.retryWait ?? ((delayMs: number) => Bun.sleep(delayMs));
 	const isCodexResponses = compactionV2Api(model) === "openai-codex-responses" || model.provider === "openai-codex";
-	const codexMetadata = isCodexResponses
-		? createOpenAICodexCompatibilityMetadata({
-				sessionId: request.sessionId,
-				providerSessionState: options?.providerSessionState,
-				requestKind: "compaction",
-				compaction: createOpenAICodexCompactionRequestContext({
-					context: options?.codexCompaction,
-					implementation: "responses_compaction_v2",
-				}),
-			})
-		: undefined;
+	const codexMetadata =
+		isCodexResponses && !shouldUseCodexProviderTransport(model)
+			? createOpenAICodexCompatibilityMetadata({
+					sessionId: request.sessionId,
+					providerSessionState: options?.providerSessionState,
+					requestKind: "compaction",
+					compaction: createOpenAICodexCompactionRequestContext({
+						context: options?.codexCompaction,
+						implementation: "responses_compaction_v2",
+					}),
+				})
+			: undefined;
 	let lastError: Error | undefined;
 
 	for (let attempt = 0; attempt <= V2_COMPACTION_MAX_RETRIES; attempt++) {
 		const timeoutSignal = withRequestTimeout(signal, options?.timeoutMs ?? V2_COMPACTION_TIMEOUT_MS);
 		try {
-			return await attemptCompactionV2Streaming(
-				endpoint,
-				apiKey,
-				model,
-				request,
-				fetchImpl,
-				timeoutSignal,
+			return await attemptCompactionV2Streaming(endpoint, apiKey, model, request, fetchImpl, timeoutSignal, {
 				codexMetadata,
-			);
+				providerSessionState: options?.providerSessionState,
+				codexCompaction: options?.codexCompaction,
+				preferWebsockets: options?.preferWebsockets,
+			});
 		} catch (err) {
 			const error = err instanceof Error ? err : new Error(String(err));
 			if (signal?.aborted) throw error;
@@ -292,15 +301,20 @@ async function attemptCompactionV2Streaming(
 	model: Model,
 	request: CompactionV2Request,
 	fetchImpl: FetchImpl,
-	signal?: AbortSignal,
-	codexMetadata?: OpenAICodexCompatibilityMetadata,
+	signal: AbortSignal | undefined,
+	options: {
+		codexMetadata?: OpenAICodexCompatibilityMetadata;
+		providerSessionState?: Map<string, ProviderSessionState>;
+		codexCompaction?: CodexCompactionContext;
+		preferWebsockets?: boolean;
+	},
 ): Promise<CompactionV2Response> {
 	// Faithful to Codex: append the compaction trigger as the final input item
 	// of an otherwise-normal Responses request, then stream the result. `store`
 	// stays false — compaction must never persist a server-side response object.
 	const cacheOptions = { sessionId: request.sessionId, promptCacheKey: request.promptCacheKey };
 	const promptCacheKey = getOpenAIPromptCacheKey(cacheOptions);
-	const body: Record<string, unknown> = {
+	const body: OpenAICodexCompactionBody = {
 		model: request.model,
 		input: [...request.input, COMPACTION_TRIGGER_ITEM],
 		instructions: request.instructions,
@@ -318,8 +332,8 @@ async function attemptCompactionV2Streaming(
 		...(promptCacheKey ? { prompt_cache_key: promptCacheKey } : {}),
 		...(request.tools && request.tools.length > 0 ? { tools: request.tools, tool_choice: "auto" } : {}),
 	};
-	if (codexMetadata) {
-		body.client_metadata = codexMetadata.clientMetadata;
+	if (options.codexMetadata) {
+		body.client_metadata = options.codexMetadata.clientMetadata;
 	}
 	// Responses Lite models take the same rewrite on the compaction stream:
 	// instructions/tools ride as input items (codex-rs `compact_remote_v2`
@@ -327,9 +341,27 @@ async function attemptCompactionV2Streaming(
 	if (model.useResponsesLite) {
 		applyCodexResponsesLiteShape(body);
 	}
+
+	if (shouldUseCodexProviderTransport(model)) {
+		const eventStream = await openCodexCompactionEventStream(model, body, {
+			apiKey,
+			signal,
+			fetch: fetchImpl,
+			sessionId: request.sessionId,
+			providerSessionState: options.providerSessionState,
+			preferWebsockets: options.preferWebsockets,
+			responsesLite: model.useResponsesLite,
+			codexCompaction: createOpenAICodexCompactionRequestContext({
+				context: options.codexCompaction,
+				implementation: "responses_compaction_v2",
+			}),
+		});
+		return collectCompactionV2Events(eventStream, request);
+	}
+
 	const response = await fetchImpl(endpoint, {
 		method: "POST",
-		headers: buildCompactionV2Headers(model, apiKey, request, codexMetadata),
+		headers: buildCompactionV2Headers(model, apiKey, request, options.codexMetadata),
 		body: stringifyJson(body),
 		signal,
 	});
@@ -400,6 +432,33 @@ function buildCompactionV2Headers(
 	return headers;
 }
 
+interface CompactionV2CollectionState {
+	outputItemCount: number;
+	compactionItems: Array<Record<string, unknown>>;
+	sawCompleted: boolean;
+	usage: CompactionV2Usage | undefined;
+}
+
+function createCompactionV2CollectionState(): CompactionV2CollectionState {
+	return {
+		outputItemCount: 0,
+		compactionItems: [],
+		sawCompleted: false,
+		usage: undefined,
+	};
+}
+
+async function collectCompactionV2Events(
+	events: AsyncIterable<Record<string, unknown>>,
+	request: CompactionV2Request,
+): Promise<CompactionV2Response> {
+	const state = createCompactionV2CollectionState();
+	for await (const event of events) {
+		handleCompactionV2Event(event, undefined, state);
+	}
+	return finishCompactionV2Collection(state, request);
+}
+
 async function collectCompactionV2Output(
 	response: Response,
 	request: CompactionV2Request,
@@ -409,13 +468,7 @@ async function collectCompactionV2Output(
 		throw new Error("No response body for V2 compaction streaming");
 	}
 
-	const state = {
-		outputItemCount: 0,
-		compactionItems: [] as Array<Record<string, unknown>>,
-		sawCompleted: false,
-		usage: undefined as CompactionV2Usage | undefined,
-	};
-
+	const state = createCompactionV2CollectionState();
 	try {
 		const decoder = new TextDecoder();
 		let buffer = "";
@@ -462,6 +515,13 @@ async function collectCompactionV2Output(
 		reader.releaseLock();
 	}
 
+	return finishCompactionV2Collection(state, request);
+}
+
+function finishCompactionV2Collection(
+	state: CompactionV2CollectionState,
+	request: CompactionV2Request,
+): CompactionV2Response {
 	if (!state.sawCompleted) {
 		throw new Error("V2 compaction stream closed before response.completed");
 	}
@@ -477,7 +537,6 @@ async function collectCompactionV2Output(
 		compactionItem,
 		request.retainedMessageBudget,
 	);
-
 	return {
 		compactionItem,
 		replacementHistory,
@@ -490,12 +549,7 @@ async function collectCompactionV2Output(
 function handleCompactionV2SseEvent(
 	data: string,
 	eventName: string | undefined,
-	state: {
-		outputItemCount: number;
-		compactionItems: Array<Record<string, unknown>>;
-		sawCompleted: boolean;
-		usage: CompactionV2Usage | undefined;
-	},
+	state: CompactionV2CollectionState,
 ): void {
 	if (data === "[DONE]") return;
 	let event: Record<string, unknown>;
@@ -504,7 +558,14 @@ function handleCompactionV2SseEvent(
 	} catch (err) {
 		throw new Error(`V2 compaction stream parse failed: ${err instanceof Error ? err.message : String(err)}`);
 	}
+	handleCompactionV2Event(event, eventName, state);
+}
 
+function handleCompactionV2Event(
+	event: Record<string, unknown>,
+	eventName: string | undefined,
+	state: CompactionV2CollectionState,
+): void {
 	const type = typeof event.type === "string" ? event.type : eventName;
 	if (type === "response.output_item.done") {
 		state.outputItemCount++;

packages/agent/src/compaction/compaction.ts

@@ -809,6 +809,8 @@ export interface SummaryOptions {
 	promptCacheKey?: string;
 	/** Mutable provider state used to keep Codex compaction on the live session identity. */
 	providerSessionState?: Map<string, ProviderSessionState>;
+	/** Whether Codex remote compaction should prefer the provider WebSocket transport. */
+	preferWebsockets?: boolean;
 	/** Classification shared by every provider request in this logical compaction. */
 	codexCompaction?: CodexCompactionContext;
 	/** Provider-visible tools for remote compaction transports that replay native tool history. */
@@ -1432,6 +1434,7 @@ export async function compact(
 		sessionId: options?.sessionId,
 		promptCacheKey: options?.promptCacheKey,
 		providerSessionState: options?.providerSessionState,
+		preferWebsockets: options?.preferWebsockets,
 		codexCompaction: options?.codexCompaction,
 		tools: options?.tools,
 		fetch: options?.fetch,
@@ -1509,6 +1512,7 @@ export async function compact(
 						requestCompactionV2Streaming(model, key, request, signal, {
 							fetch: summaryOptions.fetch,
 							providerSessionState: summaryOptions.providerSessionState,
+							preferWebsockets: summaryOptions.preferWebsockets,
 							codexCompaction: summaryOptions.codexCompaction,
 						}),
 					{ signal },

packages/agent/test/remote-compaction.test.ts

@@ -98,6 +98,87 @@ function sseResponse(events: Array<Record<string, unknown>>): Response {
 	return new Response(body, { headers: { "content-type": "text/event-stream" } });
 }
 
+interface CodexCompactionTestSocket {
+	readyState: number;
+	readonly sent: Array<Record<string, unknown>>;
+	emit(event: Record<string, unknown>): void;
+}
+
+function installCodexCompactionWebSocket(options?: {
+	failConnect?: boolean;
+	respond?: (socket: CodexCompactionTestSocket, request: Record<string, unknown>) => void;
+}): {
+	sockets: CodexCompactionTestSocket[];
+	restore(): void;
+} {
+	const originalWebSocket = globalThis.WebSocket;
+	const sockets: CodexCompactionTestSocket[] = [];
+
+	class CodexCompactionWebSocket implements CodexCompactionTestSocket {
+		static readonly CONNECTING = 0;
+		static readonly OPEN = 1;
+		static readonly CLOSING = 2;
+		static readonly CLOSED = 3;
+
+		readyState = CodexCompactionWebSocket.CONNECTING;
+		binaryType: "blob" | "arraybuffer" | "nodebuffer" = "blob";
+		onopen: ((event: Event) => void) | null = null;
+		onmessage: ((event: MessageEvent) => void) | null = null;
+		onerror: ((event: Event) => void) | null = null;
+		onclose: ((event: Event) => void) | null = null;
+		readonly sent: Array<Record<string, unknown>> = [];
+		readonly handshakeHeaders = { "x-codex-turn-state": `compaction-state-${sockets.length}` };
+
+		constructor(
+			readonly url: string,
+			readonly socketOptions?: { headers?: Record<string, string> },
+		) {
+			sockets.push(this);
+			queueMicrotask(() => {
+				if (options?.failConnect) {
+					this.readyState = CodexCompactionWebSocket.CLOSED;
+					this.onerror?.(new Event("error"));
+					this.onclose?.(new Event("close"));
+					return;
+				}
+				this.readyState = CodexCompactionWebSocket.OPEN;
+				this.onopen?.(new Event("open"));
+			});
+		}
+
+		send(data: string): void {
+			const parsed: unknown = JSON.parse(data);
+			if (!isRecord(parsed)) throw new Error("Expected Codex WebSocket request object");
+			this.sent.push(parsed);
+			options?.respond?.(this, parsed);
+		}
+
+		emit(event: Record<string, unknown>): void {
+			this.onmessage?.(new MessageEvent("message", { data: JSON.stringify(event) }));
+		}
+
+		close(): void {
+			this.readyState = CodexCompactionWebSocket.CLOSED;
+		}
+	}
+
+	Object.defineProperty(globalThis, "WebSocket", {
+		configurable: true,
+		writable: true,
+		value: CodexCompactionWebSocket,
+	});
+	return {
+		sockets,
+		restore: () => {
+			Object.defineProperty(globalThis, "WebSocket", {
+				configurable: true,
+				writable: true,
+				value: originalWebSocket,
+			});
+		},
+	};
+}
+
 describe("buildOpenAiNativeHistory custom tool calls", () => {
 	test("serializes customWireName tool calls as custom_tool_call + custom_tool_call_output", () => {
 		const patch = "*** Begin Patch\n*** End Patch\n";
@@ -828,6 +909,23 @@ describe("Responses Lite remote compaction", () => {
 		};
 	}
 
+	function compactionV2Events(encryptedContent: string): Array<Record<string, unknown>> {
+		return [
+			{
+				type: "response.output_item.done",
+				output_index: 0,
+				item: { type: "compaction", encrypted_content: encryptedContent },
+			},
+			{
+				type: "response.done",
+				response: {
+					id: `response-${encryptedContent}`,
+					usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 },
+				},
+			},
+		];
+	}
+
 	test("V1 compaction sends the lite header and input-item instructions", async () => {
 		const model = makeCodexLiteModel();
 		let captured: CapturedLiteExchange | undefined;
@@ -884,15 +982,8 @@ describe("Responses Lite remote compaction", () => {
 		);
 		let captured: CapturedLiteExchange | undefined;
 		const fetchMock: FetchImpl = async (_input, init) => {
-			captured = captureLite(init);
-			return sseResponse([
-				{
-					type: "response.output_item.done",
-					output_index: 0,
-					item: { type: "compaction", encrypted_content: "enc" },
-				},
-				{ type: "response.completed", response: { usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 } } },
-			]);
+			captured = captureStreamLite(init);
+			return sseResponse(compactionV2Events("enc"));
 		};
 
 		expect(shouldUseCompactionV2Streaming(model)).toBe(true);
@@ -930,6 +1021,139 @@ describe("Responses Lite remote compaction", () => {
 		expect(captured?.body.input?.at(-1)).toEqual({ type: "compaction_trigger" });
 	});
 
+	test("V2 compaction reuses the live Codex WebSocket transport when preferred", async () => {
+		const providerSessionState = new Map<string, ProviderSessionState>();
+		const webSocket = installCodexCompactionWebSocket({
+			respond: (socket, outbound) => {
+				const input = outbound.input;
+				const isCompaction =
+					Array.isArray(input) && input.some(item => isRecord(item) && item.type === "compaction_trigger");
+				const events = isCompaction
+					? compactionV2Events("enc-websocket")
+					: [
+							{
+								type: "response.output_item.done",
+								item: {
+									type: "message",
+									id: "message-live-turn",
+									role: "assistant",
+									status: "completed",
+									content: [{ type: "output_text", text: "live response" }],
+								},
+							},
+							{
+								type: "response.done",
+								response: {
+									id: "response-live-turn",
+									status: "completed",
+									usage: { input_tokens: 8, output_tokens: 2, total_tokens: 10 },
+								},
+							},
+						];
+				for (const event of events) socket.emit(event);
+			},
+		});
+		try {
+			const model = makeCodexLiteModel({ preferWebsockets: true });
+			const sessionId = "codex-websocket-compaction";
+			const fetchMock = vi.fn(async () => {
+				throw new Error("WebSocket-first V2 compaction unexpectedly used SSE");
+			});
+			const liveTurn = await ai
+				.streamSimple(
+					model,
+					{
+						systemPrompt: ["You are a helpful assistant."],
+						messages: [{ role: "user", content: "Start the turn", timestamp: Date.now() }],
+					},
+					{
+						apiKey: "test-key",
+						fetch: fetchMock,
+						sessionId,
+						preferWebsockets: true,
+						providerSessionState,
+					},
+				)
+				.result();
+			expect(liveTurn.stopReason).toBe("stop");
+
+			const request = buildCompactionV2Request(
+				model,
+				[{ type: "message", role: "user", content: [{ type: "input_text", text: "real user" }] }],
+				"compact instructions",
+				{ sessionId },
+			);
+			const result = await requestCompactionV2Streaming(model, "test-key", request, undefined, {
+				fetch: fetchMock,
+				preferWebsockets: true,
+				providerSessionState,
+				codexCompaction: TEST_CODEX_COMPACTION,
+			});
+
+			const sentRequest = webSocket.sockets[0]?.sent[1];
+			const sentInput = sentRequest?.input;
+			expect(fetchMock).not.toHaveBeenCalled();
+			expect(webSocket.sockets).toHaveLength(1);
+			expect(webSocket.sockets[0]?.sent).toHaveLength(2);
+			expect(sentRequest?.type).toBe("response.create");
+			expect(Array.isArray(sentInput) ? sentInput.at(-1) : undefined).toEqual({ type: "compaction_trigger" });
+			expect(result.compactionItem).toEqual({ type: "compaction", encrypted_content: "enc-websocket" });
+			expect(
+				getOpenAICodexTransportDetails(model, {
+					sessionId,
+					providerSessionState,
+				}),
+			).toMatchObject({
+				lastTransport: "websocket",
+				websocketConnected: true,
+				canAppend: false,
+			});
+		} finally {
+			for (const state of providerSessionState.values()) state.close();
+			providerSessionState.clear();
+			webSocket.restore();
+		}
+	});
+
+	test("V2 compaction falls back from a failed Codex WebSocket connection to SSE", async () => {
+		const providerSessionState = new Map<string, ProviderSessionState>();
+		const webSocket = installCodexCompactionWebSocket({ failConnect: true });
+		try {
+			const model = makeCodexLiteModel({ preferWebsockets: true });
+			const request = buildCompactionV2Request(
+				model,
+				[{ type: "message", role: "user", content: [{ type: "input_text", text: "real user" }] }],
+				"compact instructions",
+				{ sessionId: "codex-websocket-fallback" },
+			);
+			const fetchMock = vi.fn(async () => sseResponse(compactionV2Events("enc-sse")));
+
+			const result = await requestCompactionV2Streaming(model, "test-key", request, undefined, {
+				fetch: fetchMock,
+				preferWebsockets: true,
+				providerSessionState,
+				codexCompaction: TEST_CODEX_COMPACTION,
+			});
+
+			expect(webSocket.sockets).toHaveLength(1);
+			expect(fetchMock).toHaveBeenCalledTimes(1);
+			expect(result.compactionItem).toEqual({ type: "compaction", encrypted_content: "enc-sse" });
+			expect(
+				getOpenAICodexTransportDetails(model, {
+					sessionId: "codex-websocket-fallback",
+					providerSessionState,
+				}),
+			).toMatchObject({
+				lastTransport: "sse",
+				websocketDisabled: true,
+			});
+		} finally {
+			for (const state of providerSessionState.values()) state.close();
+			providerSessionState.clear();
+			webSocket.restore();
+		}
+	});
+
 	test("compact fan-out keeps local Codex summaries on one classified turn", async () => {
 		const model = makeCodexLiteModel();
 		const captured: CapturedLiteExchange[] = [];
@@ -1019,42 +1243,13 @@ describe("Responses Lite remote compaction", () => {
 	});
 
 	test("local Codex compaction isolates and closes transient websocket sessions", async () => {
-		const originalWebSocket = global.WebSocket;
-		const sockets: AgentCompactionWebSocket[] = [];
 		let responseCount = 0;
-
-		class AgentCompactionWebSocket {
-			static readonly CONNECTING = 0;
-			static readonly OPEN = 1;
-			static readonly CLOSING = 2;
-			static readonly CLOSED = 3;
-
-			readyState = AgentCompactionWebSocket.CONNECTING;
-			binaryType: "blob" | "arraybuffer" | "nodebuffer" = "blob";
-			onopen: ((event: Event) => void) | null = null;
-			onmessage: ((event: MessageEvent) => void) | null = null;
-			onerror: ((event: Event) => void) | null = null;
-			onclose: ((event: Event) => void) | null = null;
-			readonly handshakeHeaders = {
-				"x-codex-turn-state": `agent-compaction-state-${sockets.length}`,
-			};
-
-			constructor(
-				readonly url: string,
-				readonly options?: { headers?: Record<string, string> },
-			) {
-				sockets.push(this);
-				queueMicrotask(() => {
-					this.readyState = AgentCompactionWebSocket.OPEN;
-					this.onopen?.(new Event("open"));
-				});
-			}
-
-			send(_data: string): void {
+		const webSocket = installCodexCompactionWebSocket({
+			respond: socket => {
 				responseCount += 1;
 				const responseId = `response-${responseCount}`;
 				const messageId = `message-${responseCount}`;
-				const text = sockets[0] === this ? "main response" : "local summary";
+				const text = responseCount === 1 ? "main response" : "local summary";
 				const events: Record<string, unknown>[] = [
 					{
 						type: "response.output_item.added",
@@ -1086,19 +1281,12 @@ describe("Responses Lite remote compaction", () => {
 						},
 					},
 				];
-				for (const event of events) {
-					this.onmessage?.({ data: JSON.stringify(event) } as MessageEvent);
-				}
-			}
-
-			close(): void {
-				this.readyState = AgentCompactionWebSocket.CLOSED;
-			}
-		}
+				for (const event of events) socket.emit(event);
+			},
+		});
 
 		const providerSessionState = new Map<string, ProviderSessionState>();
 		try {
-			global.WebSocket = AgentCompactionWebSocket as unknown as typeof WebSocket;
 			const model = makeCodexLiteModel({ preferWebsockets: true });
 			const sessionId = "agent-compaction-isolation";
 			const fetchMock: FetchImpl = async () => {
@@ -1115,8 +1303,8 @@ describe("Responses Lite remote compaction", () => {
 				)
 				.result();
 			expect(main.stopReason).toBe("stop");
-			expect(sockets).toHaveLength(1);
-			expect(sockets[0]?.readyState).toBe(AgentCompactionWebSocket.OPEN);
+			expect(webSocket.sockets).toHaveLength(1);
+			expect(webSocket.sockets[0]?.readyState).toBe(globalThis.WebSocket.OPEN);
 
 			const preparation: CompactionPreparation = {
 				firstKeptEntryId: "kept-1",
@@ -1140,10 +1328,10 @@ describe("Responses Lite remote compaction", () => {
 			});
 
 			expect(result.summary).toContain("local summary");
-			expect(sockets).toHaveLength(3);
-			expect(sockets[0]?.readyState).toBe(AgentCompactionWebSocket.OPEN);
-			expect(sockets[1]?.readyState).toBe(AgentCompactionWebSocket.CLOSED);
-			expect(sockets[2]?.readyState).toBe(AgentCompactionWebSocket.CLOSED);
+			expect(webSocket.sockets).toHaveLength(3);
+			expect(webSocket.sockets[0]?.readyState).toBe(globalThis.WebSocket.OPEN);
+			expect(webSocket.sockets[1]?.readyState).toBe(globalThis.WebSocket.CLOSED);
+			expect(webSocket.sockets[2]?.readyState).toBe(globalThis.WebSocket.CLOSED);
 			expect(
 				getOpenAICodexTransportDetails(model, {
 					sessionId,
@@ -1156,7 +1344,7 @@ describe("Responses Lite remote compaction", () => {
 		} finally {
 			for (const state of providerSessionState.values()) state.close();
 			providerSessionState.clear();
-			global.WebSocket = originalWebSocket;
+			webSocket.restore();
 		}
 	});
 });

packages/ai/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed provider-native Codex compaction streams bypassing WebSocket-first transport selection and SSE transport fallback ([#7198](https://github.com/can1357/oh-my-pi/issues/7198)).
+
 ## [17.2.2] - 2026-07-31
 
 ### Added

packages/ai/src/providers/openai-codex-responses.ts

@@ -69,6 +69,7 @@ import { adaptSchemaForStrict, NO_STRICT, sanitizeSchemaForOpenAIResponses, tool
 import { notifyRawSseEvent } from "../utils/sse-debug";
 import { compactGrammarDefinition } from "./grammar";
 import {
+	type CodexLiteShapedBody,
 	type CodexReasoningContext,
 	type CodexRequestOptions,
 	type InputItem,
@@ -164,6 +165,17 @@ export interface OpenAICodexResponsesOptions extends StreamOptions {
 	onModerationMetadata?: (metadata: unknown) => void;
 }
 
+/** Raw V2 compaction body accepted by the Codex transport selector. */
+export interface OpenAICodexCompactionBody extends CodexLiteShapedBody {
+	model: string;
+	[key: string]: unknown;
+}
+
+/** Transport controls for a provider-native Codex V2 compaction stream. */
+export interface OpenAICodexCompactionStreamOptions extends OpenAICodexResponsesOptions {
+	apiKey: string;
+}
+
 /** Inputs for synthesizing Codex request identity outside the normal stream path. */
 export interface OpenAICodexCompatibilityMetadataOptions {
 	sessionId?: string;
@@ -1359,12 +1371,16 @@ function createRequestSetup(options: OpenAICodexResponsesOptions | undefined): C
 	};
 }
 
-async function buildCodexRequestContext(
+function createCodexRequestContext(
 	model: Model<"openai-codex-responses">,
-	context: Context,
+	transformedBody: RequestBody,
 	options: OpenAICodexResponsesOptions | undefined,
-	output: AssistantMessage,
-): Promise<CodexRequestContext> {
+	contextOptions: {
+		isolateCompactionTransport: boolean;
+		startNewTurn?: boolean;
+		turnStartedAtUnixMs?: number;
+	},
+): CodexRequestContext {
 	const apiKey = options?.apiKey || getEnvApiKey(model.provider) || "";
 	if (!apiKey) {
 		throw new AIError.MissingApiKeyError(model.provider);
@@ -1373,15 +1389,12 @@ async function buildCodexRequestContext(
 	const accountId = getCodexAccountId(apiKey);
 	const baseUrl = model.baseUrl || CODEX_BASE_URL;
 	const url = resolveCodexResponsesUrl(baseUrl);
-	const promptCacheKey = normalizeOpenAIPromptCacheKey(options?.promptCacheKey ?? options?.sessionId);
 	const transportSessionId = normalizeOpenAIPromptCacheKey(options?.sessionId);
 	const codexClientVersion = CODEX_CLIENT_VERSION;
-	const transformedBody = await buildTransformedCodexRequestBody(model, context, options, promptCacheKey);
-
 	const requestHeaders = { ...(model.headers ?? {}), ...(options?.headers ?? {}) };
 	const rawRequestDump: RawHttpRequestDump = {
 		provider: model.provider,
-		api: output.api,
+		api: model.api,
 		model: model.id,
 		method: "POST",
 		url,
@@ -1389,7 +1402,10 @@ async function buildCodexRequestContext(
 	};
 
 	const providerSessionState = getCodexProviderSessionState(options?.providerSessionState);
-	const isolatedTransportState = options?.codexCompaction ? createCodexProviderSessionState() : undefined;
+	const isolatedTransportState =
+		contextOptions.isolateCompactionTransport && options?.codexCompaction
+			? createCodexProviderSessionState()
+			: undefined;
 	const transportProviderSessionState = isolatedTransportState ?? providerSessionState;
 	const responsesLite = resolveCodexResponsesLite(model, options?.responsesLite);
 	const sessionKey = getCodexWebSocketSessionKey(transportSessionId, model, accountId, apiKey, baseUrl, responsesLite);
@@ -1412,20 +1428,14 @@ async function buildCodexRequestContext(
 		websocketState.turnState = sharedWebsocketState.turnState;
 		websocketState.modelsEtag = sharedWebsocketState.modelsEtag;
 	}
-	const withinTurnContinuation = isCodexWithinTurnContinuation(context);
 	const metadataSessionId = transportSessionId ?? crypto.randomUUID();
 	const metadataSession = getOrCreateCodexMetadataSessionState(metadataSessionId, providerSessionState);
 	const compaction = options?.codexCompaction;
 	const requestKind: OpenAICodexRequestKind = compaction ? "compaction" : "turn";
-	const startNewTurn = resolveCodexStartNewTurn(
-		metadataSession,
-		requestKind,
-		compaction,
-		compaction ? undefined : !withinTurnContinuation,
-	);
+	const startNewTurn = resolveCodexStartNewTurn(metadataSession, requestKind, compaction, contextOptions.startNewTurn);
 	if (websocketState && startNewTurn) {
-		// Codex scopes turn-state to one turn. Mid-turn compaction and tool-loop
-		// follow-ups preserve it; new user or compaction turns start without it.
+		// Codex scopes turn-state to one turn. Mid-turn compaction preserves it;
+		// a pre-turn or standalone compaction starts without it.
 		websocketState.turnState = undefined;
 	}
 	const requestMetadata = createCodexRequestMetadata(metadataSession, requestKind, {
@@ -1434,7 +1444,7 @@ async function buildCodexRequestContext(
 			? startNewTurn || !metadataSession.turnId
 				? Date.now()
 				: undefined
-			: getCodexTurnStartedAtUnixMs(context),
+			: contextOptions.turnStartedAtUnixMs,
 		clientMetadata: transformedBody.client_metadata,
 		parentTurnId: options?.parentTurnId,
 		compaction,
@@ -1458,6 +1468,20 @@ async function buildCodexRequestContext(
 	};
 }
 
+async function buildCodexRequestContext(
+	model: Model<"openai-codex-responses">,
+	context: Context,
+	options: OpenAICodexResponsesOptions | undefined,
+): Promise<CodexRequestContext> {
+	const promptCacheKey = normalizeOpenAIPromptCacheKey(options?.promptCacheKey ?? options?.sessionId);
+	const transformedBody = await buildTransformedCodexRequestBody(model, context, options, promptCacheKey);
+	return createCodexRequestContext(model, transformedBody, options, {
+		isolateCompactionTransport: true,
+		startNewTurn: options?.codexCompaction ? undefined : !isCodexWithinTurnContinuation(context),
+		turnStartedAtUnixMs: options?.codexCompaction ? undefined : getCodexTurnStartedAtUnixMs(context),
+	});
+}
+
 /** @internal Exported for tests. */
 export async function buildTransformedCodexRequestBody(
 	model: Model<"openai-codex-responses">,
@@ -1565,6 +1589,84 @@ async function openInitialCodexEventStream(
 	}
 	return openCodexSseTransport(model, requestContext, requestSetup, options, websocketState, transformedBody);
 }
+
+function toCodexRequestBody(body: OpenAICodexCompactionBody): RequestBody {
+	const request: RequestBody = { model: body.model };
+	for (const key in body) {
+		if (key !== "model") request[key] = body[key];
+	}
+	return request;
+}
+
+/**
+ * Open a provider-native V2 compaction stream through Codex's WebSocket-first
+ * transport, replaying WebSocket transport failures over SSE.
+ */
+export async function openCodexCompactionEventStream(
+	model: Model<"openai-codex-responses">,
+	body: OpenAICodexCompactionBody,
+	options: OpenAICodexCompactionStreamOptions,
+): Promise<AsyncGenerator<Record<string, unknown>>> {
+	const requestSetup = createRequestSetup(options);
+	let requestContext: CodexRequestContext;
+	let initial: {
+		eventStream: AsyncGenerator<Record<string, unknown>>;
+		requestBodyForState: RequestBody;
+		transport: CodexTransport;
+	};
+	try {
+		requestContext = createCodexRequestContext(model, toCodexRequestBody(body), options, {
+			isolateCompactionTransport: false,
+		});
+		initial = await openInitialCodexEventStream(model, options, requestSetup, requestContext);
+	} catch (error) {
+		requestSetup.requestAbortController.abort();
+		throw error;
+	}
+
+	if (requestContext.websocketState) {
+		requestContext.websocketState.lastTransport = initial.transport;
+		// The compaction request may use the existing append baseline, but the
+		// replacement history makes that baseline stale for the next normal turn.
+		resetCodexWebSocketAppendState(requestContext.websocketState);
+	}
+	return streamCodexCompactionEvents(model, options, requestSetup, requestContext, initial);
+}
+
+async function* streamCodexCompactionEvents(
+	model: Model<"openai-codex-responses">,
+	options: OpenAICodexCompactionStreamOptions,
+	requestSetup: CodexRequestSetup,
+	requestContext: CodexRequestContext,
+	initial: {
+		eventStream: AsyncGenerator<Record<string, unknown>>;
+		requestBodyForState: RequestBody;
+		transport: CodexTransport;
+	},
+): AsyncGenerator<Record<string, unknown>> {
+	let completed = false;
+	try {
+		try {
+			yield* initial.eventStream;
+		} catch (error) {
+			if (
+				options.signal?.aborted ||
+				initial.transport !== "websocket" ||
+				!(error instanceof CodexWebSocketTransportError)
+			) {
+				throw error;
+			}
+			const state = requestContext.websocketState;
+			if (state) recordCodexWebSocketFailure(state, true);
+			const fallback = await openCodexSseTransport(model, requestContext, requestSetup, options, state);
+			if (state) state.lastTransport = fallback.transport;
+			yield* fallback.eventStream;
+		}
+		completed = true;
+	} finally {
+		if (!completed) requestSetup.requestAbortController.abort();
+	}
+}
 async function openCodexWebSocketTransport(
 	model: Model<"openai-codex-responses">,
 	options: OpenAICodexResponsesOptions | undefined,
@@ -2709,7 +2811,7 @@ export const streamOpenAICodexResponses: StreamFunction<"openai-codex-responses"
 		let requestContext: CodexRequestContext | undefined;
 
 		try {
-			requestContext = await buildCodexRequestContext(model, context, options, output);
+			requestContext = await buildCodexRequestContext(model, context, options);
 			const initialTransport = await openInitialCodexEventStream(model, options, requestSetup, requestContext);
 			const runtime = new CodexStreamRuntime({
 				...initialTransport,

packages/coding-agent/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed manual and automatic Codex compaction dropping the configured OpenAI WebSocket preference ([#7198](https://github.com/can1357/oh-my-pi/issues/7198)).
+
 ## [17.2.2] - 2026-07-31
 
 ### Added

packages/coding-agent/src/session/agent-session.ts

@@ -1382,6 +1382,7 @@ export class AgentSession {
 			extensionRunner: this.#extensionRunner,
 			sideStreamFn: this.#sideStreamFn,
 			providerSessionState: this.#providerSessionState,
+			preferWebsockets: this.#preferWebsockets,
 			model: () => this.model,
 			thinkingLevel: () => this.thinkingLevel,
 			isDisposed: () => this.#isDisposed,

packages/coding-agent/src/session/session-advisors.ts

@@ -1449,6 +1449,7 @@ export class SessionAdvisors {
 						promptCacheKey: advisorProviderSessionId,
 						metadata: advisorMetadata,
 						providerSessionState: this.#host.providerSessionState,
+						preferWebsockets: this.#host.preferWebsockets,
 						codexCompaction,
 					},
 				);

packages/coding-agent/src/session/session-maintenance.ts

@@ -180,6 +180,7 @@ export interface SessionMaintenanceHost {
 	extensionRunner: ExtensionRunner | undefined;
 	sideStreamFn: StreamFn;
 	providerSessionState: Map<string, ProviderSessionState>;
+	preferWebsockets: boolean | undefined;
 	model(): Model | undefined;
 	thinkingLevel(): ThinkingLevel | undefined;
 	isDisposed(): boolean;
@@ -1541,6 +1542,7 @@ export class SessionMaintenance {
 						sessionId: this.#host.sessionId(),
 						promptCacheKey: this.#host.sessionId(),
 						providerSessionState: this.#host.providerSessionState,
+						preferWebsockets: this.#host.preferWebsockets,
 						// Route every summarization HTTP request through the
 						// session's side-stream transport so the provider
 						// concurrency cap (e.g. providers.ollama-cloud.maxConcurrency)
@@ -2587,6 +2589,7 @@ export class SessionMaintenance {
 									sessionId: this.#host.sessionId(),
 									promptCacheKey: this.#host.sessionId(),
 									providerSessionState: this.#host.providerSessionState,
+									preferWebsockets: this.#host.preferWebsockets,
 									codexCompaction,
 								},
 							);