Feat/responses websocket

README.md

@@ -194,7 +194,7 @@ We are currently hardening the Kubernetes/KubeRay path (a Helm chart ships in [`
 
 `/v1/responses` is also tested against the independent [Open Responses](https://github.com/openresponses/openresponses) compliance suite (`bun run test:compliance`), which exercises the endpoint over real HTTP against a live deployment rather than mocks.
 
-**Latest result: 10/17** (`Qwen3-VL-8B-Instruct` AWQ, vLLM, 2026-07-23):
+**Latest result: 17/17** (`Qwen3-VL-8B-Instruct` AWQ, vLLM, 2026-07-24), including the full WebSocket transport suite:
 
 | Test | Category | Status |
 |---|---|---|
@@ -208,15 +208,13 @@ We are currently hardening the Kubernetes/KubeRay path (a Helm chart ships in [`
 | Compaction Endpoint | `/v1/responses/compact` | ✅ Pass |
 | Compaction Missing Required Model | `/v1/responses/compact` | ✅ Pass |
 | Image Input | Vision | ✅ Pass |
-| WebSocket Response | WebSocket | ❌ Fail — transport not implemented |
-| WebSocket Sequential Responses | WebSocket | ❌ Fail — transport not implemented |
-| WebSocket Continuation | WebSocket | ❌ Fail — transport not implemented |
-| WebSocket Store False Reconnect Recovery | WebSocket | ❌ Fail — transport not implemented |
-| WebSocket Missing Previous Response | WebSocket | ❌ Fail — transport not implemented |
-| WebSocket Failed Continuation Evicts Cache | WebSocket | ❌ Fail — transport not implemented |
-| WebSocket Compact New Chain | WebSocket | ❌ Fail — transport not implemented |
-
-Both vision-dependent rows were verified semantically, not just schema-wise: Image Input correctly described the test fixture ("A solid red heart symbol is shown."), and Tool Calling produced a real `function_call` item (`get_weather` with parsed `location` argument), not a text fallback. The only remaining gap is WebSocket transport (7 tests) — an alternative to the HTTP/SSE `/v1/responses` API already supported today, and not yet implemented.
+| WebSocket Response | WebSocket | ✅ Pass |
+| WebSocket Sequential Responses | WebSocket | ✅ Pass |
+| WebSocket Continuation | WebSocket | ✅ Pass |
+| WebSocket Store False Reconnect Recovery | WebSocket | ✅ Pass |
+| WebSocket Missing Previous Response | WebSocket | ✅ Pass |
+| WebSocket Failed Continuation Evicts Cache | WebSocket | ✅ Pass |
+| WebSocket Compact New Chain | WebSocket | ✅ Pass |
 
 ## Contributing
 

modelship/infer/base_infer.py

@@ -232,27 +232,15 @@ async def _stream_responses(
         *,
         request_id: str,
         client_error: Callable[[Exception], str | None] = lambda _exc: None,
-    ) -> AsyncGenerator[str, None]:
-        """Drive a Responses SSE event stream from a loader's typed chat chunks.
-
-        Shared by every loader's `create_response`: each supplies its own native
-        chunk source (already wrapped in `run_cancellable_stream`) and this owns
-        the `ResponsesStreamTranslator` lifecycle — start, one `process()` per
-        chunk, then `finish()` on a clean end or `fail()` on an error.
-        `ClientDisconnectedError` (raised by `run_cancellable_stream` once the
-        client is gone) ends the stream silently, matching every other endpoint.
-        Any other exception is passed to `client_error`, which returns a
-        client-safe message to report via `fail()`, or `None` to log a full
-        stack trace and report a generic "Internal error during generation".
-
-        The `finally` block's `chunks.aclose()` is what guarantees `chunks` (and
-        transitively the native engine/subprocess stream it wraps) is torn down
-        even when neither of the `except` branches runs — e.g. the consumer
-        (Starlette/Ray Serve) closes *this* generator early while it's suspended
-        at one of the `yield`s below. `async for` does not propagate that closure
-        into the generator being iterated the way `yield from` would for a
-        synchronous generator, so without this `chunks` would otherwise be left
-        suspended forever, leaking its disconnect-poll task and underlying stream.
+    ) -> AsyncGenerator[dict[str, Any], None]:
+        """Drive a Responses event stream: start() -> process() per chunk ->
+        finish()/fail(). Yields plain event dicts; SSE/WS framing happens at the
+        transport edge, not here.
+
+        `client_error(exc)` returns a client-safe message for `fail()`, or None to
+        log the full trace and use a generic message. `chunks.aclose()` in `finally`
+        is required: closing this generator early doesn't propagate into `chunks`
+        via `async for` the way `yield from` would for a sync generator.
         """
         translator = ResponsesStreamTranslator(request)
         try:
@@ -275,7 +263,6 @@ async def _stream_responses(
                 return
             for event in translator.finish():
                 yield event
-            yield "data: [DONE]\n\n"
         finally:
             await chunks.aclose()
 
@@ -392,7 +379,7 @@ async def create_chat_completion(
 
     async def create_response(
         self, request: ResponsesRequest, raw_request: RawRequestProxy
-    ) -> ErrorResponse | ResponseObject | AsyncGenerator[str, None]:
+    ) -> ErrorResponse | ResponseObject | AsyncGenerator[dict[str, Any], None]:
         """Responses counterpart of `create_chat_completion` — see its docstring."""
         prepared = await self._prepare_responses(request, raw_request)
         if isinstance(prepared, ErrorResponse):
@@ -432,7 +419,7 @@ async def _create_chat_completion_no_stream(
 
     def _create_response_stream(
         self, request: ResponsesRequest, prepared: Prepared, raw_request: RawRequestProxy
-    ) -> AsyncGenerator[str, None]:
+    ) -> AsyncGenerator[dict[str, Any], None]:
         """Streaming Responses seam. See `_create_chat_completion_stream`."""
         raise NotImplementedError
 

modelship/infer/llama_server/llama_server_infer.py

@@ -578,7 +578,7 @@ async def _stream_chat_completion_body(self, payload: dict[str, Any], request_id
 
     async def _create_response_stream(
         self, request: ResponsesRequest, prepared: dict[str, Any], raw_request: RawRequestProxy
-    ) -> AsyncGenerator[str, None]:
+    ) -> AsyncGenerator[dict[str, Any], None]:
         """Native streaming Responses path: feeds `BaseInfer._stream_responses` directly
         from `_raw_stream_chunks`'s typed chunks, same source `_stream_chat_completion_body`
         uses — no chat SSE text round trip."""

modelship/infer/vllm/vllm_infer.py

@@ -768,7 +768,7 @@ async def _create_response_stream(
         request: ResponsesRequest,
         prepared: _VllmPrepared,
         raw_request: RawRequestProxy,
-    ) -> AsyncGenerator[str, None]:
+    ) -> AsyncGenerator[dict[str, Any], None]:
         """Native streaming Responses path: feeds `BaseInfer._stream_responses` directly
         from `engine_ops.stream_chat_completion`'s typed chunks — no chat SSE text
         round trip. Rendering already succeeded in `_prepare_responses`."""