diff --git a/ravendb/documents/operations/ai/agents/run_conversation_operation.py b/ravendb/documents/operations/ai/agents/run_conversation_operation.py index 4fc37e33..44945601 100644 --- a/ravendb/documents/operations/ai/agents/run_conversation_operation.py +++ b/ravendb/documents/operations/ai/agents/run_conversation_operation.py @@ -8,6 +8,7 @@ from ravendb.documents.conventions import DocumentConventions from ravendb.http.raven_command import RavenCommand, RavenCommandResponseType from ravendb.http.server_node import ServerNode +from ravendb.exceptions.exception_dispatcher import ExceptionDispatcher import requests from ravendb.http.misc import ResponseDisposeHandling from ravendb.documents.ai.ai_output_options import AiOutputOptions @@ -478,6 +479,17 @@ def process_response(self, cache, response: requests.Response, url) -> ResponseD continue if line.startswith("{"): response_json = json.loads(line) + # An exception raised after streaming already started (HTTP 200) is written into the stream as the + # standard error payload. Surface it (e.g. a refusal) instead of parsing it as the result. + exception_type = response_json.get("Type") if isinstance(response_json, dict) else None + if exception_type: + schema = ExceptionDispatcher.ExceptionSchema( + url=response_json.get("Url") or url, + object_type=exception_type, + message=response_json.get("Message", ""), + error=response_json.get("Error", ""), + ) + raise ExceptionDispatcher.get(schema, response.status_code, json_body=response_json) self.result = ConversationResult.from_json(response_json) return ResponseDisposeHandling.AUTOMATIC # Non-final lines are JSON-encoded chunks (e.g. "\\\"chunk\\\""). diff --git a/ravendb/tests/ai_agent_tests/test_ai_conversation_streaming_errors.py b/ravendb/tests/ai_agent_tests/test_ai_conversation_streaming_errors.py new file mode 100644 index 00000000..e9d0ad28 --- /dev/null +++ b/ravendb/tests/ai_agent_tests/test_ai_conversation_streaming_errors.py @@ -0,0 +1,99 @@ +""" +Unit tests for errors written into a streamed conversation response (RavenDB-25401). + +When the server fails after streaming already started (HTTP 200), it writes the standard error payload +as the final line of the stream. The client must raise it instead of parsing it as the conversation result. +No server is needed: the streamed response is faked. +""" + +import json +import unittest + +from ravendb.documents.operations.ai.agents.run_conversation_operation import RunConversationCommand +from ravendb.exceptions.raven_exceptions import AiException, RefusedToAnswerException + +EXPECTED_REFUSAL = "I'm sorry, I can't help with that." + + +class _FakeStreamingResponse: + def __init__(self, lines, status_code=200): + self._lines = lines + self.status_code = status_code + + def iter_lines(self, decode_unicode=False): + return iter(self._lines) + + +def _streaming_command(chunks): + return RunConversationCommand( + agent_id="agents/1", + conversation_id="conversations/1", + stream_property_path="Answer", + streamed_chunks_callback=chunks.append, + ) + + +class TestAiConversationStreamingErrors(unittest.TestCase): + def test_error_payload_in_stream_raises_dispatched_exception(self): + error_payload = { + "Url": "/databases/db1/ai/agent", + "Type": "Raven.Client.Exceptions.AI.AiException", + "Message": "Failed to communicate with the agent", + "Error": ( + "Raven.Client.Exceptions.AI.AiException: Failed to communicate with the agent" + " ---> Raven.Client.Exceptions.AI.RefusedToAnswerException: " + f"The model refused to answer: {EXPECTED_REFUSAL}" + ), + "RequestId": "req-refusal-123", + } + chunks = [] + command = _streaming_command(chunks) + response = _FakeStreamingResponse([json.dumps("partial"), "", json.dumps(error_payload)]) + + with self.assertRaises(AiException) as ctx: + command.process_response(None, response, "http://localhost:8080/databases/db1/ai/agent") + + self.assertIs(AiException, type(ctx.exception)) + self.assertIn("RefusedToAnswerException", str(ctx.exception)) + self.assertIn(EXPECTED_REFUSAL, str(ctx.exception)) + self.assertEqual("req-refusal-123", ctx.exception.request_id) + self.assertEqual(["partial"], chunks) + self.assertIsNone(command.result) + + def test_refused_to_answer_payload_in_stream_raises_refused_to_answer_exception(self): + error_payload = { + "Type": "Raven.Client.Exceptions.AI.RefusedToAnswerException", + "Message": EXPECTED_REFUSAL, + "Error": f"Raven.Client.Exceptions.AI.RefusedToAnswerException: {EXPECTED_REFUSAL}", + "Refusal": EXPECTED_REFUSAL, + "FinishReason": "content_filter", + } + command = _streaming_command([]) + response = _FakeStreamingResponse([json.dumps(error_payload)]) + + with self.assertRaises(RefusedToAnswerException) as ctx: + command.process_response(None, response, "http://localhost:8080/databases/db1/ai/agent") + + self.assertEqual(EXPECTED_REFUSAL, ctx.exception.refusal) + self.assertEqual("content_filter", ctx.exception.finish_reason) + + def test_final_result_in_stream_is_parsed(self): + final = { + "ConversationId": "conversations/1", + "ChangeVector": "A:1", + "Response": {"Answer": "streamed answer"}, + "ActionRequests": [], + } + chunks = [] + command = _streaming_command(chunks) + response = _FakeStreamingResponse([json.dumps("streamed "), json.dumps("answer"), json.dumps(final)]) + + command.process_response(None, response, "http://localhost:8080/databases/db1/ai/agent") + + self.assertEqual(["streamed ", "answer"], chunks) + self.assertEqual("conversations/1", command.result.conversation_id) + self.assertEqual({"Answer": "streamed answer"}, command.result.response) + + +if __name__ == "__main__": + unittest.main()