diff --git a/datamind/server.py b/datamind/server.py index 7a24ff3..978eeba 100644 --- a/datamind/server.py +++ b/datamind/server.py @@ -235,12 +235,19 @@ async def chat( async def stream() -> AsyncIterator[bytes]: context = RequestContext(session_id=session, profile=st.settings.data.profile) with bind_context(context): - async for event in st.system.retrieve.loop.stream_turn( - user_message=req.message, - history=req.history or [], - ): + try: + async for event in st.system.retrieve.loop.stream_turn( + user_message=req.message, + history=req.history or [], + ): + payload = json.dumps( + {"type": event.type, **event.data}, + ensure_ascii=False, + ) + yield f"data: {payload}\n\n".encode("utf-8") + except Exception as exc: # noqa: BLE001 - convert provider errors to SSE payload = json.dumps( - {"type": event.type, **event.data}, + {"type": "error", "message": f"stream failed: {type(exc).__name__}"}, ensure_ascii=False, ) yield f"data: {payload}\n\n".encode("utf-8") diff --git a/datamind/tests/test_http_api.py b/datamind/tests/test_http_api.py index e7e5b8f..65c9e47 100644 --- a/datamind/tests/test_http_api.py +++ b/datamind/tests/test_http_api.py @@ -53,6 +53,12 @@ async def stream_turn(self, *, user_message: str, history=None, final_contract=N }) +class _FailingStreamLoop(_FakeLoop): + async def stream_turn(self, *, user_message: str, history=None, final_contract=None): + yield AgentEvent("text", {"delta": "partial"}) + raise RuntimeError("provider unavailable") + + class _FakeKB: async def list_documents(self): return [{"source": "demo.md", "chunks": 1}] @@ -197,3 +203,18 @@ async def test_http_api_routes_round_trip(configured_app): graph = await client.get("/api/graph/stats") assert graph.json() == {"nodes": 2, "edges": 1} + + +@pytest.mark.asyncio +async def test_chat_converts_stream_failure_to_sse_error(configured_app): + test_app, _settings = configured_app + test_app.state.datamind.system.retrieve.loop = _FailingStreamLoop("retrieve") + transport = httpx.ASGITransport(app=test_app, raise_app_exceptions=True) + + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + response = await client.post("/api/chat", json={"message": "stream this"}) + + assert response.status_code == 200 + assert '"type": "text"' in response.text + assert '"type": "error"' in response.text + assert '"message": "stream failed: RuntimeError"' in response.text