Feat/responses websocket#138
Merged
Merged
Conversation
ResponsesStreamTranslator now yields plain event dicts instead of pre-framed SSE strings; SSE framing (and the [DONE] sentinel) moves to a new frame_sse edge wrapper applied only on the HTTP /v1/responses route. This is prep for a WebSocket transport, which needs the same event stream framed differently (raw json.dumps, no [DONE]). persist_response and store_failure_event now operate on event dicts directly instead of re-parsing our own SSE output.
…Error
ResponsesApiError (an HTTPException carrying a full ErrorResponse) lets
resolve_history_items/load_snapshot/delete_snapshot raise one error
shared by both transports: HTTP renders it via the OpenAI envelope
(_error_response, now registered as an exception handler ahead of the
plain-HTTPException one), the upcoming WS handler will render the same
error via error_ws_frame.
Also gives a previously-unknown previous_response_id a machine-readable
code ("previous_response_not_found") and param, upgrading the existing
HTTP 404's bare {"detail": ...} body to the full OpenAI error envelope.
Adds `@app.websocket("/v1/responses")`: one socket, many sequential
`response.create` turns. Each turn reuses the existing HTTP dispatch
path (handle.respond, persist_response) and reframes the resulting
event dicts as raw json.dumps text frames instead of SSE — never a
`[DONE]` sentinel, since a terminal event ends a turn, not the
connection.
- Auth: BaseHTTPMiddleware (ApiKeyMiddleware included) never runs for
websocket connections, so check_ws_auth enforces the same Bearer-key
check in-handler, before accept().
- Disconnect: a single reader task owns the socket (only one coroutine
may call Starlette's receive() at a time - the underlying `websockets`
recv() raises RuntimeError on a concurrent second caller) and feeds
frames to the turn loop over a queue, so it can also notice a
disconnect *while* a turn is generating and propagate it through the
same DisconnectRegistry actor HTTP's RequestWatcher uses.
- State: previous_response_id resolves against a connection-local cache
first (store:false turns this socket itself produced - never written
to the global store, so a reconnect or a different socket correctly
misses), falling back to the shared resolve_history_items helper
(global store) otherwise. A failed continuation evicts its cached id.
The cache is capped (oldest evicted) so a long-lived socket issuing
many store:false turns doesn't grow it unboundedly.
- Every failure mode (bad JSON, wrong frame type, invalid request body,
unknown model, resolution miss, mid-stream Ray/loader failure) is
rendered as a single error_ws_frame and never raises back into the
socket loop, so one bad turn never kills the connection.
Also closes a real gap in the request-side adapter: a function_call_output
whose call_id has no matching function_call earlier in the same input is
now rejected as a 400 (orphaned tool result) instead of silently forwarded
to the model as a dangling tool_call_id - this is what makes a failed
continuation deterministic on the WS eviction path, and benefits the
HTTP path identically since the check lives in the shared adapter.
…TTP too Same cleanup as _run_ws_turn: input_items_for_turn is computed once, right where request.input's shape is actually decided, instead of calling as_input_items() again later on a value resolve_history already returned as a list.
…te store The gateway (encrypt) and a model deployment actor (decrypt) are always separate Ray processes, so a per-process-random fallback key for MSHIP_COMPACTION_KEY meant a /v1/responses/compact blob could never actually decrypt anywhere — broken by default, not just multi-replica. The deploy driver now resolves the key once (explicit env var, or a freshly generated one) and seeds it into the cluster-wide StateStore before any actor starts, the same store already backing effective-config and conversation state. compaction_crypto._resolve_key() is now a pure store read that fails hard if nothing was seeded, instead of silently minting a key that would only make the problem harder to notice.
WebSocket transport (shipped this branch) plus the compaction-key fix take the suite from 10/17 to a full 17/17, verified with zero MSHIP_COMPACTION_KEY configuration.
There was a problem hiding this comment.
Pull request overview
Adds first-class WebSocket transport support for the OpenAI /v1/responses API, while refactoring streaming to emit transport-neutral event dicts and pushing SSE framing concerns to the HTTP gateway edge.
Changes:
- Implement
/v1/responsesWebSocket route with per-turn dispatch, connection-local continuation cache forstore:false, and shared disconnect propagation viaDisconnectRegistry. - Refactor Responses streaming pipeline to produce plain event dicts (no SSE framing, no
[DONE]sentinel), introducingframe_sse()(HTTP edge) anderror_ws_frame()(WS edge). - Make compaction encryption key cluster-consistent by seeding and reading it from the shared
StateStore(seeded during deploy).
Reviewed changes
Copilot reviewed 21 out of 21 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test_vllm_responses.py | Updates vLLM Responses streaming tests to assert event dict sequences instead of SSE-framed strings. |
| tests/test_responses_ws.py | Adds unit tests for /v1/responses WebSocket turn handling, validation, continuation, and mid-stream failures. |
| tests/test_responses_streaming.py | Updates translator tests to consume event dicts; adds tests for HTTP SSE framing and WS error framing helpers. |
| tests/test_responses_compaction.py | Refactors compaction key tests around shared store seeding and per-process caching semantics. |
| tests/test_responses_api.py | Aligns HTTP /v1/responses tests with transport-neutral event dict streaming + gateway SSE framing; validates OpenAI-shaped error envelopes. |
| tests/test_responses_adapter.py | Adds orphan function_call_output rejection test and updates compaction key setup to use seeded store. |
| tests/test_llama_server_infer.py | Updates llama_server streaming tests to assert event dict types rather than SSE string contents. |
| tests/test_auth.py | Adds WebSocket auth tests for check_ws_auth (since HTTP middleware doesn’t apply to WS). |
| README.md | Updates Open Responses compliance results to include passing WebSocket transport suite. |
| mship_deploy.py | Seeds compaction key into the shared StateStore during deploy startup. |
| modelship/openai/utils/responses.py | Introduces ResponsesApiError to carry full OpenAI error envelopes; updates persistence to operate on event dicts. |
| modelship/openai/protocol/responses/streaming.py | Makes streaming translator transport-neutral; adds frame_sse() and error_ws_frame() helpers; returns dict events. |
| modelship/openai/protocol/responses/adapter.py | Adds validation to reject orphaned function_call_output items lacking a matching earlier function_call. |
| modelship/openai/protocol/responses/init.py | Re-exports frame_sse / error_ws_frame and related streaming symbols. |
| modelship/openai/protocol/init.py | Re-exports additional Responses protocol helpers at the protocol package root. |
| modelship/openai/compaction_crypto.py | Replaces per-process ephemeral key behavior with shared-store seeding + process-local caching; fails hard if unseeded. |
| modelship/openai/auth.py | Adds check_ws_auth and broadens identity resolution helpers to accept HTTPConnection for WS compatibility. |
| modelship/openai/api.py | Implements /v1/responses WebSocket endpoint and per-turn runner; frames HTTP streaming via frame_sse(). |
| modelship/infer/vllm/vllm_infer.py | Updates Responses streaming type signature to emit event dicts instead of SSE strings. |
| modelship/infer/llama_server/llama_server_infer.py | Updates Responses streaming type signature to emit event dicts instead of SSE strings. |
| modelship/infer/base_infer.py | Refactors _stream_responses to yield event dicts and remove [DONE] emission from loader/translator layer. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
…ror-message leaks Cuts several 5-25 line docstrings/comments across the HTTP/WS transport code down to 1-2 lines, fixes _run_ws_turn's RayTaskError branch sending a raw str(exc) to the client instead of a generic message, and corrects load_snapshot's GET/DELETE 404s to say "Response ... not found" instead of the previous_response_id-specific wording they'd inherited from _not_found_error.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.