From ed25cf29ef8f111f07344675c6a765cd8bf55a78 Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:12:30 +0400 Subject: [PATCH 1/5] fix: emit heartbeat events for slow backend streams --- docs/events-and-tool-handling.md | 22 +- src/open_responses_server/api_controller.py | 267 +++++++++++--------- tests/test_api_controller_endpoints.py | 150 +++++++---- 3 files changed, 254 insertions(+), 185 deletions(-) diff --git a/docs/events-and-tool-handling.md b/docs/events-and-tool-handling.md index 3724b08..0a7d7d5 100644 --- a/docs/events-and-tool-handling.md +++ b/docs/events-and-tool-handling.md @@ -277,15 +277,19 @@ When processing Responses API input with `function_call_output` items ## Connection Keepalive (Heartbeat) -When the backend LLM is slow to respond, the server sends SSE comment lines -(`: heartbeat\n\n`) at the interval configured by `HEARTBEAT_INTERVAL` -(default: 15 seconds). This prevents proxies and load balancers from closing -idle connections. - -Heartbeats are standard SSE comments and should be ignored by compliant clients. -The mechanism is implemented by `_with_heartbeat()` in `api_controller.py`, -which wraps the response stream and injects heartbeat sentinels during idle -periods. +When the backend LLM is slow to respond, the server sends a real SSE heartbeat +event at the interval configured by `HEARTBEAT_INTERVAL` (default: 15 seconds): + +```text +event: response.heartbeat +data: {"type":"response.heartbeat"} +``` + +This is intentionally emitted as a `data:` event rather than an SSE comment so +clients like Codex CLI reset their SSE idle timer after parsing it. The +mechanism is implemented by `_stream_with_keepalive()` in `api_controller.py`, +which starts the client-facing stream immediately and injects heartbeat events +while the upstream LLM request is still waiting for the first chunk. ## Pydantic Models diff --git a/src/open_responses_server/api_controller.py b/src/open_responses_server/api_controller.py index 42d5222..312e705 100644 --- a/src/open_responses_server/api_controller.py +++ b/src/open_responses_server/api_controller.py @@ -5,55 +5,66 @@ from fastapi.middleware.cors import CORSMiddleware from open_responses_server.common.config import logger, HEARTBEAT_INTERVAL, STREAM_TIMEOUT -from open_responses_server.common.llm_client import startup_llm_client, shutdown_llm_client, LLMClient +from open_responses_server.common.llm_client import ( + startup_llm_client, + shutdown_llm_client, + LLMClient, +) from open_responses_server.common.mcp_manager import mcp_manager from open_responses_server.responses_service import convert_responses_to_chat_completions, process_chat_completions_stream from open_responses_server.chat_completions_service import handle_chat_completions -_HEARTBEAT = object() +_STREAM_DONE = object() +_STREAM_ERROR = object() +_HEARTBEAT_EVENT = 'event: response.heartbeat\ndata: {"type":"response.heartbeat"}\n\n' -async def _with_heartbeat(async_gen, interval): - """Wrap an async generator to yield _HEARTBEAT sentinels during idle periods. +async def _stream_with_keepalive(async_iter_factory, interval, keepalive_event=_HEARTBEAT_EVENT): + """Yield keepalive comments while a producer waits on upstream stream activity. - Uses asyncio.wait with timeout so the underlying task is never cancelled. - This keeps SSE connections alive when the backend LLM is slow to respond. + Starts the client-facing SSE stream immediately and runs the upstream work + in a background task, so heartbeats continue even before the backend stream + context manager yields control. The heartbeat is a real SSE data event so + clients like Codex reset idle timers after parsing it. """ if not interval or interval <= 0: interval = 1.0 - inner = async_gen.__aiter__() - task = None + queue = asyncio.Queue() + + async def producer(): + try: + async for item in async_iter_factory(): + await queue.put(item) + except Exception as exc: + await queue.put((_STREAM_ERROR, exc)) + finally: + await queue.put(_STREAM_DONE) + + task = asyncio.create_task(producer()) try: while True: - task = asyncio.ensure_future(inner.__anext__()) - while not task.done(): - done, _ = await asyncio.wait({task}, timeout=interval) - if not done: - yield _HEARTBEAT try: - yield task.result() - except StopAsyncIteration: - return - finally: - task = None - finally: - await _cleanup_heartbeat(task, inner) + item = await asyncio.wait_for(queue.get(), timeout=interval) + except asyncio.TimeoutError: + logger.debug("[STREAM-HEARTBEAT] Sending SSE keepalive") + yield keepalive_event + continue + if item is _STREAM_DONE: + break -async def _cleanup_heartbeat(task, inner): - """Cancel in-flight task and close the underlying async iterator.""" - if task is not None and not task.done(): - task.cancel() - try: - await task - except asyncio.CancelledError: - raise - if hasattr(inner, "aclose"): - try: - await inner.aclose() - except Exception: - logger.debug("Error closing heartbeat inner iterator", exc_info=True) + if isinstance(item, tuple) and len(item) == 2 and item[0] is _STREAM_ERROR: + raise item[1] + + yield item + finally: + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass app = FastAPI( @@ -220,101 +231,101 @@ async def create_response(request: Request): # Handle streaming response async def stream_response(): try: - # Fetch available MCP tools and format as functions for chat.completions - mcp_functions = [] - for server in mcp_manager.mcp_servers: - try: - for t in await server.list_tools(): - mcp_functions.append({ - "name": t["name"], - "description": t.get("description"), - "parameters": t.get("parameters", {}), - }) - except Exception as e: - logger.warning(f"Error listing tools from {server.name}: {e}") - # Only include functions if we have them - if mcp_functions: - # Convert to the "tools" format which is more broadly supported - existing_tools = chat_request.get("tools", []) - existing_functions = chat_request.get("functions", []) - - # Convert any existing functions to tools format - for func in existing_functions: - existing_tools.append({ - "type": "function", - "function": func - }) - - # Get the names of existing tools to avoid duplicates - existing_tool_names = set() - for tool in existing_tools: - if isinstance(tool, dict) and "function" in tool and "name" in tool["function"]: - existing_tool_names.add(tool["function"]["name"]) - elif isinstance(tool, dict) and "name" in tool: - existing_tool_names.add(tool["name"]) - - # Only add MCP functions that don't conflict with existing tools - for func in mcp_functions: - if func["name"] not in existing_tool_names: - existing_tools.append({ - "type": "function", - "function": func - }) + async def llm_event_stream(): + # Fetch available MCP tools and format as functions for chat.completions + mcp_functions = [] + for server in mcp_manager.mcp_servers: + try: + for t in await server.list_tools(): + mcp_functions.append({ + "name": t["name"], + "description": t.get("description"), + "parameters": t.get("parameters", {}), + }) + except Exception as e: + logger.warning(f"Error listing tools from {server.name}: {e}") + # Only include functions if we have them + if mcp_functions: + # Convert to the "tools" format which is more broadly supported + existing_tools = chat_request.get("tools", []) + existing_functions = chat_request.get("functions", []) - # Set the tools and remove functions - chat_request["tools"] = existing_tools - chat_request.pop("functions", None) - - logger.info(f"Converted {len(existing_functions)} existing functions and {len(mcp_functions)} MCP functions to tools format") - elif "functions" in chat_request: - # Convert any existing functions to tools format - existing_tools = chat_request.get("tools", []) - existing_functions = chat_request.get("functions", []) - - if existing_functions: - # Convert functions to tools format + # Convert any existing functions to tools format for func in existing_functions: existing_tools.append({ "type": "function", "function": func }) + + # Get the names of existing tools to avoid duplicates + existing_tool_names = set() + for tool in existing_tools: + if isinstance(tool, dict) and "function" in tool and "name" in tool["function"]: + existing_tool_names.add(tool["function"]["name"]) + elif isinstance(tool, dict) and "name" in tool: + existing_tool_names.add(tool["name"]) + + # Only add MCP functions that don't conflict with existing tools + for func in mcp_functions: + if func["name"] not in existing_tool_names: + existing_tools.append({ + "type": "function", + "function": func + }) + # Set the tools and remove functions chat_request["tools"] = existing_tools - logger.info(f"Converted {len(existing_functions)} existing functions to tools format") - - # Remove the functions key regardless - chat_request.pop("functions", None) - - if not chat_request.get("tools"): - # If we don't have any tools either, remove that key - chat_request.pop("tools", None) - logger.info("No tools or functions available, sending without them") - # Log the initial Chat Completions request payload - logger.info(f"Sending Chat Completions request: {json.dumps(chat_request)}") - client = await LLMClient.get_client() - async with client.stream( - "POST", - "/v1/chat/completions", - json=chat_request, - timeout=STREAM_TIMEOUT - ) as response: - logger.info(f"Stream request status: {response.status_code}") - - if response.status_code != 200: - error_content = await response.aread() - logger.error(f"Error from LLM API: {error_content}") - yield f"data: {json.dumps({'type': 'error', 'error': {'message': f'Error from LLM API: {response.status_code}'}})}\n\n" - return - - async for event in _with_heartbeat( - process_chat_completions_stream(response, chat_request), - HEARTBEAT_INTERVAL - ): - if event is _HEARTBEAT: - logger.debug("[STREAM-HEARTBEAT] Sending SSE keepalive") - yield ": heartbeat\n\n" - else: + chat_request.pop("functions", None) + + logger.info(f"Converted {len(existing_functions)} existing functions and {len(mcp_functions)} MCP functions to tools format") + elif "functions" in chat_request: + # Convert any existing functions to tools format + existing_tools = chat_request.get("tools", []) + existing_functions = chat_request.get("functions", []) + + if existing_functions: + # Convert functions to tools format + for func in existing_functions: + existing_tools.append({ + "type": "function", + "function": func + }) + + chat_request["tools"] = existing_tools + logger.info(f"Converted {len(existing_functions)} existing functions to tools format") + + # Remove the functions key regardless + chat_request.pop("functions", None) + + if not chat_request.get("tools"): + # If we don't have any tools either, remove that key + chat_request.pop("tools", None) + logger.info("No tools or functions available, sending without them") + # Log the initial Chat Completions request payload + logger.info(f"Sending Chat Completions request: {json.dumps(chat_request)}") + client = await LLMClient.get_client() + async with client.stream( + "POST", + "/v1/chat/completions", + json=chat_request, + timeout=STREAM_TIMEOUT + ) as response: + logger.info(f"Stream request status: {response.status_code}") + + if response.status_code != 200: + error_content = await response.aread() + logger.error(f"Error from LLM API: {error_content}") + yield f"data: {json.dumps({'type': 'error', 'error': {'message': f'Error from LLM API: {response.status_code}'}})}\n\n" + return + + async for event in process_chat_completions_stream(response, chat_request): yield event + + async for event in _stream_with_keepalive( + llm_event_stream, + HEARTBEAT_INTERVAL + ): + yield event except Exception as e: logger.error(f"Error in stream_response: {str(e)}") yield f"data: {json.dumps({'type': 'error', 'error': {'message': str(e)}})}\n\n" @@ -386,14 +397,26 @@ async def proxy_endpoint(request: Request, path_name: str): if is_stream: async def stream_proxy(): - async with client.stream(request.method, url, headers=headers, content=body, timeout=STREAM_TIMEOUT) as response: + async with client.stream( + request.method, + url, + headers=headers, + content=body, + timeout=STREAM_TIMEOUT + ) as response: async for chunk in response.aiter_bytes(): yield chunk return StreamingResponse(stream_proxy(), media_type=request.headers.get('accept', 'application/json')) else: - response = await client.request(request.method, url, headers=headers, content=body, timeout=STREAM_TIMEOUT) + response = await client.request( + request.method, + url, + headers=headers, + content=body, + timeout=STREAM_TIMEOUT + ) return Response(content=response.content, status_code=response.status_code, headers=response.headers) except Exception as e: logger.error(f"Error in proxy endpoint: {e}") - raise HTTPException(status_code=500, detail=f"Error proxying request: {str(e)}") \ No newline at end of file + raise HTTPException(status_code=500, detail=f"Error proxying request: {str(e)}") diff --git a/tests/test_api_controller_endpoints.py b/tests/test_api_controller_endpoints.py index b655b3b..28e126c 100644 --- a/tests/test_api_controller_endpoints.py +++ b/tests/test_api_controller_endpoints.py @@ -8,7 +8,7 @@ from fastapi.testclient import TestClient from fastapi.responses import StreamingResponse -from open_responses_server.api_controller import app, _with_heartbeat, _HEARTBEAT +from open_responses_server.api_controller import app, _stream_with_keepalive, _HEARTBEAT_EVENT class TestResponsesEndpoint: @@ -172,6 +172,58 @@ async def fake_aiter_lines(): response = client.post("/responses", json=request_data) assert response.status_code == 200 + def test_responses_streaming_sends_keepalive_before_backend_yields(self, client, mock_llm_client_fixture, monkeypatch): + """POST /responses emits SSE heartbeat events while backend setup is still waiting.""" + monkeypatch.setattr("open_responses_server.api_controller.HEARTBEAT_INTERVAL", 0.05) + + mock_client = mock_llm_client_fixture + + mock_stream_resp = MagicMock() + mock_stream_resp.status_code = 200 + + async def delayed_enter(*_args, **_kwargs): + await asyncio.sleep(0.16) + return mock_stream_resp + + mock_stream_resp.__aenter__ = delayed_enter + mock_stream_resp.__aexit__ = AsyncMock(return_value=False) + + async def fake_aiter_lines(): + yield 'data: {"choices":[{"delta":{"content":"Hi"},"index":0}],"model":"test"}' + yield 'data: [DONE]' + + mock_stream_resp.aiter_lines = fake_aiter_lines + mock_stream_resp.aread = AsyncMock(return_value=b'error') + mock_client.stream = MagicMock(return_value=mock_stream_resp) + + request_data = { + "model": "test-model", + "input": [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "Hello"}]}], + "stream": True, + } + + with client.stream("POST", "/responses", json=request_data) as response: + assert response.status_code == 200 + + lines = [] + for idx, line in enumerate(response.iter_lines(), start=1): + if line: + lines.append(line) + if 'event: response.heartbeat' in lines and any( + item.startswith('data: {"type":"response.heartbeat"}') for item in lines + ): + break + if any( + item.startswith("data: ") and 'response.created' in item for item in lines + ): + break + if idx >= 20: + pytest.fail(f"Timed out waiting for heartbeat/data lines: {lines}") + + assert 'event: response.heartbeat' in lines + assert any(item.startswith('data: {"type":"response.heartbeat"}') for item in lines) + assert any(item.startswith("data: ") for item in lines) + class TestChatCompletionsEndpoint: @pytest.fixture @@ -277,63 +329,53 @@ def test_proxy_invalid_json_body(self, client, mock_llm_client_fixture): @pytest.mark.asyncio -class TestWithHeartbeat: - """Tests for the _with_heartbeat async generator wrapper.""" - - async def test_fast_generator_no_heartbeats(self): - """Fast generators produce no heartbeat sentinels.""" - async def fast_gen(): - yield "a" - yield "b" - yield "c" - - results = [item async for item in _with_heartbeat(fast_gen(), interval=10.0)] - assert results == ["a", "b", "c"] - assert _HEARTBEAT not in results - - async def test_slow_generator_emits_heartbeats(self): - """Slow generators trigger heartbeat sentinels between items.""" - async def slow_gen(): - yield "first" - await asyncio.sleep(0.6) - yield "second" - - results = [item async for item in _with_heartbeat(slow_gen(), interval=0.2)] - # Should have at least one heartbeat between "first" and "second" - heartbeats = [r for r in results if r is _HEARTBEAT] - data = [r for r in results if r is not _HEARTBEAT] - assert len(heartbeats) >= 1 - assert data == ["first", "second"] - - async def test_empty_generator(self): - """Empty generator produces no output.""" - async def empty_gen(): +class TestStreamWithKeepalive: + """Tests for heartbeat events while upstream setup is still waiting.""" + + async def test_fast_stream_emits_no_heartbeats(self): + """Immediate upstream items should pass through without heartbeat events.""" + async def fast_stream(): + yield "data: first\n\n" + yield "data: second\n\n" + + results = [item async for item in _stream_with_keepalive(fast_stream, interval=1.0)] + + assert results == ["data: first\n\n", "data: second\n\n"] + + async def test_emits_keepalives_before_first_upstream_item(self): + """Heartbeat events should flow before the upstream stream yields its first item.""" + async def delayed_stream(): + await asyncio.sleep(0.35) + yield "data: first\n\n" + + results = [item async for item in _stream_with_keepalive(delayed_stream, interval=0.1)] + + heartbeats = [item for item in results if item == _HEARTBEAT_EVENT] + data = [item for item in results if item != _HEARTBEAT_EVENT] + + assert len(heartbeats) >= 2 + assert data == ["data: first\n\n"] + + async def test_empty_stream_produces_no_output(self): + """An upstream stream that completes immediately should stay silent.""" + async def empty_stream(): return - yield # noqa: unreachable - makes this an async generator + yield # noqa: unreachable - keeps this as an async generator + + results = [item async for item in _stream_with_keepalive(empty_stream, interval=0.1)] - results = [item async for item in _with_heartbeat(empty_gen(), interval=1.0)] assert results == [] - async def test_generator_exception_propagates(self): - """Exceptions from the wrapped generator propagate through.""" - async def error_gen(): - yield "ok" - raise ValueError("test error") + async def test_error_propagates_after_heartbeats(self): + """Producer exceptions should surface to the consumer.""" + async def error_stream(): + await asyncio.sleep(0.15) + raise ValueError("upstream failed") + yield # noqa: unreachable - keeps this as an async generator results = [] - with pytest.raises(ValueError, match="test error"): - async for item in _with_heartbeat(error_gen(), interval=1.0): + with pytest.raises(ValueError, match="upstream failed"): + async for item in _stream_with_keepalive(error_stream, interval=0.05): results.append(item) - assert results == ["ok"] - - async def test_heartbeat_count_scales_with_delay(self): - """Longer delays produce more heartbeats.""" - async def very_slow_gen(): - yield "start" - await asyncio.sleep(1.0) - yield "end" - - results = [item async for item in _with_heartbeat(very_slow_gen(), interval=0.2)] - heartbeats = [r for r in results if r is _HEARTBEAT] - # ~1.0s delay / 0.2s interval = ~5 heartbeats (allow some variance) - assert len(heartbeats) >= 3 + + assert _HEARTBEAT_EVENT in results From bae5449c2521befdb69ebba0560c4563867a541b Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:12:52 +0400 Subject: [PATCH 2/5] fix: use separate connect and read timeouts for backend streams --- README.md | 14 +++++++++++--- docs/open-responses-server.md | 4 ++-- src/open_responses_server/api_controller.py | 9 +++++---- .../chat_completions_service.py | 16 ++++++++++------ src/open_responses_server/common/llm_client.py | 16 ++++++++++++++-- tests/test_llm_client.py | 13 +++++++++++++ 6 files changed, 55 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 683625f..796bb88 100644 --- a/README.md +++ b/README.md @@ -104,8 +104,17 @@ API_ADAPTER_PORT=8080 ``` Streaming and connection: ``` -STREAM_TIMEOUT=120.0 # HTTP timeout (seconds) for streaming requests -HEARTBEAT_INTERVAL=15.0 # SSE keepalive interval (seconds) +STREAM_TIMEOUT=120.0 # Backend read timeout (seconds) for slow streaming responses +HEARTBEAT_INTERVAL=15.0 # SSE heartbeat event interval (seconds) +``` + +For slow local models such as 31B dense or larger, Codex CLI may also need a +longer client-side idle timeout: + +```toml +# ~/.codex/config.toml +[model_providers.llamacpp] +stream_idle_timeout_ms = 900000 # 15 minutes ``` Conversation and tool handling: ``` @@ -188,4 +197,3 @@ TeaBranch. (2025). open-responses-server: Open-source server the serves any AI p This repo had changed names: - openai-responses-server (Changed to avoid brand name OpenAI) - open-responses-server - diff --git a/docs/open-responses-server.md b/docs/open-responses-server.md index 10ce1e8..e854f87 100644 --- a/docs/open-responses-server.md +++ b/docs/open-responses-server.md @@ -72,8 +72,8 @@ All configuration is via environment variables, loaded from `.env` via | `MCP_SERVERS_CONFIG_PATH` | `src/open_responses_server/servers_config.json` | Path to MCP servers JSON config (use absolute path when pip-installed) | | `MAX_CONVERSATION_HISTORY` | `100` | Max stored conversation entries | | `MAX_TOOL_CALL_ITERATIONS` | `25` | Max tool-call loop iterations | -| `STREAM_TIMEOUT` | `120.0` | HTTP timeout (seconds) for streaming requests | -| `HEARTBEAT_INTERVAL` | `15.0` | SSE keepalive interval (seconds) | +| `STREAM_TIMEOUT` | `120.0` | Backend read timeout in seconds; connect timeout stays at 30s | +| `HEARTBEAT_INTERVAL` | `15.0` | SSE heartbeat event interval in seconds | | `LOG_LEVEL` | `INFO` | Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) | | `LOG_FILE_PATH` | `./log/api_adapter.log` | Path to log file | diff --git a/src/open_responses_server/api_controller.py b/src/open_responses_server/api_controller.py index 312e705..1cac567 100644 --- a/src/open_responses_server/api_controller.py +++ b/src/open_responses_server/api_controller.py @@ -4,11 +4,12 @@ from fastapi.responses import StreamingResponse, Response from fastapi.middleware.cors import CORSMiddleware -from open_responses_server.common.config import logger, HEARTBEAT_INTERVAL, STREAM_TIMEOUT +from open_responses_server.common.config import logger, HEARTBEAT_INTERVAL from open_responses_server.common.llm_client import ( startup_llm_client, shutdown_llm_client, LLMClient, + get_backend_timeout, ) from open_responses_server.common.mcp_manager import mcp_manager from open_responses_server.responses_service import convert_responses_to_chat_completions, process_chat_completions_stream @@ -308,7 +309,7 @@ async def llm_event_stream(): "POST", "/v1/chat/completions", json=chat_request, - timeout=STREAM_TIMEOUT + timeout=get_backend_timeout() ) as response: logger.info(f"Stream request status: {response.status_code}") @@ -402,7 +403,7 @@ async def stream_proxy(): url, headers=headers, content=body, - timeout=STREAM_TIMEOUT + timeout=get_backend_timeout() ) as response: async for chunk in response.aiter_bytes(): yield chunk @@ -413,7 +414,7 @@ async def stream_proxy(): url, headers=headers, content=body, - timeout=STREAM_TIMEOUT + timeout=get_backend_timeout() ) return Response(content=response.content, status_code=response.status_code, headers=response.headers) diff --git a/src/open_responses_server/chat_completions_service.py b/src/open_responses_server/chat_completions_service.py index b01ce66..137b67d 100644 --- a/src/open_responses_server/chat_completions_service.py +++ b/src/open_responses_server/chat_completions_service.py @@ -1,8 +1,8 @@ import json from fastapi import Request from fastapi.responses import StreamingResponse, Response, JSONResponse -from open_responses_server.common.llm_client import LLMClient -from open_responses_server.common.config import logger, OPENAI_BASE_URL_INTERNAL, OPENAI_API_KEY, MAX_TOOL_CALL_ITERATIONS, STREAM_TIMEOUT +from open_responses_server.common.llm_client import LLMClient, get_backend_timeout +from open_responses_server.common.config import logger, OPENAI_BASE_URL_INTERNAL, OPENAI_API_KEY, MAX_TOOL_CALL_ITERATIONS from open_responses_server.common.mcp_manager import mcp_manager, serialize_tool_result async def _handle_non_streaming_request(client: LLMClient, request_data: dict): @@ -25,7 +25,7 @@ async def _handle_non_streaming_request(client: LLMClient, request_data: dict): response = await client.post( "/v1/chat/completions", json=current_request_data, - timeout=STREAM_TIMEOUT + timeout=get_backend_timeout() ) response.raise_for_status() response_data = response.json() @@ -102,7 +102,11 @@ async def _handle_streaming_request(client: LLMClient, request_data: dict) -> St for _ in range(MAX_TOOL_CALL_ITERATIONS): try: # Make a non-streaming request first to check for tool calls - response = await client.post("/v1/chat/completions", json={**non_stream_request_data, "messages": messages}, timeout=STREAM_TIMEOUT) + response = await client.post( + "/v1/chat/completions", + json={**non_stream_request_data, "messages": messages}, + timeout=get_backend_timeout() + ) response.raise_for_status() response_data = response.json() @@ -170,7 +174,7 @@ async def stream_proxy(): "POST", "/v1/chat/completions", json=stream_request_data, - timeout=STREAM_TIMEOUT + timeout=get_backend_timeout() ) as stream_response: async for chunk in stream_response.aiter_bytes(): yield chunk @@ -233,4 +237,4 @@ async def handle_chat_completions(request: Request): if is_stream: return await _handle_streaming_request(client, request_data) else: - return await _handle_non_streaming_request(client, request_data) \ No newline at end of file + return await _handle_non_streaming_request(client, request_data) diff --git a/src/open_responses_server/common/llm_client.py b/src/open_responses_server/common/llm_client.py index 5488bd2..77558b0 100644 --- a/src/open_responses_server/common/llm_client.py +++ b/src/open_responses_server/common/llm_client.py @@ -1,6 +1,18 @@ import httpx from .config import OPENAI_BASE_URL_INTERNAL, OPENAI_API_KEY, STREAM_TIMEOUT, logger +BACKEND_CONNECT_TIMEOUT = 30.0 + + +def get_backend_timeout() -> httpx.Timeout: + """Use a long read timeout for slow local inference, but keep connect short.""" + return httpx.Timeout( + connect=BACKEND_CONNECT_TIMEOUT, + read=STREAM_TIMEOUT, + write=STREAM_TIMEOUT, + pool=BACKEND_CONNECT_TIMEOUT, + ) + class LLMClient: """ An asynchronous client for interacting with the LLM API. @@ -18,7 +30,7 @@ async def get_client(cls) -> httpx.AsyncClient: cls._client = httpx.AsyncClient( base_url=OPENAI_BASE_URL_INTERNAL, headers={"Authorization": f"Bearer {OPENAI_API_KEY}"}, - timeout=httpx.Timeout(STREAM_TIMEOUT) + timeout=get_backend_timeout() ) return cls._client @@ -41,4 +53,4 @@ async def startup_llm_client(): async def shutdown_llm_client(): """Function to be called on application shutdown.""" - await LLMClient.close_client() \ No newline at end of file + await LLMClient.close_client() diff --git a/tests/test_llm_client.py b/tests/test_llm_client.py index 198a92c..5de71c7 100644 --- a/tests/test_llm_client.py +++ b/tests/test_llm_client.py @@ -5,8 +5,11 @@ from unittest.mock import patch, AsyncMock, MagicMock import httpx +from open_responses_server.common.config import STREAM_TIMEOUT from open_responses_server.common.llm_client import ( LLMClient, + BACKEND_CONNECT_TIMEOUT, + get_backend_timeout, startup_llm_client, shutdown_llm_client, ) @@ -21,12 +24,22 @@ def reset_llm_client(): class TestLLMClient: + def test_backend_timeout_uses_long_read_and_short_connect(self): + """Backend timeout should keep connect short while allowing slow reads.""" + timeout = get_backend_timeout() + assert timeout.connect == BACKEND_CONNECT_TIMEOUT + assert timeout.pool == BACKEND_CONNECT_TIMEOUT + assert timeout.read == STREAM_TIMEOUT + assert timeout.write == STREAM_TIMEOUT + @pytest.mark.asyncio async def test_get_client_creates_instance(self): """get_client creates an httpx.AsyncClient on first call.""" client = await LLMClient.get_client() assert client is not None assert isinstance(client, httpx.AsyncClient) + assert client.timeout.connect == BACKEND_CONNECT_TIMEOUT + assert client.timeout.read == STREAM_TIMEOUT await client.aclose() @pytest.mark.asyncio From 0857f4d339c22ac5b2ca323ee07610d08640f325 Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:47:09 +0400 Subject: [PATCH 3/5] fix: move backend connect timeout into env config --- README.md | 1 + docs/open-responses-server.md | 3 ++- src/open_responses_server/common/config.py | 2 ++ src/open_responses_server/common/llm_client.py | 10 +++++++--- tests/test_llm_client.py | 5 ++--- 5 files changed, 14 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 796bb88..164823c 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,7 @@ API_ADAPTER_PORT=8080 Streaming and connection: ``` STREAM_TIMEOUT=120.0 # Backend read timeout (seconds) for slow streaming responses +BACKEND_CONNECT_TIMEOUT=30.0 # Backend connect/pool timeout (seconds) HEARTBEAT_INTERVAL=15.0 # SSE heartbeat event interval (seconds) ``` diff --git a/docs/open-responses-server.md b/docs/open-responses-server.md index e854f87..c28ab88 100644 --- a/docs/open-responses-server.md +++ b/docs/open-responses-server.md @@ -72,7 +72,8 @@ All configuration is via environment variables, loaded from `.env` via | `MCP_SERVERS_CONFIG_PATH` | `src/open_responses_server/servers_config.json` | Path to MCP servers JSON config (use absolute path when pip-installed) | | `MAX_CONVERSATION_HISTORY` | `100` | Max stored conversation entries | | `MAX_TOOL_CALL_ITERATIONS` | `25` | Max tool-call loop iterations | -| `STREAM_TIMEOUT` | `120.0` | Backend read timeout in seconds; connect timeout stays at 30s | +| `STREAM_TIMEOUT` | `120.0` | Backend read/write timeout in seconds for slow streaming requests | +| `BACKEND_CONNECT_TIMEOUT` | `30.0` | Backend connect/pool timeout in seconds | | `HEARTBEAT_INTERVAL` | `15.0` | SSE heartbeat event interval in seconds | | `LOG_LEVEL` | `INFO` | Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) | | `LOG_FILE_PATH` | `./log/api_adapter.log` | Path to log file | diff --git a/src/open_responses_server/common/config.py b/src/open_responses_server/common/config.py index 3532d98..18833a6 100644 --- a/src/open_responses_server/common/config.py +++ b/src/open_responses_server/common/config.py @@ -25,6 +25,7 @@ # Streaming Configuration STREAM_TIMEOUT = float(os.environ.get("STREAM_TIMEOUT", "120.0")) HEARTBEAT_INTERVAL = float(os.environ.get("HEARTBEAT_INTERVAL", "15.0")) +BACKEND_CONNECT_TIMEOUT = float(os.environ.get("BACKEND_CONNECT_TIMEOUT", "30.0")) # Logging Configuration LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper() @@ -70,5 +71,6 @@ def setup_logging(): logger.info(f" MAX_TOOL_CALL_ITERATIONS: {MAX_TOOL_CALL_ITERATIONS}") logger.info(f" STREAM_TIMEOUT: {STREAM_TIMEOUT}") logger.info(f" HEARTBEAT_INTERVAL: {HEARTBEAT_INTERVAL}") +logger.info(f" BACKEND_CONNECT_TIMEOUT: {BACKEND_CONNECT_TIMEOUT}") logger.info(f" LOG_LEVEL: {LOG_LEVEL}") logger.info(f" LOG_FILE_PATH: {LOG_FILE_PATH}") diff --git a/src/open_responses_server/common/llm_client.py b/src/open_responses_server/common/llm_client.py index 77558b0..69e1570 100644 --- a/src/open_responses_server/common/llm_client.py +++ b/src/open_responses_server/common/llm_client.py @@ -1,7 +1,11 @@ import httpx -from .config import OPENAI_BASE_URL_INTERNAL, OPENAI_API_KEY, STREAM_TIMEOUT, logger - -BACKEND_CONNECT_TIMEOUT = 30.0 +from .config import ( + OPENAI_BASE_URL_INTERNAL, + OPENAI_API_KEY, + STREAM_TIMEOUT, + BACKEND_CONNECT_TIMEOUT, + logger, +) def get_backend_timeout() -> httpx.Timeout: diff --git a/tests/test_llm_client.py b/tests/test_llm_client.py index 5de71c7..67d61ca 100644 --- a/tests/test_llm_client.py +++ b/tests/test_llm_client.py @@ -1,14 +1,13 @@ """ Tests for the LLM client module. """ +import httpx import pytest from unittest.mock import patch, AsyncMock, MagicMock -import httpx -from open_responses_server.common.config import STREAM_TIMEOUT +from open_responses_server.common.config import STREAM_TIMEOUT, BACKEND_CONNECT_TIMEOUT from open_responses_server.common.llm_client import ( LLMClient, - BACKEND_CONNECT_TIMEOUT, get_backend_timeout, startup_llm_client, shutdown_llm_client, From bedc50bf1683d1b847c5fb6af51fc0497db73e9c Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Fri, 10 Apr 2026 01:50:44 +0400 Subject: [PATCH 4/5] fix: preserve backpressure in stream keepalive wrapper --- src/open_responses_server/api_controller.py | 4 +++- tests/test_api_controller_endpoints.py | 23 +++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/open_responses_server/api_controller.py b/src/open_responses_server/api_controller.py index 1cac567..3663f3a 100644 --- a/src/open_responses_server/api_controller.py +++ b/src/open_responses_server/api_controller.py @@ -31,7 +31,9 @@ async def _stream_with_keepalive(async_iter_factory, interval, keepalive_event=_ if not interval or interval <= 0: interval = 1.0 - queue = asyncio.Queue() + # Keep just one prefetched item so heartbeats can cover upstream waits + # without breaking normal stream backpressure for slow clients. + queue = asyncio.Queue(maxsize=1) async def producer(): try: diff --git a/tests/test_api_controller_endpoints.py b/tests/test_api_controller_endpoints.py index 28e126c..46d455e 100644 --- a/tests/test_api_controller_endpoints.py +++ b/tests/test_api_controller_endpoints.py @@ -379,3 +379,26 @@ async def error_stream(): results.append(item) assert _HEARTBEAT_EVENT in results + + async def test_bounded_queue_preserves_backpressure(self): + """The wrapper should not let the producer run far ahead of a slow consumer.""" + produced = 0 + producer_done = asyncio.Event() + + async def fast_stream(): + nonlocal produced + for idx in range(5): + produced += 1 + yield f"data: item-{idx}\n\n" + producer_done.set() + + consumed = 0 + async for item in _stream_with_keepalive(fast_stream, interval=1.0): + await asyncio.sleep(0.05) + consumed += 1 + if consumed == 1: + assert produced < 5 + assert not producer_done.is_set() + + assert consumed == 5 + assert producer_done.is_set() From 647e68aa7cea5d6c526935cf9258f9bb2cbe7db9 Mon Sep 17 00:00:00 2001 From: Ivan Oparin Date: Wed, 27 May 2026 12:52:55 +0400 Subject: [PATCH 5/5] chore: address slow stream review cleanup --- src/open_responses_server/api_controller.py | 2 +- src/open_responses_server/chat_completions_service.py | 2 +- src/open_responses_server/common/llm_client.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/open_responses_server/api_controller.py b/src/open_responses_server/api_controller.py index 3663f3a..783c061 100644 --- a/src/open_responses_server/api_controller.py +++ b/src/open_responses_server/api_controller.py @@ -21,7 +21,7 @@ async def _stream_with_keepalive(async_iter_factory, interval, keepalive_event=_HEARTBEAT_EVENT): - """Yield keepalive comments while a producer waits on upstream stream activity. + """Yield SSE heartbeat events while a producer waits on upstream stream activity. Starts the client-facing SSE stream immediately and runs the upstream work in a background task, so heartbeats continue even before the backend stream diff --git a/src/open_responses_server/chat_completions_service.py b/src/open_responses_server/chat_completions_service.py index 137b67d..359590c 100644 --- a/src/open_responses_server/chat_completions_service.py +++ b/src/open_responses_server/chat_completions_service.py @@ -2,7 +2,7 @@ from fastapi import Request from fastapi.responses import StreamingResponse, Response, JSONResponse from open_responses_server.common.llm_client import LLMClient, get_backend_timeout -from open_responses_server.common.config import logger, OPENAI_BASE_URL_INTERNAL, OPENAI_API_KEY, MAX_TOOL_CALL_ITERATIONS +from open_responses_server.common.config import logger, MAX_TOOL_CALL_ITERATIONS from open_responses_server.common.mcp_manager import mcp_manager, serialize_tool_result async def _handle_non_streaming_request(client: LLMClient, request_data: dict): diff --git a/src/open_responses_server/common/llm_client.py b/src/open_responses_server/common/llm_client.py index 69e1570..a35f58a 100644 --- a/src/open_responses_server/common/llm_client.py +++ b/src/open_responses_server/common/llm_client.py @@ -57,4 +57,4 @@ async def startup_llm_client(): async def shutdown_llm_client(): """Function to be called on application shutdown.""" - await LLMClient.close_client() + await LLMClient.close_client()