From b16e19761810092cde3004b97a41a78a91642e03 Mon Sep 17 00:00:00 2001 From: Wang guanyu <102344225+wang1299@users.noreply.github.com> Date: Thu, 12 Nov 2026 21:55:08 +0800 Subject: [PATCH] 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 Co-authored-by: Claude Code --- CHANGELOG.md | 5 + CHANGELOG_zh.md | 3 + .../harness/deerflow/models/AGENTS.md | 5 + .../deerflow/models/openai_codex_provider.py | 171 +++++++++------ .../blocking_io/test_codex_async_http.py | 32 +++ backend/tests/test_codex_cancellation.py | 205 ++++++++++++++++++ backend/tests/test_codex_retry_after.py | 10 +- .../test_codex_stream_terminal_events.py | 11 +- backend/tests/test_codex_tool_choice.py | 10 +- backend/tests/test_codex_tool_strict.py | 5 +- backend/tests/test_model_request_admission.py | 15 +- docs/guide/configuration.md | 1 + 12 files changed, 399 insertions(+), 74 deletions(-) create mode 100644 backend/tests/blocking_io/test_codex_async_http.py create mode 100644 backend/tests/test_codex_cancellation.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9c4da50c77..8d749decdd 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 205878d99c..f4d03d46be 100644 --- a/CHANGELOG_zh.md +++ b/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 秒。现在扫描为线性耗时， diff --git a/backend/packages/harness/deerflow/models/AGENTS.md b/backend/packages/harness/deerflow/models/AGENTS.md index 9b1073d325..51ca8bec58 100644 --- a/backend/packages/harness/deerflow/models/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/models/openai_codex_provider.py b/backend/packages/harness/deerflow/models/openai_codex_provider.py index 952bc6d553..7a0ba9e6ba 100644 --- a/backend/packages/harness/deerflow/models/openai_codex_provider.py +++ b/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 diff --git a/backend/tests/blocking_io/test_codex_async_http.py b/backend/tests/blocking_io/test_codex_async_http.py new file mode 100644 index 0000000000..0c27b46276 --- /dev/null +++ b/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 diff --git a/backend/tests/test_codex_cancellation.py b/backend/tests/test_codex_cancellation.py new file mode 100644 index 0000000000..1965ecae9f --- /dev/null +++ b/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) diff --git a/backend/tests/test_codex_retry_after.py b/backend/tests/test_codex_retry_after.py index 04e01abb89..0b9ef85038 100644 --- a/backend/tests/test_codex_retry_after.py +++ b/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" diff --git a/backend/tests/test_codex_stream_terminal_events.py b/backend/tests/test_codex_stream_terminal_events.py index 944df259d4..bee9ac7a2e 100644 --- a/backend/tests/test_codex_stream_terminal_events.py +++ b/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 diff --git a/backend/tests/test_codex_tool_choice.py b/backend/tests/test_codex_tool_choice.py index 1b3477aa90..ff73d1af6f 100644 --- a/backend/tests/test_codex_tool_choice.py +++ b/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) diff --git a/backend/tests/test_codex_tool_strict.py b/backend/tests/test_codex_tool_strict.py index 46d22fa9a9..7c875c34c8 100644 --- a/backend/tests/test_codex_tool_strict.py +++ b/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": diff --git a/backend/tests/test_model_request_admission.py b/backend/tests/test_model_request_admission.py index 43a7975fce..a59dc05ef3 100644 --- a/backend/tests/test_model_request_admission.py +++ b/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)]) diff --git a/docs/guide/configuration.md b/docs/guide/configuration.md index ed9af164b2..d5801a8e1c 100644 --- a/docs/guide/configuration.md +++ b/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.