[BugFix] [load_balance_proxy] Forward decode backend errors to client instead of empty 200 (#12168)

## Description

Fixes #12166.

When the decode backend returns an error (e.g. 4xx because
`prompt_tokens + max_tokens > max_model_len`, or 5xx/OOM/connection),
the load-balance proxy currently swallows the exception inside
`generate_stream()` and ends the `StreamingResponse` generator without
yielding any body. Because `StreamingResponse` had already sent the HTTP
200 head, the client received a `200` with an **empty body** and no way
to learn the cause.

This PR makes the decode-phase error visible to the client.

## Changes
(`examples/disaggregated_prefill_v1/load_balance_proxy_server_example.py`)

1. **`open_stream_service_response_with_retry`** (renamed from the
streaming wrapper): open the decode response with `client.send(...,
stream=True)` and return `(response, chunk_iterator, first_chunk)`
instead of consuming it as a generator. On a non-2xx response it buffers
the upstream error body (`await response.aread()`), retries 5xx, and
**returns the 4xx/last-5xx immediately without raising** so the caller
can forward the real upstream status. 4xx is not retried
(deterministic). The first chunk is pre-read on success so
connection-level failures surface within the retry budget.

2. **`_open_decoder_stream_or_early_response`**: open the decode
response *before* creating `StreamingResponse`. If the decode backend
failed up front, return a verbatim
`Response(status_code=upstream.status_code, content=upstream.content)`
(or a `502` for connection failures) — i.e. the real upstream HTTP
status reaches the client before the ASGI response head is committed.
Only on success is a `StreamingResponse` created (carrying the upstream
200).

3. **`_DecodeStreamSession`** (replaces the nested `generate_stream`
closure): per-request state for chunk forwarding, recompute retry, and
error forwarding. Its `except Exception` forwards a proxy-side/late
fault to the client as an SSE `error` event + `data: [DONE]` (streaming)
or a JSON `{"error": {...}}` body (non-streaming), instead of dropping
it. A mid-stream late decode error (only reachable on a recompute retry
after the head is committed) is forwarded the same way.

4. Helpers: `DecodeUpstreamError`, `build_error_payload` (returns the
standard `{"error":{...}}` shape verbatim, with a flat-`{"message":}`
wrap and a raw-body-truncation fallback), `_encode_error_body`,
`_parse_stream_chunk`, `_upstream_error_response`,
`_decode_unavailable_response`, `_late_decode_error_body`.

## Behavior

| Scenario | Before | After |
| --- | --- | --- |
| Streaming + initial decode 4xx/5xx | `200` + 0 bytes | **real
`4xx`/`5xx`** + upstream body verbatim |
| Non-streaming + initial decode 4xx/5xx | `200` + 0 bytes | **real
`4xx`/`5xx`** + upstream body verbatim |
| Decode backend unreachable (retries exhausted) | `200` + 0 bytes |
**`502`** + `{"error":{...,"code":"decode_backend_unavailable"}}` |
| Mid-stream late decode error (recompute retry) | `200` + 0 bytes (or
truncated) | `200` head already sent + SSE/JSON `{"error":{...}}` body |
| Normal request | unchanged | unchanged |

Note on the streaming status code: the original draft only did
body-level forwarding because `StreamingResponse` commits the 200 head
before the generator runs. The merged implementation goes further — by
pre-opening the decode response before `StreamingResponse` is
constructed, **initial** decode 4xx/5xx are returned with the real
upstream status code. Only errors that occur *after* the head is
committed (a recompute retry failing mid-stream) fall back to body-level
forwarding, since the status can no longer be changed.

## Follow-up fixes in this PR

- `max_retries=0` boundary: `open_stream_service_response_with_retry`
now uses `max(1, max_retries)` so `max_retries=0` still makes one
attempt (previously `range(1, 0+1)` was empty and raised `RuntimeError`
without sending any request). Matches the prefill path's existing
convention.
- Non-streaming late-error corruption: once a non-streaming body chunk
has been forwarded, a subsequent late decode error body is dropped
instead of appended (appending would concatenate with the already-sent
single JSON into invalid JSON). Streaming is unaffected (SSE errors are
independent events).

## Verification

Unit-checked locally with a mocked decode backend (`max_retries=3`):
- 4xx → returned immediately, not retried; real status + body forwarded.
- 5xx → retried, then last 5xx returned on exhaustion; 5xx-then-200
succeeds.
- `RequestError` → retried, then `502 decode_backend_unavailable`.
- `build_error_payload` extracts vLLM's
`{"error":{"message":...,"type":"BadRequestError"}}` verbatim; flat
`{"message":}` wrapped; non-JSON body truncated to 1000B; non-upstream
exceptions fall back to `502`.
- `max_retries=0` now makes one attempt (no `RuntimeError`).
- 54 mock unit cases pass (0 failures): covers all branches of
`open_stream_service_response_with_retry`,
`_open_decoder_stream_or_early_response`, `_DecodeStreamSession` (incl.
recompute loop, mid-stream late error, raw/skip chunk parsing,
cancellation), and `handle_completions_impl` (incl. `/v1/completions`
prompt path and outer `HTTPStatusError` handling).

Reproduction from the issue against a deployed 1P1D instance
(Qwen2.5-7B-Instruct on Atlas 800T A2):
- Non-streaming over-limit (`max_tokens` > `max_model_len`): previously
`200` + 0 bytes → now returns the real upstream `400` +
`{"error":{"message":"max_tokens=... cannot be greater than
max_model_len..."}}`.
- Streaming over-limit: previously `200` + 0 bytes → now returns the
real upstream `400` + the same error body.
- Decode backend down: previously `200` + 0 bytes → now `502` +
`decode_backend_unavailable`.
- Normal requests: unchanged.

- vLLM main:
https://github.com/vllm-project/vllm/commit/ba07e4a48fc951300d97eb506217dd530583dea3

Signed-off-by: xlshaoscu <xlshaoscu@foxmail.com>
Co-authored-by: Claude <noreply@anthropic.com>