Return automation errors before streaming

Author: Damue01

app2api/api.py

@@ -220,35 +220,6 @@ async def create_chat_completion(payload: ChatCompletionRequest):
                 status_code=400,
             )
 
-        if payload.stream:
-
-            async def generate_chat_stream():
-                job = await execute_protocol(
-                    target,
-                    question,
-                    payload.user,
-                    "openai-chat-completions",
-                )
-                if job.status != JobStatus.SUCCEEDED or job.answer is None:
-                    yield sse_data(
-                        error_body(
-                            job.error or "application automation failed",
-                            "server_error",
-                            "automation_failed",
-                        )
-                    )
-                    yield "data: [DONE]\n\n"
-                    return
-                for chunk in chat_completion_chunks(job, payload.model):
-                    yield sse_data(chunk)
-                yield "data: [DONE]\n\n"
-
-            return StreamingResponse(
-                generate_chat_stream(),
-                media_type="text/event-stream",
-                headers={"Cache-Control": "no-cache"},
-            )
-
         job = await execute_protocol(
             target,
             question,
@@ -264,6 +235,20 @@ async def generate_chat_stream():
                 ),
                 status_code=502,
             )
+
+        if payload.stream:
+
+            async def generate_chat_stream():
+                for chunk in chat_completion_chunks(job, payload.model):
+                    yield sse_data(chunk)
+                yield "data: [DONE]\n\n"
+
+            return StreamingResponse(
+                generate_chat_stream(),
+                media_type="text/event-stream",
+                headers={"Cache-Control": "no-cache"},
+            )
+
         return chat_completion(job, payload.model)
 
     @app.post("/v1/responses")
@@ -292,34 +277,6 @@ async def create_response(payload: ResponsesRequest):
                 status_code=400,
             )
 
-        if payload.stream:
-
-            async def generate_response_stream():
-                job = await execute_protocol(
-                    target,
-                    question,
-                    payload.user,
-                    "openai-responses",
-                )
-                if job.status != JobStatus.SUCCEEDED or job.answer is None:
-                    yield response_sse(
-                        "error",
-                        error_body(
-                            job.error or "application automation failed",
-                            "server_error",
-                            "automation_failed",
-                        ),
-                    )
-                    return
-                for event, value in responses_stream_events(job, payload):
-                    yield response_sse(event, value)
-
-            return StreamingResponse(
-                generate_response_stream(),
-                media_type="text/event-stream",
-                headers={"Cache-Control": "no-cache"},
-            )
-
         job = await execute_protocol(
             target,
             question,
@@ -335,6 +292,19 @@ async def generate_response_stream():
                 ),
                 status_code=502,
             )
+
+        if payload.stream:
+
+            async def generate_response_stream():
+                for event, value in responses_stream_events(job, payload):
+                    yield response_sse(event, value)
+
+            return StreamingResponse(
+                generate_response_stream(),
+                media_type="text/event-stream",
+                headers={"Cache-Control": "no-cache"},
+            )
+
         return response_object(job, payload)
 
     @app.post("/v1/messages")
@@ -365,26 +335,23 @@ async def create_anthropic_message(payload: AnthropicMessagesRequest):
         if user is not None and not isinstance(user, str):
             user = str(user)
 
+        job = await execute_protocol(
+            target,
+            question,
+            user,
+            "anthropic-messages",
+        )
+        if job.status != JobStatus.SUCCEEDED or job.answer is None:
+            return anthropic_json_error(
+                job.error or "application automation failed",
+                502,
+                "api_error",
+            )
+
         if payload.stream:
             request_id = anthropic_request_id()
 
             async def generate_anthropic_stream():
-                job = await execute_protocol(
-                    target,
-                    question,
-                    user,
-                    "anthropic-messages",
-                )
-                if job.status != JobStatus.SUCCEEDED or job.answer is None:
-                    yield anthropic_sse(
-                        "error",
-                        anthropic_error_body(
-                            job.error or "application automation failed",
-                            "api_error",
-                            request_id,
-                        ),
-                    )
-                    return
                 for event, value in anthropic_stream_events(job, payload):
                     yield anthropic_sse(event, value)
 
@@ -397,18 +364,6 @@ async def generate_anthropic_stream():
                 },
             )
 
-        job = await execute_protocol(
-            target,
-            question,
-            user,
-            "anthropic-messages",
-        )
-        if job.status != JobStatus.SUCCEEDED or job.answer is None:
-            return anthropic_json_error(
-                job.error or "application automation failed",
-                502,
-                "api_error",
-            )
         request_id = anthropic_request_id()
         return JSONResponse(
             anthropic_message_object(job, payload),

tests/test_openai_compat.py

@@ -207,3 +207,33 @@ def test_xiaohuoren_is_a_separate_openai_model(tmp_path):
             item for item in models["models"] if item["slug"] == "douyin_xiaohuoren"
         )
         assert catalog["display_name"] == "Douyin Xiaohuoren"
+
+
+def test_responses_stream_returns_http_error_before_sse_on_automation_failure(
+    tmp_path,
+):
+    app = create_app(
+        Settings(
+            _env_file=None,
+            driver="mock",
+            enabled_targets="meituan",
+            config_dir=tmp_path / "missing-config",
+            artifact_dir=tmp_path / "artifacts",
+            recording_enabled=False,
+            job_timeout_seconds=5,
+        )
+    )
+    with TestClient(app) as client:
+        response = client.post(
+            "/v1/responses",
+            json={
+                "model": "meituan_xiaotuan",
+                "input": "这个请求不应开启不完整的 SSE 流",
+                "stream": True,
+            },
+        )
+
+    assert response.status_code == 502
+    assert response.headers["content-type"].startswith("application/json")
+    assert response.json()["error"]["code"] == "automation_failed"
+    assert "缺少目标流程配置" in response.json()["error"]["message"]