fix(benchmark): account for failed requests and incomplete streams (#19498)

test/benchmark/README.md

@@ -165,6 +165,56 @@ What this CLI can do
   - It is not a general admin CLI; there are no standalone "create-only" or
     "manage" commands. Use the reports to capture created IDs for reuse.
 
+## Interpreting failures and throughput
+
+Each measured request contributes one success or failure, including when using
+`--concurrency`. Connection failures, timeouts, non-2xx HTTP responses, invalid
+response payloads, and API errors are recorded as failed samples so the remaining
+iterations can run and produce a report. There are no automatic retries: a retry
+would hide the original failure and change the measured workload. Authentication
+and dataset/chat setup are outside the measured requests; setup failures still
+stop the command.
+
+A successful chat sample requires non-empty assistant content **and** a stream
+completion marker (`[DONE]` or a non-empty `finish_reason`). Receiving some text
+before an unexpected EOF is a failed, partial response. Partial text remains
+available with `--print-response`, but failed samples are excluded from both
+first-token and total-latency statistics. Completion measures protocol success,
+not answer quality: a `length` finish reason can still indicate an answer limited
+by the model's token budget. An empty retrieval result with `code: 0` is a
+successful API request, not evidence of good retrieval quality.
+
+Reports distinguish:
+
+| JSON field | Definition |
+| --- | --- |
+| `qps` | All measured requests / measured wall-clock duration |
+| `success_qps` | Successful requests / measured wall-clock duration |
+| `failure_rate` | Failed requests / all measured requests, in the range 0–1 |
+
+For example, if 10 measured requests finish in 5 seconds but only 6 succeed,
+`qps` is 2, `success_qps` is 1.2, and `failure_rate` is 0.4. Comparing only QPS
+can make a server that quickly rejects requests look faster than a healthy one.
+The text report shows the same distinction and renders failure rate as a
+percentage. If all requests fail, successful QPS is zero and successful latency
+statistics are unavailable (`null` in JSON / `n/a` in text).
+
+The command exits with status 1 if any measured request fails and 0 if all
+measured requests succeed. Unexpected programming errors are not converted into
+request failures.
+
+### Testing the benchmark client
+
+From the repository root, with the test dependencies installed:
+
+```sh
+uv run --group test pytest test/benchmark/tests
+```
+
+These tests use response fixtures and a loopback HTTP server. They also invoke
+the actual CLI with sequential and multiprocessing workloads. No RAGFlow server,
+database, model, or external credentials are required.
+
 Do I need the dataset ID?
   - If the CLI creates a dataset, it uses the returned dataset ID internally.
     You do not need to supply it for that same run.

test/benchmark/chat.py

@@ -2,6 +2,8 @@ import json
 import time
 from typing import Any, Dict, List, Optional
 
+import requests
+
 from .http_client import HttpClient
 from .metrics import ChatSample
 
@@ -55,15 +57,18 @@ def resolve_model(model: Optional[str], chat_data: Optional[Dict[str, Any]]) ->
 
 
 def _parse_stream_error(response) -> Optional[str]:
+    """Reject HTTP failures and responses that cannot contain a chat stream."""
+    if not 200 <= response.status_code < 300:
+        return f"HTTP {response.status_code}"
     content_type = response.headers.get("Content-Type", "")
     if "text/event-stream" in content_type:
         return None
     try:
         payload = response.json()
-    except Exception:
+    except ValueError:
         return f"Unexpected non-stream response (status {response.status_code})"
-    if payload.get("code") not in (0, None):
-        return payload.get("message", "Unknown error")
+    if isinstance(payload, dict) and payload.get("code") not in (0, None):
+        return str(payload.get("message") or f"Chat failed (code {payload['code']})")
     return f"Unexpected non-stream response (status {response.status_code})"
 
 
@@ -74,26 +79,27 @@ def stream_chat_completion(
     messages: List[Dict[str, Any]],
     extra_body: Optional[Dict[str, Any]] = None,
 ) -> ChatSample:
+    """Measure one text completion, retaining partial text when the stream fails."""
     payload: Dict[str, Any] = {"model": model, "messages": messages, "stream": True}
     if extra_body:
         payload["extra_body"] = extra_body
     t0 = time.perf_counter()
-    response = client.request(
-        "POST",
-        f"/openai/{chat_id}/chat/completions",
-        json_body=payload,
-        stream=True,
-    )
-    error = _parse_stream_error(response)
-    if error:
-        response.close()
-        return ChatSample(t0=t0, t1=None, t2=None, error=error)
-
+    response = None
     t1: Optional[float] = None
     t2: Optional[float] = None
     stream_error: Optional[str] = None
+    completed = False
     content_parts: List[str] = []
     try:
+        response = client.request(
+            "POST",
+            f"/openai/{chat_id}/chat/completions",
+            json_body=payload,
+            stream=True,
+        )
+        error = _parse_stream_error(response)
+        if error:
+            return ChatSample(t0=t0, t1=None, t2=time.perf_counter(), error=error)
         for raw_line in response.iter_lines(decode_unicode=True):
             if raw_line is None:
                 continue
@@ -104,17 +110,33 @@ def stream_chat_completion(
             if not data:
                 continue
             if data == "[DONE]":
+                completed = True
                 t2 = time.perf_counter()
                 break
             try:
                 chunk = json.loads(data)
-            except Exception as exc:
+            except ValueError as exc:
                 stream_error = f"Invalid JSON chunk: {exc}"
                 t2 = time.perf_counter()
                 break
-            choices = chunk.get("choices") or []
-            choice = choices[0] if choices else {}
-            delta = choice.get("delta") or {}
+            if not isinstance(chunk, dict):
+                stream_error = "Invalid stream chunk: expected an object"
+                break
+            if chunk.get("error") is not None or chunk.get("code") not in (0, None):
+                stream_error = "Chat stream reported an error"
+                break
+            choices = chunk.get("choices")
+            if not isinstance(choices, list) or not all(isinstance(item, dict) for item in choices):
+                stream_error = "Invalid stream chunk: expected choices to be a list of objects"
+                break
+            # Usage-only chunks can have no choices; they do not complete the stream.
+            if not choices:
+                continue
+            if not all(isinstance(item.get("delta"), dict) for item in choices):
+                stream_error = "Invalid stream chunk: expected delta to be an object"
+                break
+            choice = choices[0]
+            delta = choice["delta"]
             content = delta.get("content")
             if t1 is None and isinstance(content, str) and content != "":
                 t1 = time.perf_counter()
@@ -122,10 +144,14 @@ def stream_chat_completion(
                 content_parts.append(content)
             finish_reason = choice.get("finish_reason")
             if finish_reason:
+                completed = True
                 t2 = time.perf_counter()
                 break
+    except requests.RequestException as exc:
+        stream_error = f"Transport error ({type(exc).__name__})"
     finally:
-        response.close()
+        if response is not None:
+            response.close()
 
     if t2 is None:
         t2 = time.perf_counter()
@@ -134,4 +160,6 @@ def stream_chat_completion(
         return ChatSample(t0=t0, t1=t1, t2=t2, error=stream_error, response_text=response_text)
     if t1 is None:
         return ChatSample(t0=t0, t1=None, t2=t2, error="No assistant content received", response_text=response_text)
+    if not completed:
+        return ChatSample(t0=t0, t1=t1, t2=t2, error="Stream ended before a completion marker", response_text=response_text)
     return ChatSample(t0=t0, t1=t1, t2=t2, error=None, response_text=response_text)

test/benchmark/cli.py

@@ -22,7 +22,7 @@ from .dataset import (
     wait_for_parse_done,
 )
 from .http_client import HttpClient
-from .metrics import ChatSample, RetrievalSample, summarize
+from .metrics import ChatSample, RetrievalSample, request_rates, summarize
 from .report import chat_report, retrieval_report
 from .retrieval import RetrievalError, build_payload, run_retrieval as run_retrieval_request
 from .utils import eprint, load_json_arg, split_csv
@@ -457,7 +457,7 @@ def run_chat(client: HttpClient, args: argparse.Namespace) -> int:
             "errors": [e for e in errors if e],
             "created": created,
             "total_duration_s": total_duration,
-            "qps": (args.iterations / total_duration) if total_duration > 0 else None,
+            **request_rates(success, failure, total_duration),
         }
         if args.print_response:
             payload["responses"] = responses
@@ -557,7 +557,7 @@ def run_retrieval(client: HttpClient, args: argparse.Namespace) -> int:
             "errors": [e for e in errors if e],
             "created": created,
             "total_duration_s": total_duration,
-            "qps": (args.iterations / total_duration) if total_duration > 0 else None,
+            **request_rates(success, failure, total_duration),
         }
         if args.print_response:
             payload["responses"] = responses

test/benchmark/metrics.py

@@ -65,3 +65,14 @@ def summarize(values: List[float]) -> dict:
         "p90": _percentile(sorted_vals, 90),
         "p95": _percentile(sorted_vals, 95),
     }
+
+
+def request_rates(success: int, failure: int, total_duration_s: float | None) -> dict:
+    """Separate attempted throughput from successfully completed work."""
+    total = success + failure
+    timed = total_duration_s is not None and total_duration_s > 0
+    return {
+        "qps": total / total_duration_s if timed else None,
+        "success_qps": success / total_duration_s if timed else None,
+        "failure_rate": failure / total if total else None,
+    }

test/benchmark/report.py

@@ -1,5 +1,7 @@
 from typing import Dict, List, Optional
 
+from .metrics import request_rates
+
 
 def _fmt_seconds(value: Optional[float]) -> str:
     if value is None:
@@ -14,15 +16,19 @@ def _fmt_ms(value: Optional[float]) -> str:
 
 
 def _fmt_qps(qps: Optional[float]) -> str:
-    if qps is None or qps <= 0:
+    if qps is None:
         return "n/a"
     return f"{qps:.2f}"
 
 
-def _calc_qps(total_duration_s: Optional[float], total_requests: int) -> Optional[float]:
-    if total_duration_s is None or total_duration_s <= 0:
-        return None
-    return total_requests / total_duration_s
+def _request_rate_lines(success: int, failure: int, total_duration_s: float | None) -> list[str]:
+    rates = request_rates(success, failure, total_duration_s)
+    failure_rate = "n/a" if rates["failure_rate"] is None else f"{rates['failure_rate']:.2%}"
+    return [
+        f"QPS (requests / total duration): {_fmt_qps(rates['qps'])}",
+        f"Successful QPS (success / total duration): {_fmt_qps(rates['success_qps'])}",
+        f"Failure Rate: {failure_rate}",
+    ]
 
 
 def render_report(lines: List[str]) -> str:
@@ -62,7 +68,7 @@ def chat_report(
             f"avg={_fmt_ms(first_token_stats['avg'])}, min={_fmt_ms(first_token_stats['min'])}, "
             f"p50={_fmt_ms(first_token_stats['p50'])}, p90={_fmt_ms(first_token_stats['p90'])}, p95={_fmt_ms(first_token_stats['p95'])}",
             f"Total Duration: {_fmt_seconds(total_duration_s)}",
-            f"QPS (requests / total duration): {_fmt_qps(_calc_qps(total_duration_s, iterations))}",
+            *_request_rate_lines(success, failure, total_duration_s),
         ]
     )
     if errors:
@@ -95,7 +101,7 @@ def retrieval_report(
         [
             f"Latency: avg={_fmt_ms(stats['avg'])}, min={_fmt_ms(stats['min'])}, p50={_fmt_ms(stats['p50'])}, p90={_fmt_ms(stats['p90'])}, p95={_fmt_ms(stats['p95'])}",
             f"Total Duration: {_fmt_seconds(total_duration_s)}",
-            f"QPS (requests / total duration): {_fmt_qps(_calc_qps(total_duration_s, iterations))}",
+            *_request_rate_lines(success, failure, total_duration_s),
         ]
     )
     if errors:

test/benchmark/retrieval.py

@@ -1,6 +1,8 @@
 import time
 from typing import Any, Dict, List, Optional
 
+import requests
+
 from .http_client import HttpClient
 from .metrics import RetrievalSample
 
@@ -26,14 +28,27 @@ def build_payload(
 
 
 def run_retrieval(client: HttpClient, payload: Dict[str, Any]) -> RetrievalSample:
+    """Measure one retrieval, recording transport and API failures as samples."""
     t0 = time.perf_counter()
-    response = client.request("POST", "/retrieval", json_body=payload, stream=False)
-    raw = response.content
-    t1 = time.perf_counter()
+    response = None
     try:
-        res = client.parse_json_bytes(raw)
-    except Exception as exc:
-        return RetrievalSample(t0=t0, t1=t1, error=f"Invalid JSON response: {exc}")
+        response = client.request("POST", "/retrieval", json_body=payload, stream=False)
+        raw = response.content
+        t1 = time.perf_counter()
+        if not 200 <= response.status_code < 300:
+            return RetrievalSample(t0=t0, t1=t1, error=f"HTTP {response.status_code}")
+        try:
+            res = client.parse_json_bytes(raw)
+        except ValueError as exc:
+            return RetrievalSample(t0=t0, t1=t1, error=f"Invalid JSON response: {exc}")
+    except requests.RequestException as exc:
+        return RetrievalSample(t0=t0, t1=time.perf_counter(), error=f"Transport error ({type(exc).__name__})")
+    finally:
+        if response is not None:
+            response.close()
+    if not isinstance(res, dict):
+        return RetrievalSample(t0=t0, t1=t1, error="Invalid retrieval response: expected an object")
     if res.get("code") != 0:
-        return RetrievalSample(t0=t0, t1=t1, error=res.get("message"), response=res)
+        error = str(res.get("message") or f"Retrieval failed (code {res.get('code')!r})")
+        return RetrievalSample(t0=t0, t1=t1, error=error, response=res)
     return RetrievalSample(t0=t0, t1=t1, error=None, response=res)

test/benchmark/tests/test_cli_failures.py

@@ -0,0 +1,100 @@
+"""Exercise the real CLI and process workers against a loopback HTTP fixture."""
+
+import json
+import os
+import subprocess
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def benchmark_server():
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *args):
+            pass
+
+        def send_body(self, body, status=200, content_type="application/json"):
+            self.send_response(status)
+            self.send_header("Content-Type", content_type)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def do_GET(self):
+            self.send_body(b'{"code":0,"data":{"id":"chat","llm_id":"fixture-model"}}')
+
+        def do_POST(self):
+            self.rfile.read(int(self.headers.get("Content-Length", "0")))
+            with self.server.lock:
+                index = self.server.requests
+                self.server.requests += 1
+            if index == 4:
+                # Close without response headers: requests raises ConnectionError.
+                self.close_connection = True
+                return
+            if self.path.endswith("/retrieval"):
+                bodies = [b'{"code":0,"data":{"chunks":[]}}', b'{"code":102}', b'{"code":0}', b"[]"]
+                self.send_body(bodies[index], status=503 if index == 2 else 200)
+            else:
+                content = b'data: {"choices":[{"delta":{"content":"answer"}}]}\n\n'
+                bodies = [content + b"data: [DONE]\n\n", content, content + b"data: [DONE]\n\n", content + b'data: {"error":{"message":"failed"}}\n\n']
+                self.send_body(bodies[index], status=503 if index == 2 else 200, content_type="text/event-stream; charset=utf-8")
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    server.requests = 0
+    server.lock = threading.Lock()
+    worker = threading.Thread(target=server.serve_forever, daemon=True)
+    worker.start()
+    try:
+        yield server
+    finally:
+        server.shutdown()
+        worker.join()
+        server.server_close()
+
+
+@pytest.mark.parametrize("command", ["chat", "retrieval"])
+@pytest.mark.parametrize("concurrency", [1, 2])
+def test_cli_finishes_mixed_workload_and_reports_only_successful_latency(benchmark_server, command, concurrency):
+    repo = Path(__file__).resolve().parents[3]
+    args = [
+        sys.executable,
+        "-m",
+        "benchmark",
+        command,
+        "--base-url",
+        f"http://127.0.0.1:{benchmark_server.server_port}",
+        "--api-key",
+        "local-fixture-token",
+        "--iterations",
+        "5",
+        "--concurrency",
+        str(concurrency),
+        "--json",
+        "--print-response",
+    ]
+    if command == "chat":
+        args += ["--chat-id", "chat", "--model", "fixture-model", "--message", "hello"]
+    else:
+        args += ["--dataset-id", "dataset", "--question", "hello"]
+    env = {**os.environ, "PYTHONPATH": str(repo / "test"), "NO_PROXY": "127.0.0.1", "no_proxy": "127.0.0.1"}
+    result = subprocess.run(args, cwd=repo, env=env, capture_output=True, text=True, timeout=30, check=False)
+    assert result.returncode == 1, result.stderr
+    report = json.loads(result.stdout)
+    assert benchmark_server.requests == 5
+    assert report["success"] == 1
+    assert report["failure"] == 4
+    assert report["failure_rate"] == 0.8
+    assert report["success_qps"] == pytest.approx(report["qps"] / 5)
+    assert len(report["errors"]) == 4
+    assert len(report["responses"]) == 5
+    latency = "total_latency" if command == "chat" else "latency"
+    assert report[latency]["count"] == 1
+    if command == "chat":
+        assert report["first_token_latency"]["count"] == 1
+        assert any("[error]" in response and "answer" in response for response in report["responses"])
+    assert "Traceback" not in result.stderr

test/benchmark/tests/test_reports.py

@@ -0,0 +1,30 @@
+import pytest
+
+from test.benchmark.metrics import request_rates, summarize
+from test.benchmark.report import chat_report, retrieval_report
+
+
+def test_request_rates_separate_attempts_and_successes():
+    assert request_rates(2, 3, 5) == {"qps": 1.0, "success_qps": 0.4, "failure_rate": 0.6}
+
+
+def test_zero_success_is_zero_throughput_not_unavailable():
+    assert request_rates(0, 4, 2) == {"qps": 2.0, "success_qps": 0.0, "failure_rate": 1.0}
+
+
+@pytest.mark.parametrize("duration", [None, 0])
+def test_missing_duration_does_not_invent_throughput(duration):
+    assert request_rates(0, 0, duration) == {"qps": None, "success_qps": None, "failure_rate": None}
+
+
+@pytest.mark.parametrize("command", ["chat", "retrieval"])
+def test_text_report_shows_all_failed_run(command):
+    kwargs = {"interface": command, "concurrency": 2, "total_duration_s": 2, "iterations": 4, "success": 0, "failure": 4, "errors": ["HTTP 503"], "created": {}}
+    stats = summarize([])
+    if command == "chat":
+        report = chat_report(**kwargs, model="model", total_stats=stats, first_token_stats=stats)
+    else:
+        report = retrieval_report(**kwargs, stats=stats)
+    assert "Successful QPS (success / total duration): 0.00" in report
+    assert "Failure Rate: 100.00%" in report
+    assert "avg=n/a" in report

test/benchmark/tests/test_samples.py

@@ -0,0 +1,190 @@
+"""Request failures must become samples, not successful or missing measurements."""
+
+import json
+from unittest.mock import Mock
+
+import pytest
+import requests
+
+from test.benchmark.chat import stream_chat_completion
+from test.benchmark.http_client import HttpClient
+from test.benchmark.retrieval import run_retrieval
+
+
+def stream_response(events, status=200):
+    response = Mock(spec=requests.Response)
+    response.status_code = status
+    response.headers = {"Content-Type": "text/event-stream"}
+
+    def lines(**kwargs):
+        for event in events:
+            if isinstance(event, Exception):
+                raise event
+            yield "data: " + (event if isinstance(event, str) else json.dumps(event))
+
+    response.iter_lines.side_effect = lines
+    return response
+
+
+CONTENT = {"choices": [{"delta": {"content": "partial answer"}}]}
+
+
+def test_chat_eof_after_content_is_failure():
+    client = Mock(spec=HttpClient)
+    client.request.return_value = stream_response([CONTENT])
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error
+    assert sample.response_text == "partial answer"
+    client.request.return_value.close.assert_called_once()
+
+
+@pytest.mark.parametrize("failure", [requests.ConnectTimeout(), requests.ConnectionError()])
+def test_connection_failure_returns_chat_sample(failure):
+    client = Mock(spec=HttpClient)
+    client.request.side_effect = failure
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error
+    assert sample.first_token_latency is None
+
+
+def test_midstream_timeout_retains_partial_response():
+    client = Mock(spec=HttpClient)
+    client.request.return_value = stream_response([CONTENT, requests.ReadTimeout()])
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error
+    assert sample.response_text == "partial answer"
+    assert sample.first_token_latency is not None
+    client.request.return_value.close.assert_called_once()
+
+
+@pytest.mark.parametrize("failure", [requests.ReadTimeout(), requests.ConnectionError()])
+def test_connection_failure_returns_retrieval_sample(failure):
+    client = Mock(spec=HttpClient)
+    client.request.side_effect = failure
+    sample = run_retrieval(client, {"question": "question", "dataset_ids": ["dataset"]})
+    assert sample.error
+
+
+def test_retrieval_error_without_message_is_failure():
+    response = requests.Response()
+    response.status_code = 200
+    response._content = b'{"code": 102}'
+    client = HttpClient("http://unused.invalid")
+    client.request = Mock(return_value=response)
+    sample = run_retrieval(client, {})
+    assert sample.error
+
+
+@pytest.mark.parametrize("ending", ["[DONE]", {"choices": [{"delta": {}, "finish_reason": "stop"}]}])
+def test_completed_chat_remains_successful(ending):
+    client = Mock(spec=HttpClient)
+    client.request.return_value = stream_response([CONTENT, ending])
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error is None
+    assert sample.response_text == "partial answer"
+    assert sample.total_latency >= sample.first_token_latency >= 0
+    client.request.return_value.close.assert_called_once()
+
+
+@pytest.mark.parametrize(
+    "event",
+    [
+        "not-json",
+        [],
+        None,
+        {},
+        *({"choices": value} for value in [None, False, 0, "", {}, [None], [{}, "invalid"]]),
+        {"choices": {"delta": {}}},
+        *({"choices": [{"delta": value}]} for value in [None, False, 0, "", [], "bad"]),
+        {"choices": [{}]},
+        {"choices": [{"delta": {}}, {"delta": None}]},
+        {"error": {"message": "unavailable"}},
+        {"code": 102},
+    ],
+)
+def test_malformed_or_error_event_after_content_is_failure(event):
+    client = Mock(spec=HttpClient)
+    client.request.return_value = stream_response([CONTENT, event, "[DONE]"])
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error
+    assert sample.response_text == "partial answer"
+    client.request.return_value.close.assert_called_once()
+
+
+def test_chat_http_error_cannot_be_hidden_by_valid_sse_body():
+    client = Mock(spec=HttpClient)
+    response = stream_response([CONTENT, "[DONE]"], status=503)
+    client.request.return_value = response
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error == "HTTP 503"
+    response.iter_lines.assert_not_called()
+    response.close.assert_called_once()
+
+
+@pytest.mark.parametrize("payload", [[], None, {"code": 102}, {"code": 102, "message": ""}])
+def test_non_stream_chat_errors_do_not_escape(payload):
+    client = Mock(spec=HttpClient)
+    response = Mock(spec=requests.Response)
+    response.status_code = 200
+    response.headers = {"Content-Type": "application/json"}
+    response.json.return_value = payload
+    client.request.return_value = response
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error
+    response.close.assert_called_once()
+
+
+@pytest.mark.parametrize("status,payload", [(503, {"code": 0}), (200, []), (200, None), (200, {}), (200, {"code": 102, "message": ""})])
+def test_invalid_retrieval_response_is_failure(status, payload):
+    response = requests.Response()
+    response.status_code = status
+    response._content = json.dumps(payload).encode()
+    client = HttpClient("http://unused.invalid")
+    client.request = Mock(return_value=response)
+    sample = run_retrieval(client, {})
+    assert sample.error
+
+
+def test_successful_empty_retrieval_is_not_a_transport_failure():
+    response = requests.Response()
+    response.status_code = 200
+    response._content = b'{"code": 0, "data": {"chunks": []}}'
+    client = HttpClient("http://unused.invalid")
+    client.request = Mock(return_value=response)
+    sample = run_retrieval(client, {})
+    assert sample.error is None
+    assert sample.response["data"]["chunks"] == []
+
+
+def test_empty_completed_chat_still_fails():
+    client = Mock(spec=HttpClient)
+    client.request.return_value = stream_response(["[DONE]"])
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error == "No assistant content received"
+
+
+def test_programming_errors_are_not_swallowed():
+    client = Mock(spec=HttpClient)
+    client.request.side_effect = RuntimeError("unexpected bug")
+    with pytest.raises(RuntimeError, match="unexpected bug"):
+        stream_chat_completion(client, "chat", "model", [])
+    with pytest.raises(RuntimeError, match="unexpected bug"):
+        run_retrieval(client, {})
+
+
+@pytest.mark.parametrize(
+    "event",
+    [
+        {"choices": [], "usage": {"total_tokens": 4}},
+        {"choices": [{"delta": {"role": "assistant", "content": None}}]},
+        {"choices": [{"delta": {}}, {"delta": {}}]},
+    ],
+)
+def test_valid_metadata_chunks_preserve_success(event):
+    """Usage-only and empty deltas must not invalidate completed text streams."""
+    client = Mock(spec=HttpClient)
+    client.request.return_value = stream_response([CONTENT, event, "[DONE]"])
+    sample = stream_chat_completion(client, "chat", "model", [])
+    assert sample.error is None
+    assert sample.response_text == "partial answer"
+    client.request.return_value.close.assert_called_once()