fix(models): close Codex async HTTP requests on cancellation (#8053)

* fix(models): close Codex async HTTP requests on cancellation

* test(models): isolate loopback cancellation checks from proxies

* docs: record Codex cancellation fix in changelog

* docs: translate Codex cancellation changelog entry

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
Co-authored-by: Claude Code <noreply@anthropic.com>

CHANGELOG.md

@@ -203,6 +203,11 @@ This section accumulates work toward the **2.3.0** milestone
   mirrors `ThreadOwnershipConflictError`; a duplicate key with no surviving row
   is still surfaced as a 500. ([#8057])
 
+- **models:** Cancelling a Codex-backed run now closes its upstream HTTP/SSE
+  connection, including while waiting for response headers, and retry waits
+  are cancellable. Local resources are released; the remote service may still
+  continue generation. ([#8053](https://github.com/bytedance/deer-flow/pull/8053))
+
 - **middleware:** With PII redaction enabled, a long run of text without
   spaces (such as a pasted hex dump) no longer stalls the Gateway. The email
   detector was quadratic in the length of such a run, and redaction re-scans

CHANGELOG_zh.md

@@ -69,6 +69,9 @@
   `/goal` 读取残留的标题和消息，所有者也无法重试。现在删除会返回 500 并保留线程，所有者
   可以再次删除。([#7992])
 
+- **模型：** 取消使用 Codex 的任务时，会关闭上游 HTTP/SSE 连接，包括等待响应头期间；
+  重试等待也可取消。本地资源会释放，远端服务仍可能继续生成。([#8053](https://github.com/bytedance/deer-flow/pull/8053))
+
 - **中间件：** 启用 PII 脱敏时，不含空格的长文本（例如粘贴的十六进制转储）不再让 Gateway
   卡住。此前邮箱检测器的耗时与这类文本长度呈平方关系，而脱敏会在每次模型调用时重新扫描全部
   用户消息，因此一次 100 KB 的粘贴会让每次模型调用阻塞所有线程约 6 秒。现在扫描为线性耗时，

backend/packages/harness/deerflow/models/AGENTS.md

@@ -59,6 +59,11 @@ transport, so they check the payload that DeerFlow sends.
 - **A completed response with null or empty usage keeps its text.**
 - **A terminal stream error returns no partial output.** The caller must not take
   a truncated answer for a finished one.
+- **Cancelling a Codex async call closes its HTTP stream.** Async generation owns
+  an HTTPX async stream, not an executor-backed sync request. Load TLS trust
+  off-loop, keep retry sleeps cancellable, and drain the client close across
+  repeated cancellation. `test_codex_cancellation.py` observes real loopback EOF
+  before fixture release.
 - **MindIE requests `stream=False` and simulates streaming**, and its XML tool
   calls and numeric arguments are parsed in the adapter.
 - **Claude prompt caching copies before it marks.** The request payload shares

backend/packages/harness/deerflow/models/openai_codex_provider.py

@@ -11,18 +11,20 @@ Supports:
 - Retry with exponential backoff
 """
 
+import asyncio
 import json
 import logging
 import time
 from typing import Any
 
 import httpx
-from langchain_core.callbacks import CallbackManagerForLLMRun
+from langchain_core.callbacks import AsyncCallbackManagerForLLMRun, CallbackManagerForLLMRun
 from langchain_core.language_models.chat_models import BaseChatModel
 from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage, ToolMessage
 from langchain_core.outputs import ChatGeneration, ChatResult
 
 from deerflow.models.credential_loader import CodexCliCredential, load_codex_cli_credential
+from deerflow.utils.file_io import await_drained
 from deerflow.utils.retry_after import bounded_retry_after_ms, parse_retry_after_date_ms
 
 logger = logging.getLogger(__name__)
@@ -125,6 +127,61 @@ def _build_usage_metadata(oai_usage: dict) -> dict:
 MAX_RETRIES = 3
 
 
+class _CodexResponseCollector:
+    """Shared terminal-event and output recovery semantics for both transports."""
+
+    def __init__(self) -> None:
+        self.completed_response: dict | None = None
+        self.output_items: dict[int, dict[str, Any]] = {}
+
+    def feed(self, data: dict[str, Any]) -> bool:
+        """Return True at completion, before requesting another transport read."""
+        event_type = data.get("type")
+        if event_type == "response.output_item.done":
+            output_index = data.get("output_index")
+            output_item = data.get("item")
+            if isinstance(output_index, int) and isinstance(output_item, dict):
+                self.output_items[output_index] = output_item
+        elif event_type == "response.completed":
+            self.completed_response = data["response"]
+            return True
+        elif event_type in ("response.failed", "response.incomplete", "error"):
+            response = data.get("response") or {}
+            if event_type == "error":
+                details = data.get("error") or data
+            elif not isinstance(response, dict):
+                details = response
+            elif event_type == "response.failed":
+                details = response.get("error") or {}
+            else:
+                details = response.get("incomplete_details") or {}
+            if not isinstance(details, dict):
+                details = {"message": str(details)}
+            code = details.get("code")
+            reason = details.get("message") or details.get("reason") or "No details provided"
+            code_suffix = f" ({code})" if code else ""
+            raise RuntimeError(f"Codex API {event_type}{code_suffix}: {reason}")
+        return False
+
+    def finish(self) -> dict:
+        if not self.completed_response:
+            raise RuntimeError("Codex API stream ended without response.completed event")
+        completed = self.completed_response
+        # The completed response can omit content already emitted as done items.
+        if self.output_items:
+            response_output = completed.get("output")
+            merged = list(response_output) if isinstance(response_output, list) else []
+            max_index = max(max(self.output_items), len(merged) - 1)
+            if len(merged) <= max_index:
+                merged.extend([None] * (max_index + 1 - len(merged)))
+            for index, item in self.output_items.items():
+                if not isinstance(merged[index], dict):
+                    merged[index] = item
+            completed = dict(completed)
+            completed["output"] = [item for item in merged if isinstance(item, dict)]
+        return completed
+
+
 class CodexChatModel(BaseChatModel):
     """LangChain chat model using ChatGPT Codex Responses API.
 
@@ -296,8 +353,8 @@ class CodexChatModel(BaseChatModel):
             responses_tools.append(converted)
         return responses_tools
 
-    def _call_codex_api(self, messages: list[BaseMessage], tools: list[dict] | None = None, tool_choice: str | dict | bool | None = None) -> dict:
-        """Call the Codex Responses API and return the completed response."""
+    def _prepare_request(self, messages: list[BaseMessage], tools: list[dict] | None, tool_choice: str | dict | bool | None) -> tuple[dict, dict]:
+        """Build the same validated request for synchronous and async callers."""
         instructions, input_items = self._convert_messages(messages)
 
         payload = {
@@ -324,6 +381,11 @@ class CodexChatModel(BaseChatModel):
             "Accept": "text/event-stream",
             "originator": "codex_cli_rs",
         }
+        return headers, payload
+
+    def _call_codex_api(self, messages: list[BaseMessage], tools: list[dict] | None = None, tool_choice: str | dict | bool | None = None) -> dict:
+        """Call the Codex Responses API and return the completed response."""
+        headers, payload = self._prepare_request(messages, tools, tool_choice)
 
         last_error = None
         for attempt in range(1, self.retry_max_attempts + 1):
@@ -344,73 +406,49 @@ class CodexChatModel(BaseChatModel):
 
         raise last_error
 
+    async def _acall_codex_api(self, messages: list[BaseMessage], tools: list[dict] | None = None, tool_choice: str | dict | bool | None = None) -> dict:
+        """Own the HTTP operation in the caller task, including retry waits."""
+        headers, payload = self._prepare_request(messages, tools, tool_choice)
+        for attempt in range(1, self.retry_max_attempts + 1):
+            try:
+                return await self._astream_response(headers, payload)
+            except httpx.HTTPStatusError as error:
+                if error.response.status_code not in (429, 500, 529) or attempt >= self.retry_max_attempts:
+                    raise
+                wait_ms = _calc_backoff_ms(attempt, error)
+                logger.warning(f"Codex API error {error.response.status_code}, retrying {attempt}/{self.retry_max_attempts} after {wait_ms}ms")
+                await asyncio.sleep(wait_ms / 1000)
+        raise RuntimeError("Codex API retry attempts exhausted")
+
     def _stream_response(self, headers: dict, payload: dict) -> dict:
         """Stream SSE from Codex API and collect the final response."""
-        completed_response = None
-        streamed_output_items: dict[int, dict[str, Any]] = {}
-
+        collector = _CodexResponseCollector()
         with httpx.Client(timeout=300) as client:
             with client.stream("POST", f"{CODEX_BASE_URL}/responses", headers=headers, json=payload) as resp:
                 resp.raise_for_status()
                 for line in resp.iter_lines():
                     data = self._parse_sse_data_line(line)
-                    if not data:
-                        continue
-
-                    event_type = data.get("type")
-                    if event_type == "response.output_item.done":
-                        output_index = data.get("output_index")
-                        output_item = data.get("item")
-                        if isinstance(output_index, int) and isinstance(output_item, dict):
-                            streamed_output_items[output_index] = output_item
-                    elif event_type == "response.completed":
-                        completed_response = data["response"]
-                        # A terminal event completes the request even if the server
-                        # keeps the connection open. Do not let a later read timeout
-                        # replace the completed output with a transport error.
+                    if data and collector.feed(data):
                         break
-                    elif event_type in ("response.failed", "response.incomplete", "error"):
-                        response = data.get("response") or {}
-                        if event_type == "error":
-                            details = data.get("error") or data
-                        elif not isinstance(response, dict):
-                            details = response
-                        elif event_type == "response.failed":
-                            details = response.get("error") or {}
-                        else:
-                            details = response.get("incomplete_details") or {}
+        return collector.finish()
 
-                        if not isinstance(details, dict):
-                            details = {"message": str(details)}
-                        code = details.get("code")
-                        reason = details.get("message") or details.get("reason") or "No details provided"
-                        code_suffix = f" ({code})" if code else ""
-                        raise RuntimeError(f"Codex API {event_type}{code_suffix}: {reason}")
-
-        if not completed_response:
-            raise RuntimeError("Codex API stream ended without response.completed event")
-
-        # ChatGPT Codex can emit the final assistant content only in stream events.
-        # When response.completed arrives, response.output may still be empty.
-        if streamed_output_items:
-            merged_output = []
-            response_output = completed_response.get("output")
-            if isinstance(response_output, list):
-                merged_output = list(response_output)
-
-            max_index = max(max(streamed_output_items), len(merged_output) - 1)
-            if max_index >= 0 and len(merged_output) <= max_index:
-                merged_output.extend([None] * (max_index + 1 - len(merged_output)))
-
-            for output_index, output_item in streamed_output_items.items():
-                existing_item = merged_output[output_index]
-                if not isinstance(existing_item, dict):
-                    merged_output[output_index] = output_item
-
-            completed_response = dict(completed_response)
-            completed_response["output"] = [item for item in merged_output if isinstance(item, dict)]
-
-        return completed_response
+    async def _astream_response(self, headers: dict, payload: dict) -> dict:
+        """Consume SSE asynchronously so cancellation closes the owned stream."""
+        # Loading the default TLS trust store performs filesystem I/O. Do it
+        # before constructing a client: cancellation here owns no HTTP handles.
+        verify = await asyncio.to_thread(httpx.create_ssl_context)
+        collector = _CodexResponseCollector()
+        client = httpx.AsyncClient(timeout=300, verify=verify)
+        try:
+            async with client.stream("POST", f"{CODEX_BASE_URL}/responses", headers=headers, json=payload) as resp:
+                resp.raise_for_status()
+                async for line in resp.aiter_lines():
+                    data = self._parse_sse_data_line(line)
+                    if data and collector.feed(data):
+                        break
+        finally:
+            await await_drained(client.aclose())
+        return collector.finish()
 
     @staticmethod
     def _parse_sse_data_line(line: str) -> dict[str, Any] | None:
@@ -535,6 +573,17 @@ class CodexChatModel(BaseChatModel):
         response = self._call_codex_api(messages, tools=tools, tool_choice=kwargs.get("tool_choice"))
         return self._parse_response(response)
 
+    async def _agenerate(
+        self,
+        messages: list[BaseMessage],
+        stop: list[str] | None = None,
+        run_manager: AsyncCallbackManagerForLLMRun | None = None,
+        **kwargs: Any,
+    ) -> ChatResult:
+        """Generate without leaving synchronous HTTP work in an executor."""
+        response = await self._acall_codex_api(messages, tools=kwargs.get("tools"), tool_choice=kwargs.get("tool_choice"))
+        return self._parse_response(response)
+
     def bind_tools(self, tools: list, *, tool_choice: str | dict | bool | None = None, **kwargs: Any) -> Any:
         """Bind function tools and their selection policy."""
         from langchain_core.runnables import RunnableBinding

backend/tests/blocking_io/test_codex_async_http.py

@@ -0,0 +1,32 @@
+"""Codex async requests load TLS trust outside the event loop."""
+
+import json
+
+import httpx
+import pytest
+
+from deerflow.models import openai_codex_provider as provider
+from deerflow.models.credential_loader import CodexCliCredential
+
+
+@pytest.mark.asyncio
+async def test_codex_async_http_with_real_tls_setup(monkeypatch):
+    monkeypatch.setattr(provider, "load_codex_cli_credential", lambda: CodexCliCredential("synthetic-token", "synthetic-account"))
+    model = provider.CodexChatModel(retry_max_attempts=1)
+    client_class = httpx.AsyncClient
+    clients = []
+
+    def respond(request):
+        event = {"type": "response.completed", "response": {"output": [{"type": "message", "content": [{"type": "output_text", "text": "OK"}]}]}}
+        return httpx.Response(200, text=f"data: {json.dumps(event)}\n\n")
+
+    def client(**kwargs):
+        instance = client_class(transport=httpx.MockTransport(respond), **kwargs)
+        clients.append(instance)
+        return instance
+
+    monkeypatch.setattr(provider.httpx, "AsyncClient", client)
+    result = await model.ainvoke("Synthetic nonblocking request")
+    assert result.content == "OK"
+    assert len(clients) == 1
+    assert clients[0].is_closed

backend/tests/test_codex_cancellation.py

@@ -0,0 +1,205 @@
+"""Exercise production Codex HTTP ownership with a real loopback SSE peer."""
+
+import asyncio
+import json
+import threading
+
+import httpx
+import pytest
+from langchain_core.messages import HumanMessage
+
+from deerflow.models import openai_codex_provider as provider
+from deerflow.models.credential_loader import CodexCliCredential
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("phase", ["headers", "body"])
+@pytest.mark.parametrize("api", ["ainvoke", "astream"])
+async def test_cancelled_ainvoke_closes_upstream_connection(monkeypatch, phase, api):
+    monkeypatch.setenv("NO_PROXY", "127.0.0.1,localhost")
+    monkeypatch.setenv("no_proxy", "127.0.0.1,localhost")
+    started = asyncio.Event()
+    disconnected = asyncio.Event()
+    release = asyncio.Event()
+    handlers = set()
+
+    async def handle(reader, writer):
+        task = asyncio.current_task()
+        handlers.add(task)
+        try:
+            headers = await reader.readuntil(b"\r\n\r\n")
+            size = next(int(line.split(b":", 1)[1]) for line in headers.split(b"\r\n") if line.lower().startswith(b"content-length:"))
+            await reader.readexactly(size)
+            if phase == "body":
+                writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n")
+                writer.write(b'data: {"type":"response.created","response":{"id":"synthetic"}}\n\n')
+                await writer.drain()
+            started.set()
+            # The fixture never completes the model response. Only client EOF
+            # can satisfy the assertion, and teardown always releases the peer.
+            eof = asyncio.create_task(reader.read())
+            done = asyncio.create_task(release.wait())
+            try:
+                finished, _ = await asyncio.wait({eof, done}, return_when=asyncio.FIRST_COMPLETED)
+                if eof in finished and eof.result() == b"":
+                    disconnected.set()
+            finally:
+                eof.cancel()
+                done.cancel()
+                await asyncio.gather(eof, done, return_exceptions=True)
+        finally:
+            writer.close()
+            await writer.wait_closed()
+            handlers.discard(task)
+
+    server = await asyncio.start_server(handle, "127.0.0.1", 0)
+    monkeypatch.setattr(provider, "CODEX_BASE_URL", f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}")
+    monkeypatch.setattr(provider, "load_codex_cli_credential", lambda: CodexCliCredential("synthetic-token", "synthetic-account"))
+    model = provider.CodexChatModel(retry_max_attempts=1)
+
+    async def invoke():
+        messages = [HumanMessage(content="Synthetic cancellation test")]
+        if api == "ainvoke":
+            await model.ainvoke(messages)
+        else:
+            async for _ in model.astream(messages):
+                pass
+
+    task = asyncio.create_task(invoke())
+    try:
+        await asyncio.wait_for(started.wait(), 5)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        await asyncio.wait_for(disconnected.wait(), 2)
+    finally:
+        release.set()
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+        server.close()
+        await server.wait_closed()
+        if handlers:
+            await asyncio.wait_for(asyncio.gather(*handlers), 5)
+
+
+@pytest.fixture
+def model(monkeypatch):
+    monkeypatch.setattr(provider, "load_codex_cli_credential", lambda: CodexCliCredential("synthetic-token", "synthetic-account"))
+    return provider.CodexChatModel(retry_max_attempts=3)
+
+
+@pytest.mark.asyncio
+async def test_cancellation_during_retry_closes_response_and_does_not_retry(monkeypatch, model):
+    waiting = asyncio.Event()
+    calls = []
+    responses = []
+    clients = []
+    client_class = httpx.AsyncClient
+
+    def respond(request):
+        calls.append(request)
+        response = httpx.Response(429, headers={"Retry-After": "600"}, json={"error": "synthetic"})
+        responses.append(response)
+        return response
+
+    def create_client(**kwargs):
+        client = client_class(transport=httpx.MockTransport(respond), **kwargs)
+        clients.append(client)
+        return client
+
+    async def sleep(delay):
+        assert delay == 600
+        assert all(response.is_closed for response in responses)
+        assert all(client.is_closed for client in clients)
+        waiting.set()
+        await asyncio.Event().wait()
+
+    monkeypatch.setattr(provider.httpx, "AsyncClient", create_client)
+    monkeypatch.setattr(provider.asyncio, "sleep", sleep)
+    task = asyncio.create_task(model.ainvoke("Synthetic retry cancellation"))
+    try:
+        await asyncio.wait_for(waiting.wait(), 5)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert len(calls) == 1
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+
+@pytest.mark.asyncio
+async def test_tls_setup_is_off_loop_and_cancellation_creates_no_client(monkeypatch, model):
+    loop = asyncio.get_running_loop()
+    started = asyncio.Event()
+    finished = asyncio.Event()
+    release = threading.Event()
+    loop_thread = threading.get_ident()
+    real_context = httpx.create_ssl_context
+
+    def context():
+        assert threading.get_ident() != loop_thread
+        loop.call_soon_threadsafe(started.set)
+        try:
+            assert release.wait(5)
+            return real_context()
+        finally:
+            loop.call_soon_threadsafe(finished.set)
+
+    def unexpected_client(**kwargs):
+        pytest.fail("Cancellation before TLS setup finishes must not create an HTTP client")
+
+    monkeypatch.setattr(provider.httpx, "create_ssl_context", context)
+    monkeypatch.setattr(provider.httpx, "AsyncClient", unexpected_client)
+    task = asyncio.create_task(model.ainvoke("Synthetic TLS cancellation"))
+    try:
+        await asyncio.wait_for(started.wait(), 5)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    finally:
+        release.set()
+        await asyncio.gather(task, return_exceptions=True)
+        await asyncio.wait_for(finished.wait(), 5)
+
+
+@pytest.mark.asyncio
+async def test_repeated_cancellation_drains_client_close(monkeypatch, model):
+    closing = asyncio.Event()
+    release = asyncio.Event()
+    clients = []
+    client_class = httpx.AsyncClient
+
+    def respond(request):
+        event = {"type": "response.completed", "response": {"output": []}}
+        return httpx.Response(200, text=f"data: {json.dumps(event)}\n\n")
+
+    class ClosingClient(client_class):
+        async def aclose(self):
+            closing.set()
+            await release.wait()
+            await super().aclose()
+
+    def create_client(**kwargs):
+        client = ClosingClient(transport=httpx.MockTransport(respond), **kwargs)
+        clients.append(client)
+        return client
+
+    monkeypatch.setattr(provider.httpx, "AsyncClient", create_client)
+    task = asyncio.create_task(model.ainvoke("Synthetic cleanup cancellation"))
+    try:
+        await asyncio.wait_for(closing.wait(), 5)
+        task.cancel()
+        await asyncio.sleep(0)
+        task.cancel()
+        await asyncio.sleep(0)
+        assert not task.done()
+        release.set()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        assert len(clients) == 1
+        assert clients[0].is_closed
+    finally:
+        release.set()
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)

backend/tests/test_codex_retry_after.py

@@ -181,10 +181,16 @@ async def test_public_invocation_waits_for_http_date_retry_after(monkeypatch, fi
         }
         return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=f"data: {json.dumps(event)}\n\n")
 
-    client_class = httpx.Client
-    monkeypatch.setattr(provider.httpx, "Client", lambda **kwargs: client_class(transport=httpx.MockTransport(respond), trust_env=False, **kwargs))
+    client_name = "AsyncClient" if mode == "async" else "Client"
+    client_class = getattr(httpx, client_name)
+    monkeypatch.setattr(provider.httpx, client_name, lambda **kwargs: client_class(transport=httpx.MockTransport(respond), trust_env=False, **kwargs))
     monkeypatch.setattr(provider.time, "sleep", sleeps.append)
 
+    async def async_sleep(delay):
+        sleeps.append(delay)
+
+    monkeypatch.setattr(provider.asyncio, "sleep", async_sleep)
+
     result = model.invoke("hi") if mode == "sync" else await model.ainvoke("hi")
 
     assert result.content == "recovered"

backend/tests/test_codex_stream_terminal_events.py

@@ -13,7 +13,7 @@ from deerflow.models import openai_codex_provider as provider
 from deerflow.models.credential_loader import CodexCliCredential
 
 
-class EventStream(httpx.SyncByteStream):
+class EventStream(httpx.SyncByteStream, httpx.AsyncByteStream):
     def __init__(self, events: list[dict], tail_error: Exception | None = None):
         self.events = events
         self.tail_error = tail_error
@@ -30,6 +30,13 @@ class EventStream(httpx.SyncByteStream):
     def close(self) -> None:
         self.closed = True
 
+    async def __aiter__(self):
+        for chunk in self:
+            yield chunk
+
+    async def aclose(self) -> None:
+        self.close()
+
 
 @pytest.fixture
 def model(monkeypatch):
@@ -40,6 +47,7 @@ def model(monkeypatch):
 @pytest.fixture
 def serve_events(monkeypatch):
     client_class = httpx.Client
+    async_client_class = httpx.AsyncClient
 
     def install(events: list[dict], *, tail_error: Exception | None = None, status_code: int = 200):
         stream = EventStream(events, tail_error)
@@ -51,6 +59,7 @@ def serve_events(monkeypatch):
 
         transport = httpx.MockTransport(handle)
         monkeypatch.setattr(provider.httpx, "Client", lambda **kwargs: client_class(transport=transport, **kwargs))
+        monkeypatch.setattr(provider.httpx, "AsyncClient", lambda **kwargs: async_client_class(transport=transport, **kwargs))
         return stream, requests
 
     return install

backend/tests/test_codex_tool_choice.py

@@ -42,7 +42,8 @@ def test_tool_choice_survives_request_and_tool_followup(monkeypatch, tool_choice
     requests = []
     clients = []
     responses = []
-    client_class = httpx.Client
+    client_name = "AsyncClient" if invocation.endswith("ainvoke") else "Client"
+    client_class = getattr(httpx, client_name)
 
     def handle(request):
         payload = json.loads(request.content)
@@ -61,7 +62,7 @@ def test_tool_choice_survives_request_and_tool_followup(monkeypatch, tool_choice
         clients.append(client)
         return client
 
-    monkeypatch.setattr(provider.httpx, "Client", make_client)
+    monkeypatch.setattr(provider.httpx, client_name, make_client)
     history = [HumanMessage(content="Read the quarterly report.")]
 
     if invocation == "invoke":
@@ -175,14 +176,15 @@ def test_no_function_tools_preserves_nonforcing_choices(monkeypatch, tools, tool
     monkeypatch.setattr(provider, "load_codex_cli_credential", lambda: CodexCliCredential("fake-token", "fake-account"))
     model = provider.CodexChatModel(retry_max_attempts=1)
     payloads = []
-    client_class = httpx.Client
+    client_name = "AsyncClient" if invocation.endswith("ainvoke") else "Client"
+    client_class = getattr(httpx, client_name)
 
     def handle(request):
         payloads.append(json.loads(request.content))
         event = {"type": "response.completed", "response": {"output": [{"type": "message", "content": [{"type": "output_text", "text": "OK"}]}]}}
         return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=f"data: {json.dumps(event)}\n\n".encode())
 
-    monkeypatch.setattr(provider.httpx, "Client", lambda **options: client_class(transport=httpx.MockTransport(handle), **options))
+    monkeypatch.setattr(provider.httpx, client_name, lambda **options: client_class(transport=httpx.MockTransport(handle), **options))
     runnable = model.bind_tools(tools or [], tool_choice=tool_choice) if invocation.startswith("bound_") else model
     kwargs = {} if invocation.startswith("bound_") else {"tools": tools, "tool_choice": tool_choice}
     answer = asyncio.run(runnable.ainvoke([HumanMessage(content="Say OK.")], **kwargs)) if invocation.endswith("ainvoke") else runnable.invoke([HumanMessage(content="Say OK.")], **kwargs)

backend/tests/test_codex_tool_strict.py

@@ -67,7 +67,8 @@ def test_function_strictness_survives_tool_followup(monkeypatch, shape, strict,
     requests = []
     clients = []
     responses = []
-    client_class = httpx.Client
+    client_name = "AsyncClient" if invocation == "ainvoke" else "Client"
+    client_class = getattr(httpx, client_name)
 
     def handle(request):
         payload = json.loads(request.content)
@@ -86,7 +87,7 @@ def test_function_strictness_survives_tool_followup(monkeypatch, shape, strict,
         clients.append(client)
         return client
 
-    monkeypatch.setattr(provider.httpx, "Client", make_client)
+    monkeypatch.setattr(provider.httpx, client_name, make_client)
     history = [HumanMessage(content="Read the quarterly report.")]
 
     if invocation == "invoke":

backend/tests/test_model_request_admission.py

@@ -402,10 +402,16 @@ async def test_codex_factory_retry_policy_respects_admission(monkeypatch, regist
         event = {"type": "response.completed", "response": {"output": [{"type": "message", "content": [{"type": "output_text", "text": "unexpected retry"}]}], "usage": {}}}
         return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=f"data: {json.dumps(event)}\n\n")
 
-    client_class = httpx.Client
-    monkeypatch.setattr(provider.httpx, "Client", lambda **kwargs: client_class(transport=httpx.MockTransport(respond), **kwargs))
+    client_name = "AsyncClient" if mode == "async" else "Client"
+    client_class = getattr(httpx, client_name)
+    monkeypatch.setattr(provider.httpx, client_name, lambda **kwargs: client_class(transport=httpx.MockTransport(respond), **kwargs))
     monkeypatch.setattr(provider.time, "sleep", sleeps.append)
 
+    async def async_sleep(delay):
+        sleeps.append(delay)
+
+    monkeypatch.setattr(asyncio, "sleep", async_sleep)
+
     if enabled:
         with pytest.raises(httpx.HTTPStatusError) as caught:
             if mode == "sync":
@@ -484,8 +490,9 @@ async def test_codex_529_middleware_retry_reenters_admission(monkeypatch, regist
         sleep(delay)
         await real_async_sleep(0)
 
-    client_class = httpx.Client
-    monkeypatch.setattr(provider.httpx, "Client", lambda **kwargs: client_class(transport=httpx.MockTransport(respond), **kwargs))
+    client_name = "AsyncClient" if mode == "async" else "Client"
+    client_class = getattr(httpx, client_name)
+    monkeypatch.setattr(provider.httpx, client_name, lambda **kwargs: client_class(transport=httpx.MockTransport(respond), **kwargs))
     monkeypatch.setattr(provider.time, "sleep", sleep)
     monkeypatch.setattr(asyncio, "sleep", async_sleep)
     agent = create_agent(model=model, tools=[], middleware=[errors.LLMErrorHandlingMiddleware(app_config=config)])

docs/guide/configuration.md

@@ -381,6 +381,7 @@ fences, including unclosed fences, in complete and streamed replies.
    `ClaudeChatModel` normalizes manual extended-thinking budgets when `auto_thinking_budget: true` (the default). An omitted or null budget gets 80% of `max_tokens`, with a minimum of 1024 tokens. Explicit budgets must be integers of at least 1024 and, for ordinary thinking, strictly below `max_tokens`; output limits of 1024 or less fail locally. For supported manual interleaved thinking, configure tools and `betas: ["interleaved-thinking-2025-05-14"]`: the thinking budget may equal or exceed the positive output limit. See [Anthropic's interleaved-thinking rules](https://platform.claude.com/docs/en/build-with-claude/extended-thinking#interleaved-thinking-in-manual-mode) for supported models. `auto_thinking_budget: false` bypasses this normalization and validation; disabled and adaptive thinking are unchanged.
 
    - Codex CLI reads `~/.codex/auth.json`
+   - Codex async model calls close their HTTP connection when cancelled, including while waiting for a response; retry waits are cancellable too. This releases local resources and does not guarantee that the remote service stops generation or billing.
    - Codex function tools preserve explicit `strict: true` or `strict: false` in wrapped or flat dictionary definitions. Missing or null settings keep the provider default. `bind_tools` applies the same conversion to dictionaries and `BaseTool` schemas.
    - Codex honors `tool_choice` in `bind_tools` and direct calls: `auto`, `none`, `required`, or a function name/selection. `any` and `True` mean `required`; `None` and `False` keep the provider default. Per-call choices override the bound policy. Bound choices are validated when binding; direct calls and overrides are validated before HTTP. `required` needs at least one function tool, and named selections must match a supplied function. `auto` and `none` also work without tools.
    - Completed Codex responses still return their text and tool calls when token usage is null, omitted, or empty; usage metadata remains unavailable.