PreprocessWorkerThread failures never reach _fail_request, so they surface as a generic 500 timeout

[t-avil]

What happened?

Summary

PreprocessWorkerThread.run() only routes ONE failure path to _fail_request —
an exception from _process_input (data_worker.py:562-572):

try:
    self._process_input(pre_input)
except Exception as exc:  # noqa: BLE001 — any failure must reach the client
    self._fail_request(pre_input.request_id, exc, "preprocessing")
    ...

Everything else in the loop body — _process_messages(), _process_read_tensors(),
the abort/discard/cleanup queue drains — is only covered by the outer catch-all:

except Exception:
    logger.exception("PreprocessWorkerThread error")

which logs and moves on. It does not call _fail_request, so a request whose
failure happens there never gets an error chunk, never sets req.error, and
the client's collect_results just polls until timeout_seconds and raises a
generic "500 Request timed out" — masking whatever the real failure (and status)
was.

This was flagged during PR #226 review as "non-streaming errors surface as an
error chunk instead of an HTTP error." That specific path was already fixed by
#200 (e22e886e, 2026-08-04) — collect_results does raise the right status
today for a process_prompt failure. This issue is the part of that concern
that's still true.

Proposed fix

Route the outer catch-all through _fail_request too, scoped to whatever
request was being handled when it threw, or split the loop body so each queue
drain gets its own guarded call the way _process_input already does.

Acceptance criteria

• A failure in any of _process_messages, _process_read_tensors, or the
  abort/discard/cleanup drains reaches the client as an error chunk /
  HTTP status, not a bare timeout
• PreprocessWorkerThread error log line is no longer the only signal for
  these failures
• Test covering at least one non-_process_input failure path

How to reproduce

Hard to hit deterministically from the outside (needs a failure inside
_process_messages/_process_read_tensors, not _process_input). Easiest
repro is to raise synthetically inside _process_read_tensors in a test and
confirm the client times out instead of getting a 4xx/5xx immediately.

Environment

No response