refactor(streaming): dedupe transport conversion, usage merge, cancel finalize

Three behavior-preserving consolidations behind the #937 fix, each
deleting a hand-rolled twin of a now-shared rule:

- drain_stream consumes transport_guarded(chunks) and drops its inline
  `except httpx.TransportError` arm — one conversion rule for mid-body
  wire deaths across the drained and interactive lanes. The post-finish
  tolerance now logs under the wrapper's `stream.post_finish_blip` name
  (formerly `drain_stream.post_finish_blip`) and no longer carries
  `usage_captured`; changelog notes the rename for external log
  filters. The possible usage=None result on a post-finish blip is
  documented on drain_stream itself.
- _stream_attempt's hand-rolled per-chunk usage max-merge becomes a
  local UsageInfo accumulator folded through merge_usage (drain's
  rule), re-projected into the _last_usage dict on EVERY usage chunk —
  that dict has mid-stream readers (_estimated_prompt_tokens, the
  status line), so the per-chunk write timing is load-bearing and
  unchanged.
- The twin cancelled-partial sequences in _stream_attempt's two cancel
  arms (cooperative GenerationCancelled, stream-close-converted) merge
  into one local _record_cancelled_partial helper carrying both arms'
  tool_calls/_provider_content omission rationale in one place.
This commit is contained in:
Patrick Buckley
2026-08-03 23:38:48 -07:00
parent 5fb27e8f81
commit a1dfe0bd4f
5 changed files with 85 additions and 97 deletions
+16
View File
@@ -351,6 +351,22 @@ class TestTransportGuarded:
class TestErrorPropagation:
def test_post_finish_blip_keeps_completed_result(self):
# The generation completed (finish reason in hand) — a trailing
# transport blip forfeits only trailing metadata (here: the usage
# chunk), never the completed result.
import httpx
def chunks():
yield StreamChunk(content_delta="whole answer")
yield StreamChunk(finish_reason="stop")
raise httpx.ReadError("late blip")
result = drain_stream(chunks())
assert result.content == "whole answer"
assert result.finish_reason == "stop"
assert result.usage is None
def test_httpx_transport_error_becomes_retryable_incomplete(self):
# Streaming moves the body read out of the SDK's wrapped request:
# a mid-body wire failure surfaces as a raw httpx.TransportError