Prevent failed SSE requests from masquerading as successful completions
The streaming API currently logs backend exceptions but still emits the same `finish_reason: "stop"` + `[DONE]` terminator used for successful responses. That makes a failed streamed request look successful to OpenAI-compatible clients. This keeps the fix narrow: track whether the stream backend failed and suppress the success terminator in that case. A regression test locks in the expected behavior. Constraint: Keep the non-streaming response path untouched Constraint: Follow up on the known limitation called out during PR #3222 review without redesigning the SSE protocol Rejected: Introduce a custom SSE error event shape in the same patch | expands API surface and review scope Confidence: high Scope-risk: narrow Reversibility: clean Directive: If explicit streamed error events are added later, keep them distinct from the success stop+[DONE] terminator to preserve client retry semantics Tested: PYTHONPATH=$PWD pytest -q tests/test_api_stream.py /Users/jh0927/Workspace/nanobot-validation-artifacts-2026-04-18/test_api_stream_error_regression.py Not-tested: Full repository test suite Related: #3260 Related: #3222
This commit is contained in:
@@ -256,6 +256,7 @@ async def handle_chat_completions(request: web.Request) -> web.Response:
|
|||||||
|
|
||||||
chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
|
chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
|
||||||
queue: asyncio.Queue[str | None] = asyncio.Queue()
|
queue: asyncio.Queue[str | None] = asyncio.Queue()
|
||||||
|
stream_failed = False
|
||||||
|
|
||||||
async def _on_stream(token: str) -> None:
|
async def _on_stream(token: str) -> None:
|
||||||
await queue.put(token)
|
await queue.put(token)
|
||||||
@@ -264,6 +265,7 @@ async def handle_chat_completions(request: web.Request) -> web.Response:
|
|||||||
await queue.put(None)
|
await queue.put(None)
|
||||||
|
|
||||||
async def _run() -> None:
|
async def _run() -> None:
|
||||||
|
nonlocal stream_failed
|
||||||
try:
|
try:
|
||||||
async with session_lock:
|
async with session_lock:
|
||||||
await asyncio.wait_for(
|
await asyncio.wait_for(
|
||||||
@@ -279,6 +281,7 @@ async def handle_chat_completions(request: web.Request) -> web.Response:
|
|||||||
timeout=timeout_s,
|
timeout=timeout_s,
|
||||||
)
|
)
|
||||||
except Exception:
|
except Exception:
|
||||||
|
stream_failed = True
|
||||||
logger.exception("Streaming error for session {}", session_key)
|
logger.exception("Streaming error for session {}", session_key)
|
||||||
await queue.put(None)
|
await queue.put(None)
|
||||||
|
|
||||||
@@ -292,6 +295,7 @@ async def handle_chat_completions(request: web.Request) -> web.Response:
|
|||||||
finally:
|
finally:
|
||||||
task.cancel()
|
task.cancel()
|
||||||
|
|
||||||
|
if not stream_failed:
|
||||||
await resp.write(_sse_chunk("", model_name, chunk_id, finish_reason="stop"))
|
await resp.write(_sse_chunk("", model_name, chunk_id, finish_reason="stop"))
|
||||||
await resp.write(_SSE_DONE)
|
await resp.write(_SSE_DONE)
|
||||||
return resp
|
return resp
|
||||||
|
|||||||
@@ -251,3 +251,30 @@ async def test_stream_with_session_id(aiohttp_client) -> None:
|
|||||||
)
|
)
|
||||||
assert resp.status == 200
|
assert resp.status == 200
|
||||||
assert captured_key == "api:my-session"
|
assert captured_key == "api:my-session"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(not HAS_AIOHTTP, reason="aiohttp not installed")
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_streaming_backend_failure_does_not_emit_success_terminator(aiohttp_client) -> None:
|
||||||
|
"""Backend exceptions should not surface as a normal stop+[DONE] stream."""
|
||||||
|
agent = MagicMock()
|
||||||
|
|
||||||
|
async def boom(**kwargs):
|
||||||
|
raise RuntimeError("backend blew up")
|
||||||
|
|
||||||
|
agent.process_direct = boom
|
||||||
|
agent._connect_mcp = AsyncMock()
|
||||||
|
agent.close_mcp = AsyncMock()
|
||||||
|
|
||||||
|
app = create_app(agent, model_name="m")
|
||||||
|
client = await aiohttp_client(app)
|
||||||
|
|
||||||
|
resp = await client.post(
|
||||||
|
"/v1/chat/completions",
|
||||||
|
json={"messages": [{"role": "user", "content": "hi"}], "stream": True},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert resp.status == 200
|
||||||
|
body = await resp.text()
|
||||||
|
assert '"finish_reason": "stop"' not in body
|
||||||
|
assert "[DONE]" not in body
|
||||||
|
|||||||
Reference in New Issue
Block a user