fix: improve reliability for slow backend streams

README.md

@@ -104,8 +104,18 @@ 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
+BACKEND_CONNECT_TIMEOUT=30.0        # Backend connect/pool timeout (seconds)
+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 +198,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 
-

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
 

docs/open-responses-server.md

@@ -72,8 +72,9 @@ 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/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 |
 

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, 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) 
+        return await _handle_non_streaming_request(client, request_data) 

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}")

src/open_responses_server/common/llm_client.py

@@ -1,5 +1,21 @@
 import httpx
-from .config import OPENAI_BASE_URL_INTERNAL, OPENAI_API_KEY, STREAM_TIMEOUT, logger
+from .config import (
+    OPENAI_BASE_URL_INTERNAL,
+    OPENAI_API_KEY,
+    STREAM_TIMEOUT,
+    BACKEND_CONNECT_TIMEOUT,
+    logger,
+)
+
+
+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:
     """
@@ -18,7 +34,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 +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()

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,76 @@ 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
+
+    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()