refactor(codex): 移除 envelope/request_patching 层，用 context var 统一 compact 状态判断

- 删除 CodexOAuthEnvelope 和 request_patching 模块，Codex 不再需要 envelope 层
- 移除 _aether_compact 请求体内部标记，改用 is_codex_compact_request() 集中查询
- 简化 OpenAI CLI adapter，移除 Codex 专用的 get_cli_extra_headers/build_test_request_body 逻辑
- 移除 Codex behavior variant 注册（same_format/cross_format）
- normalizer patch_same_format_request 对 codex 变为 no-op
- 更新相关测试适配新的架构

src/api/handlers/base/chat_handler_base.py

@@ -785,7 +785,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
         else:
             # 同格式：按原逻辑做轻量清理（子类可覆盖以移除不需要的字段）
             request_body = self.prepare_provider_request_body(request_body)
-            # 同格式时也需要应用 target_variant 转换（如 Codex）
+            # 同格式 Provider 仍可能声明 target_variant
             if same_format_variant:
                 request_body = registry.convert_request(
                     request_body,

src/api/handlers/base/cli_stream_mixin.py

@@ -375,7 +375,7 @@ class CliStreamMixin:
             url_model = (
                 self.get_model_for_url(request_body, mapped_model) or mapped_model or ctx.model
             )
-            # 同格式时也需要应用 target_variant 转换（如 Codex）
+            # 同格式 Provider 仍可能声明 target_variant
             if target_variant and provider_api_format:
                 registry = get_format_converter_registry()
                 request_body = registry.convert_request(

src/api/handlers/base/cli_sync_mixin.py

@@ -196,7 +196,7 @@ class CliSyncMixin:
                 url_model = (
                     self.get_model_for_url(request_body, mapped_model) or mapped_model or model
                 )
-                # 同格式时也需要应用 target_variant 转换（如 Codex）
+                # 同格式 Provider 仍可能声明 target_variant
                 if target_variant and provider_api_format:
                     registry = get_format_converter_registry()
                     request_body = registry.convert_request(

src/api/handlers/openai_cli/adapter.py

@@ -46,21 +46,20 @@ class OpenAICliAdapter(CliAdapterBase):
         self._compact = compact
 
     async def handle(self, context: ApiRequestContext) -> Any:
-        """处理 CLI API 请求 -- compact 模式下注入标记并强制非流式"""
+        """处理 CLI API 请求。"""
         if self._compact:
-            body = await context.ensure_json_body_async()
-            body["_aether_compact"] = True
-            # compact 端点永远非流式
-            body.pop("stream", None)
-            # 预设 Codex compact 上下文 -- finalize_provider_request 在 envelope
-            # 之前运行，会清除 _aether_compact sentinel，所以在此处提前设置
-            # context var 供 Codex envelope 和 build_codex_url 读取
             from src.services.provider.adapters.codex.context import (
                 CodexRequestContext,
                 set_codex_request_context,
             )
 
+            # Keep compact routing state out of the request body. Transport/policy layers
+            # read this request-scoped flag directly when legacy compact fallback is needed.
             set_codex_request_context(CodexRequestContext(is_compact=True))
+
+            body = await context.ensure_json_body_async()
+            # compact 端点永远非流式
+            body.pop("stream", None)
         return await super().handle(context)
 
     @classmethod
@@ -109,64 +108,17 @@ class OpenAICliAdapter(CliAdapterBase):
         base_url: str | None = None,
         provider_type: str | None = None,
     ) -> dict[str, Any]:
-        """构建测试请求体（Codex 端点需要强制 stream=true 等特性）
+        """构建测试请求体。"""
-
-        provider_type 优先：仅当 provider_type 为 codex 时才应用 Codex 变体；
-        未传入 provider_type 时回退到 URL 模式匹配（兼容旧调用方）。
-        """
         from src.api.handlers.base.request_builder import build_test_request_body
 
-        is_codex = (
+        del base_url, provider_type
-            (provider_type or "").lower() == ProviderType.CODEX
+        return build_test_request_body(cls.FORMAT_ID, request_data)
-            if provider_type
-            else (bool(base_url) and is_codex_url(base_url))
-        )
-        target_variant = "codex" if is_codex else None
-        return build_test_request_body(
-            cls.FORMAT_ID,
-            request_data,
-            target_variant=target_variant,
-        )
 
     @classmethod
     def get_cli_user_agent(cls) -> str | None:
         """获取OpenAI CLI User-Agent"""
         return config.internal_user_agent_openai_cli
 
-    @classmethod
-    def get_cli_extra_headers(
-        cls, *, base_url: str | None = None, provider_type: str | None = None
-    ) -> dict[str, str]:
-        """
-        获取额外请求头
-
-        对于 Codex OAuth 端点，添加特定头部（缺少可能导致 Cloudflare 拦截）。
-        对于标准 OpenAI API 端点，仅添加 User-Agent。
-
-        provider_type 优先：仅当 provider_type 为 codex 时才添加 Codex 头部；
-        未传入 provider_type 时回退到 URL 模式匹配（兼容旧调用方）。
-        """
-        headers: dict[str, str] = {}
-
-        # User-Agent
-        cli_user_agent = cls.get_cli_user_agent()
-        if cli_user_agent:
-            headers["User-Agent"] = cli_user_agent
-
-        # 仅 Codex 端点添加特定头部
-        is_codex = (
-            (provider_type or "").lower() == ProviderType.CODEX
-            if provider_type
-            else (bool(base_url) and is_codex_url(base_url))
-        )
-        if is_codex:
-            # 与运行时路径保持一致：使用 Codex envelope 的 best-effort headers。
-            from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-
-            headers.update(codex_oauth_envelope.extra_headers() or {})
-
-        return headers
-
 
 __all__ = ["OpenAICliAdapter"]
 

src/api/handlers/openai_cli/handler.py

@@ -72,20 +72,6 @@ class OpenAICliMessageHandler(CliMessageHandlerBase):
         result["model"] = mapped_model
         return result
 
-    def finalize_provider_request(
-        self,
-        request_body: dict[str, Any],
-        *,
-        mapped_model: str | None,
-        provider_api_format: str | None,
-    ) -> dict[str, Any]:
-        # Strip internal sentinel before sending upstream (non-Codex providers
-        # don't have an envelope that removes it).
-        request_body.pop("_aether_compact", None)
-        return super().finalize_provider_request(
-            request_body, mapped_model=mapped_model, provider_api_format=provider_api_format
-        )
-
     def _process_event_data(
         self,
         ctx: StreamContext,

src/core/api_format/conversion/normalizers/openai_cli.py

@@ -146,13 +146,9 @@ class OpenAICliNormalizer(FormatNormalizer):
         request: dict[str, Any],
         variant: str,
     ) -> dict[str, Any] | None:
-        """Codex 同格式透传：做最小补丁并保持稳定的请求前缀顺序。"""
+        """OpenAI CLI currently has no provider-specific same-format patching."""
-        if variant.lower() != "codex":
+        del request, variant
-            return None
+        return None
-        out: dict[str, Any] = dict(request)
-        # 内部路由标记：绝不能透传到上游。
-        out.pop("_aether_compact", None)
-        return reorder_openai_cli_request_prefix_keys(out)
 
     def request_to_internal(self, request: dict[str, Any]) -> InternalRequest:
         model = str(request.get("model") or "")
@@ -268,7 +264,6 @@ class OpenAICliNormalizer(FormatNormalizer):
         openai_extra = internal.extra.get("openai", {})
         openai_cli_extra = internal.extra.get("openai_cli", {})
         request_flags = internal.extra.get("openai_cli_request_flags", {}) if internal.extra else {}
-        is_codex_variant = (target_variant or "").lower() == "codex"
         has_explicit_instructions = bool(
             isinstance(request_flags, dict) and request_flags.get("has_instructions")
         )
@@ -414,9 +409,6 @@ class OpenAICliNormalizer(FormatNormalizer):
                 ):
                     result[key] = value
 
-        if is_codex_variant and "store" not in result:
-            result["store"] = False
-
         return self._reorder_request_prefix_keys(result)
 
     # =========================

src/services/provider/adapters/codex/context.py

@@ -1,7 +1,8 @@
-"""Codex request context using contextvars.
+"""Codex request-scoped context.
 
-Similar to Kiro's context pattern, this bridges data from `CodexOAuthEnvelope.wrap_request()`
+Codex only needs a small amount of per-request runtime state that does not belong in
-(which receives the decrypted auth_config) to `extra_headers()` which is parameterless.
+the outbound payload itself. Today that state is the compact-mode flag used by the
+transport and upstream stream-policy layers.
 """
 
 from __future__ import annotations
@@ -12,15 +13,8 @@ from dataclasses import dataclass
 
 @dataclass(frozen=True, slots=True)
 class CodexRequestContext:
-    """Per-request context for the Codex adapter.
+    """Per-request context for the Codex adapter."""
 
-    This bridges data from `CodexOAuthEnvelope.wrap_request()` (which receives the
-    decrypted auth_config) to other layers that only expose parameterless hooks
-    (extra_headers).
-    """
-
-    account_id: str | None = None
-    user_api_key_id: str | None = None
     is_compact: bool = False
 
 
@@ -38,8 +32,24 @@ def get_codex_request_context() -> CodexRequestContext | None:
     return _codex_request_context.get()
 
 
+def is_codex_compact_request(*, endpoint_sig: str | None = None) -> bool:
+    """Return whether the current Codex request should use compact semantics.
+
+    Modern configurations use a dedicated ``openai:compact`` endpoint. Older ones may
+    still route compact traffic through ``openai:cli`` and rely on request-scoped
+    context instead.
+    """
+    normalized_sig = str(endpoint_sig or "").strip().lower()
+    if normalized_sig == "openai:compact":
+        return True
+
+    ctx = get_codex_request_context()
+    return bool(ctx and ctx.is_compact)
+
+
 __all__ = [
     "CodexRequestContext",
     "get_codex_request_context",
+    "is_codex_compact_request",
     "set_codex_request_context",
 ]

src/services/provider/adapters/codex/envelope.py

@@ -1,112 +0,0 @@
-"""Codex upstream envelope hooks.
-
-Codex OAuth upstreams (e.g. `chatgpt.com/backend-api/codex`) behave like the OpenAI
-Responses API (`openai:cli`) but may require additional transport-level headers
-to avoid upstream blocks (Cloudflare, etc.).
-
-Request/response shape quirks should live in the conversion layer as a same-format
-variant (`target_variant="codex"` in the `openai:cli` normalizer). This envelope
-only adds headers and keeps the rest as a no-op wrapper.
-
-We use contextvars to pass request-scoped values (account_id) from wrap_request()
-to extra_headers().
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from src.services.provider.adapters.codex.context import (
-    CodexRequestContext,
-    get_codex_request_context,
-    set_codex_request_context,
-)
-from src.services.provider.adapters.codex.request_patching import patch_openai_cli_request_for_codex
-from src.services.provider.request_context import get_selected_base_url
-
-
-class CodexOAuthEnvelope:
-    """Provider envelope hooks for Codex OAuth upstream."""
-
-    name = "codex:oauth"
-
-    def extra_headers(self) -> dict[str, str] | None:
-        # Codex desktop clients already send the protocol-specific headers they need.
-        # Preserve the original request headers as much as possible and avoid injecting
-        # synthetic CLI identity headers here.
-        return None
-
-    def prepare_context(
-        self,
-        *,
-        provider_config: Any,  # noqa: ARG002
-        key_id: str,  # noqa: ARG002
-        user_api_key_id: str | None = None,
-        is_stream: bool,  # noqa: ARG002
-        provider_id: str | None = None,  # noqa: ARG002
-        key: Any = None,  # noqa: ARG002
-    ) -> str | None:
-        existing_ctx = get_codex_request_context()
-        set_codex_request_context(
-            CodexRequestContext(
-                account_id=existing_ctx.account_id if existing_ctx else None,
-                user_api_key_id=str(user_api_key_id or "").strip() or None,
-                is_compact=existing_ctx.is_compact if existing_ctx else False,
-            )
-        )
-        return None
-
-    def wrap_request(
-        self,
-        request_body: dict[str, Any],
-        *,
-        model: str,  # noqa: ARG002
-        url_model: str | None,
-        decrypted_auth_config: dict[str, Any] | None,
-    ) -> tuple[dict[str, Any], str | None]:
-        # Extract account_id from auth_config and set context for extra_headers()
-        account_id = (decrypted_auth_config or {}).get("account_id")
-        # Compact sentinel may have been popped earlier by finalize_provider_request;
-        # prefer the pre-set context var (set by adapter), fall back to request body.
-        existing_ctx = get_codex_request_context()
-        is_compact = (existing_ctx.is_compact if existing_ctx else False) or bool(
-            request_body.get("_aether_compact", False)
-        )
-        patched_request_body = patch_openai_cli_request_for_codex(request_body)
-        set_codex_request_context(
-            CodexRequestContext(
-                account_id=str(account_id) if account_id else None,
-                user_api_key_id=existing_ctx.user_api_key_id if existing_ctx else None,
-                is_compact=is_compact,
-            )
-        )
-        # Context 不需要手动清理: FastAPI 每个请求运行在独立的 asyncio Task 中,
-        # contextvars 天然隔离, Task 结束后自动回收。
-        # No wire envelope for Codex; keep request body as-is.
-        return patched_request_body, url_model
-
-    def unwrap_response(self, data: Any) -> Any:
-        # No response envelope for Codex.
-        return data
-
-    def postprocess_unwrapped_response(self, *, model: str, data: Any) -> None:  # noqa: ARG002
-        return
-
-    def capture_selected_base_url(self) -> str | None:
-        # Keep interface consistent with Antigravity. Transport currently doesn't set this for Codex.
-        return get_selected_base_url()
-
-    def on_http_status(self, *, base_url: str | None, status_code: int) -> None:  # noqa: ARG002
-        return
-
-    def on_connection_error(self, *, base_url: str | None, exc: Exception) -> None:  # noqa: ARG002
-        return
-
-    def force_stream_rewrite(self) -> bool:
-        return False
-
-
-codex_oauth_envelope = CodexOAuthEnvelope()
-
-
-__all__ = ["CodexOAuthEnvelope", "codex_oauth_envelope"]

src/services/provider/adapters/codex/plugin.py

@@ -1,10 +1,9 @@
 """Codex provider plugin — 统一注册入口。
 
 将 Codex 对各通用 registry / capability registry 的注册集中在一个文件中：
-- Envelope (OAuth headers)
 - Transport Hook (URL 构建)
 - Auth Enricher (OAuth enrichment)
-- Provider Format Capability（格式变体 + 默认 body_rules）
+- Provider Format Capability（默认 body_rules）
 - Model Fetcher (fixed catalog — Codex has no /v1/models endpoint)
 
 新增 provider 时参照此文件创建对应的 plugin.py 即可。
@@ -46,11 +45,10 @@ def build_codex_url(
     """
     _ = is_stream  # Codex 不需要根据 stream 切换路径
 
-    from src.services.provider.adapters.codex.context import get_codex_request_context
-
-    ctx = get_codex_request_context()
     endpoint_sig = str(getattr(endpoint, "api_format", "") or "").strip().lower()
-    is_compact = bool((ctx.is_compact if ctx else False) or endpoint_sig == "openai:compact")
+    from src.services.provider.adapters.codex.context import is_codex_compact_request
+
+    is_compact = is_codex_compact_request(endpoint_sig=endpoint_sig)
 
     base = str(endpoint.base_url).rstrip("/")
     # 如果用户已在 base_url 中包含了 /responses，不要重复追加
@@ -163,21 +161,11 @@ async def enrich_codex(
 
 def register_all() -> None:
     """一次性注册 Codex 的所有 hooks 到各通用 registry。"""
-    from src.core.api_format.capabilities import (
+    from src.core.api_format.capabilities import register_provider_default_body_rules
-        register_provider_behavior_variant,
-        register_provider_default_body_rules,
-    )
     from src.core.provider_oauth_utils import register_auth_enricher
     from src.services.model.upstream_fetcher import UpstreamModelsFetcherRegistry
-    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-    from src.services.provider.envelope import register_envelope
     from src.services.provider.transport import register_transport_hook
 
-    # Envelope
-    register_envelope("codex", "openai:cli", codex_oauth_envelope)
-    register_envelope("codex", "openai:compact", codex_oauth_envelope)
-    register_envelope("codex", "", codex_oauth_envelope)
-
     # Transport
     register_transport_hook("codex", "openai:cli", build_codex_url)
     register_transport_hook("codex", "openai:compact", build_codex_url)
@@ -185,10 +173,9 @@ def register_all() -> None:
     # Auth
     register_auth_enricher("codex", enrich_codex)
 
-    # Provider Format Capability：格式变体 + 默认 body_rules
+    # Provider Format Capability：默认 body_rules
     from src.core.api_format.metadata import CODEX_DEFAULT_BODY_RULES
 
-    register_provider_behavior_variant("codex", same_format=True, cross_format=True)
     register_provider_default_body_rules("codex", "openai:cli", CODEX_DEFAULT_BODY_RULES)
 
     # Export: Codex uses the default export builder (strip null + temp fields)

src/services/provider/adapters/codex/request_patching.py

@@ -1,59 +0,0 @@
-"""Codex provider request patching helpers.
-
-Codex requests are mostly passthrough:
-- Do not mutate client payload fields unless Codex-specific compatibility requires it.
-- Strip internal sentinel fields that must never reach upstream.
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from src.core.api_format.conversion.normalizers.openai_cli import (
-    reorder_openai_cli_request_prefix_keys,
-)
-from src.core.provider_types import ProviderType
-
-
-def patch_openai_cli_request_for_codex(
-    request_body: dict[str, Any],
-) -> dict[str, Any]:
-    """
-    Patch an OpenAI CLI (Responses API style) request body for Codex gateways.
-
-    This function never mutates the input object.
-    """
-    out: dict[str, Any] = dict(request_body)
-    # Internal routing marker; never send upstream.
-    out.pop("_aether_compact", None)
-    # Match the normalizer's stable prefix ordering even on same-format passthrough.
-    return reorder_openai_cli_request_prefix_keys(out)
-
-
-def maybe_patch_request_for_codex(
-    *,
-    provider_type: str | None,
-    provider_api_format: str | None,
-    request_body: Any,
-) -> Any:
-    """
-    Conditionally patch request body for Codex gateways.
-
-    No-op for:
-    - Non-Codex providers
-    - Non OpenAI CLI / Responses-style endpoints
-    - Non-dict request bodies
-    """
-    if (provider_type or "").lower() != ProviderType.CODEX:
-        return request_body
-    if (provider_api_format or "").lower() not in {"openai:cli", "openai:compact"}:
-        return request_body
-    if not isinstance(request_body, dict):
-        return request_body
-    return patch_openai_cli_request_for_codex(request_body)
-
-
-__all__ = [
-    "maybe_patch_request_for_codex",
-    "patch_openai_cli_request_for_codex",
-]

src/services/provider/prompt_cache.py

@@ -8,6 +8,7 @@ from collections.abc import Mapping
 from typing import Any
 
 from src.core.provider_types import ProviderType, normalize_provider_type
+from src.services.provider.adapters.codex.context import is_codex_compact_request
 from src.utils.url_utils import is_official_openai_api_url
 
 _OFFICIAL_OPENAI_PROMPT_CACHE_FORMATS: frozenset[str] = frozenset({"openai:chat", "openai:cli"})
@@ -105,7 +106,6 @@ def build_stable_codex_prompt_cache_key(
 
 def resolve_prompt_cache_key_scope(
     *,
-    request_body: dict[str, Any] | None = None,
     provider_api_format: str | None,
     provider_type: str | None = None,
     base_url: str | None = None,
@@ -115,14 +115,10 @@ def resolve_prompt_cache_key_scope(
     if fmt == "openai:compact":
         return None
 
-    # Belt-and-suspenders: finalize_provider_request 通常已 pop _aether_compact，
-    # 但 openai:compact format 检查在上方已拦截；此处防御非 compact 格式端点意外携带标记。
-    request = request_body if isinstance(request_body, dict) else {}
-    if bool(request.get("_aether_compact", False)):
-        return None
-
     pt = normalize_provider_type(provider_type)
     if pt == ProviderType.CODEX.value and fmt == "openai:cli":
+        if is_codex_compact_request(endpoint_sig=fmt):
+            return None
         return "codex"
 
     if fmt in _OFFICIAL_OPENAI_PROMPT_CACHE_FORMATS and is_official_openai_api_url(base_url):
@@ -145,7 +141,6 @@ def maybe_patch_request_with_prompt_cache_key(
         return request_body
 
     scope = resolve_prompt_cache_key_scope(
-        request_body=request_body,
         provider_api_format=provider_api_format,
         provider_type=provider_type,
         base_url=base_url,

src/services/provider/stream_policy.py

@@ -17,6 +17,7 @@ from typing import Any
 
 from src.core.api_format.metadata import resolve_endpoint_definition
 from src.core.provider_types import ProviderType
+from src.services.provider.adapters.codex.context import is_codex_compact_request
 
 
 class UpstreamStreamPolicy(str, Enum):
@@ -64,13 +65,7 @@ def get_upstream_stream_policy(
     is_codex_cli = pt == ProviderType.CODEX and sig == "openai:cli"
     is_codex_compact = pt == ProviderType.CODEX and sig == "openai:compact"
     if is_codex_cli:
-        try:
+        is_codex_compact = is_codex_compact_request(endpoint_sig=sig)
-            from src.services.provider.adapters.codex.context import get_codex_request_context
-
-            ctx = get_codex_request_context()
-            is_codex_compact = bool(ctx and ctx.is_compact)
-        except Exception:
-            is_codex_compact = False
 
     # Explicit config wins (unless upstream has a hard constraint).
     cfg = getattr(endpoint, "config", None)
@@ -142,16 +137,9 @@ def enforce_stream_mode_for_upstream(
         return request_body
 
     # Backward compatibility: Codex compact routed through openai:cli + context marker.
-    if provider_fmt == "openai:cli":
+    if provider_fmt == "openai:cli" and is_codex_compact_request(endpoint_sig=provider_fmt):
-        try:
+        request_body.pop("stream", None)
-            from src.services.provider.adapters.codex.context import get_codex_request_context
+        return request_body
-
-            ctx = get_codex_request_context()
-            if ctx and ctx.is_compact:
-                request_body.pop("stream", None)
-                return request_body
-        except Exception:
-            pass
 
     if provider_uses_stream:
         request_body["stream"] = bool(upstream_is_stream)

tests/api/handlers/base/test_cli_request_body_isolation.py

@@ -51,7 +51,6 @@ class _DummyCliStreamHandler(CliStreamMixin):
         return out
 
     def prepare_provider_request_body(self, request_body: dict[str, Any]) -> dict[str, Any]:
-        request_body.pop("_aether_compact", None)
         request_body["input"][0]["content"][0]["text"] = "prepared"
         return request_body
 
@@ -116,7 +115,6 @@ async def test_execute_stream_request_does_not_mutate_original_request_body(
 
     original_request_body = {
         "model": "gpt-test",
-        "_aether_compact": True,
         "input": [
             {
                 "role": "user",
@@ -141,6 +139,5 @@ async def test_execute_stream_request_does_not_mutate_original_request_body(
 
     assert original_request_body == snapshot
     assert handler._request_builder.request_body is not None
-    assert "_aether_compact" not in handler._request_builder.request_body
     assert handler._request_builder.request_body["input"][0]["content"][0]["text"] == "prepared"
     assert handler._request_builder.request_body["input"][0]["content"][-1]["text"] == "finalized"

tests/services/test_codex_request.py

@@ -1,144 +1,20 @@
 from __future__ import annotations
 
 from types import SimpleNamespace
-from typing import cast
 
 import jwt
 import pytest
 
 from src.api.handlers.base.request_builder import PassthroughRequestBuilder
-from src.services.provider.adapters.codex.context import set_codex_request_context
+from src.services.provider.behavior import get_provider_behavior
-from src.services.provider.adapters.codex.request_patching import (
-    maybe_patch_request_for_codex,
-    patch_openai_cli_request_for_codex,
-)
-from src.services.provider.envelope import ProviderEnvelope
 
 
-def test_patch_openai_cli_request_for_codex_is_passthrough_except_internal_sentinel() -> None:
+def test_codex_provider_behavior_has_no_runtime_envelope_or_variants() -> None:
-    req = {
+    behavior = get_provider_behavior(provider_type="codex", endpoint_sig="openai:cli")
-        "model": "gpt-test",
-        "input": [
-            {
-                "type": "message",
-                "role": "system",
-                "content": [{"type": "input_text", "text": "Hello"}],
-            }
-        ],
-        "store": True,
-        "stream": False,
-        "instructions": "keep",
-        "include": ["foo"],
-        "parallel_tool_calls": False,
-        "temperature": 0.7,
-        "context_management": {"compaction": {"type": "summary"}},
-        "user": "u_123",
-        "_aether_compact": True,
-    }
-    out = patch_openai_cli_request_for_codex(req)
 
-    assert out is not req
+    assert behavior.envelope is None
-    assert "_aether_compact" not in out
+    assert behavior.same_format_variant is None
-    assert out["store"] is True
+    assert behavior.cross_format_variant is None
-    assert out["stream"] is False
-    assert out["instructions"] == "keep"
-    assert out["include"] == ["foo"]
-    assert out["parallel_tool_calls"] is False
-    assert out["temperature"] == 0.7
-    assert out["context_management"] == {"compaction": {"type": "summary"}}
-    assert out["user"] == "u_123"
-    assert out["input"][0]["role"] == "system"
-
-
-def test_patch_openai_cli_request_for_codex_preserves_existing_prompt_cache_key() -> None:
-    req = {"model": "gpt-test", "input": [], "prompt_cache_key": "client-cache-key"}
-
-    out = patch_openai_cli_request_for_codex(req)
-
-    assert out["prompt_cache_key"] == "client-cache-key"
-
-
-def test_patch_openai_cli_request_for_codex_reorders_stable_prefix_keys() -> None:
-    req = {
-        "temperature": 0.7,
-        "input": [],
-        "metadata": {"request_id": "abc"},
-        "model": "gpt-test",
-        "tools": [{"type": "function", "name": "demo"}],
-        "instructions": "keep",
-        "store": True,
-        "_aether_compact": True,
-    }
-
-    out = patch_openai_cli_request_for_codex(req)
-
-    assert list(out.keys()) == [
-        "model",
-        "instructions",
-        "tools",
-        "input",
-        "temperature",
-        "metadata",
-        "store",
-    ]
-
-
-def test_patch_openai_cli_request_for_codex_does_not_inject_prompt_cache_key() -> None:
-    req = {"model": "gpt-test", "input": [], "_aether_compact": True}
-
-    out = patch_openai_cli_request_for_codex(req)
-
-    assert out is not req
-    assert "_aether_compact" not in out
-    assert "prompt_cache_key" not in out
-
-
-def test_maybe_patch_request_for_codex_is_noop_for_non_codex() -> None:
-    req = {"model": "gpt-test", "input": []}
-    out = maybe_patch_request_for_codex(
-        provider_type="custom",
-        provider_api_format="openai:cli",
-        request_body=req,
-    )
-    assert out is req
-
-
-def test_maybe_patch_request_for_codex_is_noop_for_non_openai_cli() -> None:
-    req = {"model": "gpt-test", "input": []}
-    out = maybe_patch_request_for_codex(
-        provider_type="codex",
-        provider_api_format="openai:chat",
-        request_body=req,
-    )
-    assert out is req
-
-
-def test_maybe_patch_request_for_codex_patches_for_codex_openai_cli() -> None:
-    req = {"model": "gpt-test", "input": [], "_aether_compact": True, "store": True}
-    out = maybe_patch_request_for_codex(
-        provider_type="codex",
-        provider_api_format="openai:cli",
-        request_body=req,
-    )
-
-    assert out is not req
-    assert out["store"] is True
-    assert "_aether_compact" not in out
-    assert "prompt_cache_key" not in out
-
-
-def test_maybe_patch_request_for_codex_patches_for_codex_openai_compact() -> None:
-    req = {"model": "gpt-test", "input": [], "_aether_compact": True, "store": True}
-    out = maybe_patch_request_for_codex(
-        provider_type="codex",
-        provider_api_format="openai:compact",
-        request_body=req,
-    )
-
-    assert out is not req
-    assert out["store"] is True
-    assert "_aether_compact" not in out
-    assert "prompt_cache_key" not in out
 
 
 def test_openai_cli_normalizer_request_from_internal_codex_variant_preserves_store() -> None:
@@ -151,14 +27,14 @@ def test_openai_cli_normalizer_request_from_internal_codex_variant_preserves_sto
     assert out["store"] is True
 
 
-def test_openai_cli_normalizer_request_from_internal_codex_variant_defaults_store_false() -> None:
+def test_openai_cli_normalizer_request_from_internal_codex_variant_does_not_inject_store() -> None:
     from src.core.api_format.conversion.normalizers.openai_cli import OpenAICliNormalizer
 
     normalizer = OpenAICliNormalizer()
     internal = normalizer.request_to_internal({"model": "gpt-test", "input": []})
     out = normalizer.request_from_internal(internal, target_variant="codex")
 
-    assert out["store"] is False
+    assert "store" not in out
 
 
 def test_openai_cli_normalizer_codex_variant_keeps_instructions_missing_for_default_rule() -> None:
@@ -176,7 +52,7 @@ def test_openai_cli_normalizer_codex_variant_keeps_instructions_missing_for_defa
     assert patched["instructions"] == "You are GPT-5."
 
 
-def test_openai_cli_normalizer_patch_for_codex_reorders_stable_prefix_keys() -> None:
+def test_openai_cli_normalizer_patch_for_codex_is_noop() -> None:
     from src.core.api_format.conversion.normalizers.openai_cli import OpenAICliNormalizer
 
     normalizer = OpenAICliNormalizer()
@@ -188,114 +64,14 @@ def test_openai_cli_normalizer_patch_for_codex_reorders_stable_prefix_keys() ->
             "model": "gpt-test",
             "tools": [{"type": "function", "name": "demo"}],
             "instructions": "keep",
-            "_aether_compact": True,
         },
         "codex",
     )
 
-    assert out is not None
+    assert out is None
-    assert list(out.keys()) == [
-        "model",
-        "instructions",
-        "tools",
-        "input",
-        "temperature",
-        "metadata",
-    ]
-
-
-def test_codex_envelope_extra_headers_does_not_inject_synthetic_headers() -> None:
-    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-
-    assert codex_oauth_envelope.extra_headers() is None
-
-
-def test_codex_envelope_wrap_request_injects_stable_prompt_cache_key_from_user_api_key() -> None:
-    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-
-    try:
-        codex_oauth_envelope.prepare_context(
-            provider_config=None,
-            key_id="provider-key-123",
-            user_api_key_id="user-key-123",
-            is_stream=True,
-        )
-        out, url_model = codex_oauth_envelope.wrap_request(
-            {"model": "gpt-test", "input": []},
-            model="gpt-test",
-            url_model=None,
-            decrypted_auth_config=None,
-        )
-    finally:
-        set_codex_request_context(None)
-
-    assert url_model is None
-    assert "prompt_cache_key" not in out
-
-
-def test_codex_envelope_wrap_request_same_user_different_provider_keys_do_not_mutate_prompt_cache_key() -> (
-    None
-):
-    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-
-    try:
-        codex_oauth_envelope.prepare_context(
-            provider_config=None,
-            key_id="provider-key-123",
-            user_api_key_id="user-key-123",
-            is_stream=True,
-        )
-        out_a, _ = codex_oauth_envelope.wrap_request(
-            {"model": "gpt-test", "input": []},
-            model="gpt-test",
-            url_model=None,
-            decrypted_auth_config=None,
-        )
-        codex_oauth_envelope.prepare_context(
-            provider_config=None,
-            key_id="provider-key-456",
-            user_api_key_id="user-key-123",
-            is_stream=True,
-        )
-        out_b, _ = codex_oauth_envelope.wrap_request(
-            {"model": "gpt-test", "input": []},
-            model="gpt-test",
-            url_model=None,
-            decrypted_auth_config=None,
-        )
-    finally:
-        set_codex_request_context(None)
-
-    assert "prompt_cache_key" not in out_a
-    assert "prompt_cache_key" not in out_b
-
-
-def test_codex_envelope_wrap_request_compact_does_not_inject_prompt_cache_key() -> None:
-    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-
-    try:
-        codex_oauth_envelope.prepare_context(
-            provider_config=None,
-            key_id="provider-key-123",
-            user_api_key_id="user-key-123",
-            is_stream=False,
-        )
-        out, _ = codex_oauth_envelope.wrap_request(
-            {"model": "gpt-test", "input": [], "_aether_compact": True},
-            model="gpt-test",
-            url_model=None,
-            decrypted_auth_config=None,
-        )
-    finally:
-        set_codex_request_context(None)
-
-    assert "_aether_compact" not in out
-    assert "prompt_cache_key" not in out
 
 
 def test_codex_passthrough_builder_preserves_real_codex_headers() -> None:
-    from src.services.provider.adapters.codex.envelope import codex_oauth_envelope
-
     builder = PassthroughRequestBuilder()
     endpoint = SimpleNamespace(api_family="openai", endpoint_kind="cli", header_rules=None)
     key = SimpleNamespace(api_key="unused")
@@ -314,7 +90,6 @@ def test_codex_passthrough_builder_preserves_real_codex_headers() -> None:
         endpoint=endpoint,
         key=key,
         pre_computed_auth=("Authorization", "Bearer upstream-token"),
-        envelope=cast(ProviderEnvelope, codex_oauth_envelope),
     )
 
     assert headers["accept"] == "text/event-stream"

tests/services/test_openai_prompt_cache.py

@@ -1,5 +1,9 @@
 from __future__ import annotations
 
+from src.services.provider.adapters.codex.context import (
+    CodexRequestContext,
+    set_codex_request_context,
+)
 from src.services.provider.prompt_cache import (
     build_stable_codex_prompt_cache_key,
     build_stable_openai_prompt_cache_key,
@@ -103,7 +107,7 @@ def test_maybe_patch_request_with_prompt_cache_key_for_codex_openai_cli() -> Non
 
 
 def test_maybe_patch_request_with_prompt_cache_key_skips_official_compact() -> None:
-    req = {"model": "gpt-5", "input": [], "_aether_compact": True}
+    req = {"model": "gpt-5", "input": []}
 
     out = maybe_patch_request_with_prompt_cache_key(
         req,
@@ -117,16 +121,20 @@ def test_maybe_patch_request_with_prompt_cache_key_skips_official_compact() -> N
     assert "prompt_cache_key" not in out
 
 
-def test_maybe_patch_request_with_prompt_cache_key_skips_codex_compact_marker() -> None:
+def test_maybe_patch_request_with_prompt_cache_key_skips_legacy_codex_compact_context() -> None:
-    req = {"model": "gpt-5", "input": [], "_aether_compact": True}
+    req = {"model": "gpt-5", "input": []}
 
-    out = maybe_patch_request_with_prompt_cache_key(
+    try:
-        req,
+        set_codex_request_context(CodexRequestContext(is_compact=True))
-        provider_api_format="openai:cli",
+        out = maybe_patch_request_with_prompt_cache_key(
-        provider_type="codex",
+            req,
-        base_url="https://chatgpt.com/backend-api/codex",
+            provider_api_format="openai:cli",
-        user_api_key_id="user-key-123",
+            provider_type="codex",
-    )
+            base_url="https://chatgpt.com/backend-api/codex",
+            user_api_key_id="user-key-123",
+        )
+    finally:
+        set_codex_request_context(None)
 
     assert out is req
     assert "prompt_cache_key" not in out