test(e2e): make the streaming spend test judge its own stream

test_streaming_chat_completion_tracks_spend asserted only result.ok, which for
a stream is just the HTTP 200 the proxy commits before the upstream finishes. A
stream that breaks mid-flight comes back as a 200 carrying an SSE error frame
and no [DONE], so the only symptom was a missing spend row E2E_POLL_TIMEOUT
later, blaming spend tracking for a failure on the streaming path. It now
asserts is_streaming, chunks, stream_error and stream_done, the set its passing
sibling already used and the last streaming test in the suite that did not

ProxyClient's spend-log poller had the matching hole on the read side: every
non-success /spend/logs outcome became an empty list, so a window where the
endpoint 500'd or timed out on every read was indistinguishable from a row that
was never written. One successful read anywhere in the window still trusts the
emptiness, so a transient blip polls through, but a window with no successful
read now raises and carries the read that failed

tests/e2e/proxy_client.py

@@ -443,14 +443,16 @@ class ProxyClient:
 
     # ---- spend read-back ------------------------------------------------
 
-    def spend_logs(self, params: SpendLogsParams) -> list[SpendLogRow]:
-        result = self.transport.get(
+    def _read_spend_logs(self, params: SpendLogsParams) -> Result[SpendLogs]:
+        return self.transport.get(
             "/spend/logs",
             headers=self.transport.master,
             params=params,
             response_type=SpendLogs,
         )
-        match result:
+
+    def spend_logs(self, params: SpendLogsParams) -> list[SpendLogRow]:
+        match self._read_spend_logs(params):
             case Success(data=logs):
                 return logs.root
             case _:
@@ -481,7 +483,7 @@ class ProxyClient:
     def poll_logs_for_key(
         self, key: str, *, min_rows: int = 1, predicate: RowsPredicate | None = None
     ) -> list[SpendLogRow]:
-        return self._poll(lambda: self.spend_logs(SpendLogsParams(api_key=key)), min_rows, predicate)
+        return self._poll(lambda: self._read_spend_logs(SpendLogsParams(api_key=key)), min_rows, predicate)
 
     def poll_logs_for_request_id(
         self,
@@ -491,24 +493,47 @@ class ProxyClient:
         predicate: RowsPredicate | None = None,
     ) -> list[SpendLogRow]:
         return self._poll(
-            lambda: self.spend_logs(SpendLogsParams(request_id=request_id)),
+            lambda: self._read_spend_logs(SpendLogsParams(request_id=request_id)),
             min_rows,
             predicate,
         )
 
     def _poll(
         self,
-        fetch: Callable[[], list[SpendLogRow]],
+        fetch: Callable[[], Result[SpendLogs]],
         min_rows: int,
         predicate: RowsPredicate | None,
     ) -> list[SpendLogRow]:
+        """Poll /spend/logs to the deadline, returning the last rows the proxy served.
+
+        A read that never succeeded is not evidence the rows are absent. /spend/logs
+        500s and times out under load (see test_spend_logs_endpoint_returns_spend), and
+        counting those as "no rows yet" makes an unreadable endpoint indistinguishable
+        from a spend row that was never written - the caller then asserts an empty list
+        and blames the write path for a failure on the read path. A single successful
+        read anywhere in the window is enough to trust the emptiness, so a transient
+        blip still polls through; a window where every read failed raises instead.
+        """
         deadline = time.monotonic() + self.poll_timeout
         rows: list[SpendLogRow] = []
+        last_failure: Result[SpendLogs] | None = None
+        served = False
         while time.monotonic() < deadline:
-            rows = fetch()
-            if len(rows) >= min_rows and (predicate is None or predicate(rows)):
-                return rows
+            match fetch():
+                case Success(data=logs):
+                    served = True
+                    rows = logs.root
+                    if len(rows) >= min_rows and (predicate is None or predicate(rows)):
+                        return rows
+                case failure:
+                    last_failure = failure
             time.sleep(self.poll_interval)
+        if not served and last_failure is not None:
+            raise AssertionError(
+                f"/spend/logs never returned a successful read in {self.poll_timeout}s of "
+                f"polling, so an empty result says nothing about the spend rows. "
+                f"Last read: {last_failure}"
+            )
         return rows
 
     # ---- route probe ----------------------------------------------------

tests/e2e/quota_management/spend_tracking/test_spend_tracking_e2e.py

@@ -106,9 +106,26 @@ def test_streaming_chat_completion_tracks_spend(
         f"count to three {unique_marker()}",
         max_tokens=64,
     )
+    # A stream commits its HTTP 200 before the upstream completes, so result.ok alone
+    # says nothing about what the stream did: a break mid-flight comes back as a 200
+    # carrying an SSE error frame and no [DONE] (async_data_generator's except arm).
+    # Judge the stream on its own terms here, or a broken stream is only noticed
+    # E2E_POLL_TIMEOUT later as a missing spend row, blaming the wrong subsystem.
     assert (
         result.ok
     ), f"stream failed (status {result.status_code}): {result.body[:300]}"
+    assert result.is_streaming, (
+        f"expected an SSE stream from /chat/completions, got content-type "
+        f"{result.content_type!r}"
+    )
+    assert result.chunks > 0, "no SSE events were consumed from the stream"
+    assert (
+        result.stream_error is None
+    ), f"the stream carried an error event: {result.stream_error}"
+    assert result.stream_done, (
+        f"the stream never terminated with [DONE], so it broke mid-flight: "
+        f"chunks={result.chunks} last_events={result.stream_events[-2:]}"
+    )
 
     rows = client.poll_logs_for_key(
         scoped_key, predicate=lambda rs: any((r.spend or 0) > 0 for r in rs)

tests/e2e/test_proxy_client.py

@@ -0,0 +1,195 @@
+"""Harness coverage for ProxyClient's spend-log poll contract.
+
+No proxy needed and no ``e2e`` marker: this pins how the poller reports an
+unreadable /spend/logs. The poller is what every spend, budget and logging
+suite reads its assertions off, so conflating "the endpoint answered and there
+are no rows" with "the endpoint never answered" mislabels a read-path outage as
+missing spend, and the resulting failure message accuses the wrong subsystem
+120s after the fact. The transport is injected, so nothing here talks HTTP or
+monkeypatches anything.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator, Sequence
+from dataclasses import dataclass
+
+import pytest
+from pydantic import BaseModel
+
+from e2e_http import (
+    AuthHeaders,
+    BinaryStream,
+    ProbeResult,
+    Result,
+    StreamingResponse,
+    Success,
+    UnknownApiError,
+)
+from models import SpendLogRow, SpendLogs
+from proxy_client import ProxyClient
+
+
+class _UnusedTransport:
+    """Every Transport member the poll path does not touch.
+
+    ProxyClient depends on the whole Transport Protocol, so a fake must satisfy
+    all of it to type-check. Raising keeps the fake honest: a test that reaches
+    one of these is exercising something it did not set up.
+    """
+
+    def post[R: BaseModel](self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]) -> Result[R]:
+        raise NotImplementedError(path)
+
+    def stream(self, path: str, *, headers: BaseModel, json: BaseModel) -> StreamingResponse:
+        raise NotImplementedError(path)
+
+    def stream_binary(
+        self, path: str, *, headers: BaseModel, json: BaseModel, chunk_size: int = 8192
+    ) -> BinaryStream:
+        raise NotImplementedError(path)
+
+    def send(
+        self,
+        path: str,
+        *,
+        headers: BaseModel,
+        json: BaseModel,
+        params: BaseModel | None = None,
+        stream: bool = False,
+    ) -> StreamingResponse:
+        raise NotImplementedError(path)
+
+    def delete[R: BaseModel](
+        self,
+        path: str,
+        *,
+        headers: BaseModel,
+        json: BaseModel,
+        response_type: type[R],
+        params: BaseModel | None = None,
+    ) -> Result[R]:
+        raise NotImplementedError(path)
+
+    def patch[R: BaseModel](
+        self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
+    ) -> Result[R]:
+        raise NotImplementedError(path)
+
+    def put[R: BaseModel](
+        self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]
+    ) -> Result[R]:
+        raise NotImplementedError(path)
+
+    def probe(self, path: str, *, params: BaseModel) -> ProbeResult:
+        raise NotImplementedError(path)
+
+    def upload[R: BaseModel](
+        self,
+        path: str,
+        *,
+        headers: BaseModel,
+        form: BaseModel,
+        filename: str,
+        content: bytes,
+        file_content_type: str = "application/jsonl",
+        file_field: str = "file",
+        params: BaseModel | None = None,
+        response_type: type[R],
+    ) -> Result[R]:
+        raise NotImplementedError(path)
+
+    def download(self, path: str, *, headers: BaseModel) -> StreamingResponse:
+        raise NotImplementedError(path)
+
+    def bearer(self, key: str) -> AuthHeaders:
+        raise NotImplementedError(key)
+
+    @property
+    def master(self) -> AuthHeaders:
+        return AuthHeaders(authorization="Bearer sk-fake-master")
+
+
+@dataclass
+class ScriptedReads(_UnusedTransport):
+    """Serves one scripted /spend/logs outcome per GET, then repeats the last."""
+
+    reads: Sequence[Result[SpendLogs]]
+    calls: int = 0
+
+    def get[R: BaseModel](
+        self,
+        path: str,
+        *,
+        headers: BaseModel,
+        params: BaseModel,
+        response_type: type[R],
+        timeout: float | None = None,
+    ) -> Result[R]:
+        assert path == "/spend/logs", f"fake only serves /spend/logs, got {path}"
+        outcome = self.reads[min(self.calls, len(self.reads) - 1)]
+        self.calls += 1
+        return outcome  # pyright: ignore[reportReturnType]  # scripted per-test as Result[SpendLogs]
+
+
+def _row(request_id: str, spend: float) -> SpendLogRow:
+    return SpendLogRow(request_id=request_id, spend=spend)
+
+
+def _served(rows: Iterator[SpendLogRow] | list[SpendLogRow]) -> Success[SpendLogs]:
+    return Success(status_code=200, data=SpendLogs(list(rows)))
+
+
+def _client(reads: Sequence[Result[SpendLogs]]) -> tuple[ProxyClient, ScriptedReads]:
+    transport = ScriptedReads(reads=reads)
+    # Short budget with no sleep: the contract under test is which outcome the
+    # window produces, not how long it waits.
+    return ProxyClient(transport=transport, poll_timeout=0.05, poll_interval=0.0), transport
+
+
+class TestSpendLogPollDistinguishesUnreadableFromEmpty:
+    def test_rows_are_returned_once_the_predicate_holds(self) -> None:
+        rows = [_row("r1", 0.01)]
+        client, transport = _client([_served([]), _served(rows)])
+
+        polled = client.poll_logs_for_key("sk-test", predicate=lambda rs: any((r.spend or 0) > 0 for r in rs))
+
+        assert [r.request_id for r in polled] == ["r1"]
+        assert transport.calls == 2, "poller must keep reading until the predicate holds"
+
+    def test_a_window_of_only_failed_reads_raises_instead_of_reporting_no_rows(self) -> None:
+        """The regression: 24 consecutive 500s used to come back as an empty list, so
+        the caller asserted "no spend row was written" for a request whose row it had
+        never been able to read."""
+        client, _ = _client([UnknownApiError(status_code=500, body='{"error":"deadlock detected"}')])
+
+        with pytest.raises(AssertionError) as failure:
+            client.poll_logs_for_key("sk-test")
+
+        message = str(failure.value)
+        assert "never returned a successful read" in message
+        assert "500" in message and "deadlock detected" in message, (
+            f"the failure must carry the read that failed, not just its own verdict: {message}"
+        )
+
+    def test_a_readable_endpoint_with_no_rows_still_reports_no_rows(self) -> None:
+        """The other side of the same contract: emptiness the proxy actually served is
+        a real answer about the spend pipeline and must reach the caller unchanged."""
+        client, _ = _client([_served([])])
+
+        assert client.poll_logs_for_key("sk-test") == []
+
+    def test_a_transient_read_failure_is_polled_through(self) -> None:
+        rows = [_row("r1", 0.02)]
+        client, transport = _client([UnknownApiError(status_code=500, body="blip"), _served(rows)])
+
+        polled = client.poll_logs_for_key("sk-test")
+
+        assert [r.request_id for r in polled] == ["r1"]
+        assert transport.calls == 2, "one failed read must not end the poll"
+
+    def test_request_id_polling_shares_the_contract(self) -> None:
+        client, _ = _client([UnknownApiError(status_code=503, body="upstream unavailable")])
+
+        with pytest.raises(AssertionError, match="never returned a successful read"):
+            client.poll_logs_for_request_id("chatcmpl-abc")