feat(responses): add WebSocket transport and fix OpenResponses conformance gaps

Bring the Responses API up to full OpenResponses conformance (17/17, previously
2/17 in the conformance workflow).

Add a WebSocket transport at /v1/responses. Each text frame is a response.create
event that is streamed back as individual JSON event frames, terminating on
response.completed/failed/incomplete. The transport keeps a connection-local
cache of store=false response output so a follow-up previous_response_id on the
same connection can continue a chain that was never persisted; a failed
continuation evicts the entry it built on, and a fresh connection starts empty.
Unknown or missing previous responses return a previous_response_not_found error
envelope. This requires the websockets library, which is added as a dependency
so uvicorn can negotiate the upgrade.

Fix two response schema gaps: the response now always carries safety_identifier
(string or null) and text.verbosity defaults to "medium" instead of serializing
as null, which the OpenResponses optional-enum schema rejects.

Fix the compaction endpoint to normalize bare-string user message content into a
content-part list so compacted output items match the response message schema.

Add replay recordings for the assistant-phase, compaction, and WebSocket turns
so the conformance workflow passes without live inference.

Signed-off-by: Charlie Doern <cdoern@redhat.com>

Author: cdoern

client-sdks/stainless/openapi.yml

@@ -7382,6 +7382,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
         input:
           items:
             $ref: '#/components/schemas/OpenAIResponseMessageOutputUnion'
@@ -7477,6 +7481,7 @@ components:
             - medium
             - high
           - type: 'null'
+          default: medium
       type: object
       title: OpenAIResponseText
       description: Text response configuration for OpenAI responses.
@@ -7795,6 +7800,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
       type: object
       required:
       - created_at

docs/static/deprecated-ogx-spec.yaml

@@ -3350,6 +3350,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
         input:
           items:
             $ref: '#/components/schemas/OpenAIResponseMessageOutputUnion'
@@ -3445,6 +3449,7 @@ components:
             - medium
             - high
           - type: 'null'
+          default: medium
       type: object
       title: OpenAIResponseText
       description: Text response configuration for OpenAI responses.
@@ -3763,6 +3768,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
       type: object
       required:
       - created_at

docs/static/experimental-ogx-spec.yaml

@@ -3795,6 +3795,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
         input:
           items:
             $ref: '#/components/schemas/OpenAIResponseMessageOutputUnion'
@@ -3890,6 +3894,7 @@ components:
             - medium
             - high
           - type: 'null'
+          default: medium
       type: object
       title: OpenAIResponseText
       description: Text response configuration for OpenAI responses.
@@ -4208,6 +4213,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
       type: object
       required:
       - created_at

docs/static/ogx-spec.yaml

@@ -6935,6 +6935,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
         input:
           items:
             $ref: '#/components/schemas/OpenAIResponseMessageOutputUnion'
@@ -7030,6 +7034,7 @@ components:
             - medium
             - high
           - type: 'null'
+          default: medium
       type: object
       title: OpenAIResponseText
       description: Text response configuration for OpenAI responses.
@@ -7348,6 +7353,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
       type: object
       required:
       - created_at

docs/static/stainless-ogx-spec.yaml

@@ -7382,6 +7382,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
         input:
           items:
             $ref: '#/components/schemas/OpenAIResponseMessageOutputUnion'
@@ -7477,6 +7481,7 @@ components:
             - medium
             - high
           - type: 'null'
+          default: medium
       type: object
       title: OpenAIResponseText
       description: Text response configuration for OpenAI responses.
@@ -7795,6 +7800,10 @@ components:
         store:
           type: boolean
           title: Store
+        safety_identifier:
+          anyOf:
+          - type: string
+          - type: 'null'
       type: object
       required:
       - created_at

pyproject.toml

@@ -65,6 +65,7 @@ dependencies = [
     "termcolor",
     "tiktoken",
     "uvicorn>=0.34.0",                                # server
+    "websockets>=14.0",                               # server - WebSocket transport for the Responses API
     "opentelemetry-sdk>=1.42.1",                      # server
     "opentelemetry-exporter-otlp-proto-http>=1.30.0", # server
     "opentelemetry-distro>=0.60b1",                  # optional CLI instrumentation; only pre-releases on PyPI (latest 0.60b1)

src/ogx/providers/inline/responses/builtin/responses/openai_responses.py

@@ -1307,13 +1307,19 @@ async def compact_openai_response(
         output_items: list[OpenAIResponseInput] = []
         for item in all_input:
             if isinstance(item, OpenAIResponseMessage) and item.role == "user":
+                # Normalize bare-string content to a content-part list so the
+                # compacted output message matches the response message schema,
+                # which requires content to be an array of parts.
+                content = item.content
+                if isinstance(content, str):
+                    content = [OpenAIResponseInputMessageContentText(text=content)]
                 output_items.append(
                     OpenAIResponseMessage(
                         id=f"msg_{uuid.uuid4().hex[:24]}",
                         type="message",
                         status="completed",
                         role="user",
-                        content=item.content,
+                        content=content,
                     )
                 )
 

src/ogx_api/openai_responses.py

@@ -480,7 +480,9 @@ class OpenAIResponseText(BaseModel):
     """
 
     format: OpenAIResponseTextFormat | None = None
-    verbosity: Literal["low", "medium", "high"] | None = None
+    # Defaults to "medium" to match OpenAI: the OpenResponses schema types verbosity as an
+    # optional enum that rejects an explicit null, so the response must not serialize it as null.
+    verbosity: Literal["low", "medium", "high"] | None = "medium"
 
 
 @json_schema_type
@@ -848,6 +850,7 @@ class OpenAIResponseObject(BaseModel):
     :param max_output_tokens: (Optional) An upper bound for the number of tokens that can be generated for a response, including visible output tokens.
     :param service_tier: (Optional) The service tier to use for this response.
     :param metadata: (Optional) Dictionary of metadata key-value pairs
+    :param safety_identifier: (Optional) Stable identifier used to associate the request with an end user for safety monitoring
     """
 
     background: bool | None = Field(default=None, json_schema_extra=remove_null_from_anyof)
@@ -884,6 +887,7 @@ class OpenAIResponseObject(BaseModel):
     metadata: dict[str, str] | None = None
     presence_penalty: float | None = Field(default=None, json_schema_extra=remove_null_from_anyof)
     store: bool
+    safety_identifier: str | None = None
 
 
 @json_schema_type

src/ogx_api/responses/fastapi_routes.py

@@ -17,9 +17,9 @@
 from collections.abc import AsyncIterator
 from typing import Annotated, Any
 
-from fastapi import APIRouter, Body, Depends, Path, Query, Request, Response
+from fastapi import APIRouter, Body, Depends, Path, Query, Request, Response, WebSocket, WebSocketDisconnect
 from fastapi.responses import StreamingResponse
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 
 from ogx_api.common.responses import Order
 from ogx_api.openai_responses import (
@@ -214,6 +214,134 @@ async def wrapper() -> AsyncIterator[str]:
     return wrapper()
 
 
+# Fields that the WebSocket response.create event forbids (they only apply to
+# the HTTP streaming transport) and the discriminator we strip before building
+# the request.
+_WS_STRIPPED_FIELDS = ("type", "stream", "stream_options", "background")
+
+
+def _ws_normalize_input(input_value: Any) -> list[Any]:
+    """Normalize a response.create `input` to a list of items for context reuse."""
+    if input_value is None:
+        return []
+    if isinstance(input_value, str):
+        return [{"type": "message", "role": "user", "content": input_value}]
+    if isinstance(input_value, list):
+        return list(input_value)
+    return [input_value]
+
+
+async def _send_ws_error(
+    websocket: WebSocket,
+    status: int,
+    code: str,
+    message: str,
+    param: str | None = None,
+) -> None:
+    """Send a WebSocket error envelope (matches the OpenResponses error event)."""
+    await websocket.send_text(
+        json.dumps(
+            {
+                "type": "error",
+                "status": status,
+                "error": {"code": code, "message": message, "param": param},
+            }
+        )
+    )
+
+
+async def _handle_ws_responses_turn(
+    websocket: WebSocket,
+    impl: Responses,
+    raw: str,
+    session_cache: dict[str, tuple[list[Any], list[Any]]],
+) -> None:
+    """Process a single `response.create` event received over the WebSocket.
+
+    Streams each response event back as an individual JSON text frame. For
+    `store=false` turns, the response output is cached connection-locally so a
+    follow-up `previous_response_id` on the same connection can continue a chain
+    that was never persisted server-side.
+    """
+    try:
+        payload = json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        await _send_ws_error(websocket, 400, "invalid_json", "Failed to parse WebSocket message as JSON.")
+        return
+    if not isinstance(payload, dict):
+        await _send_ws_error(
+            websocket, 400, "invalid_request", "Failed to read response.create event; expected a JSON object."
+        )
+        return
+
+    for field in _WS_STRIPPED_FIELDS:
+        payload.pop(field, None)
+
+    store = payload.get("store", True)
+    previous_response_id = payload.get("previous_response_id")
+    building_on: str | None = None
+
+    # store=false chains are not persisted, so continuation is served from the
+    # connection-local cache rather than the responses store.
+    if previous_response_id is not None and store is False:
+        cached = session_cache.get(previous_response_id)
+        if cached is None:
+            await _send_ws_error(
+                websocket,
+                404,
+                "previous_response_not_found",
+                f"Previous response '{previous_response_id}' was not found.",
+                param="previous_response_id",
+            )
+            return
+        prev_input, prev_output = cached
+        payload["input"] = [*prev_input, *prev_output, *_ws_normalize_input(payload.get("input"))]
+        payload.pop("previous_response_id", None)
+        building_on = previous_response_id
+
+    sent_input = _ws_normalize_input(payload.get("input"))
+
+    try:
+        request = CreateResponseRequest(**{**payload, "stream": True})
+    except ValidationError as exc:
+        await _send_ws_error(websocket, 400, "invalid_request", str(exc))
+        return
+
+    final_response: OpenAIResponseObject | None = None
+    failed = False
+    try:
+        result = await impl.create_openai_response(request)
+        if not isinstance(result, AsyncIterator):
+            await _send_ws_error(websocket, 500, "server_error", "Expected a streaming response over WebSocket.")
+            return
+        async for event in result:
+            await websocket.send_text(event.model_dump_json())
+            event_type = getattr(event, "type", None)
+            if event_type in ("response.completed", "response.incomplete"):
+                final_response = getattr(event, "response", None)
+            elif event_type == "response.failed":
+                final_response = getattr(event, "response", None)
+                failed = True
+    except Exception as exc:
+        logger.exception("WebSocket responses turn failed")
+        failed = True
+        http_exc = try_translate_to_http_exception(exc)
+        status = http_exc.status_code if http_exc else 500
+        detail = http_exc.detail if http_exc else "Internal server error: An unexpected error occurred."
+        await _send_ws_error(websocket, status, "server_error", detail)
+
+    if failed:
+        # A failed continuation evicts the response it built on, so subsequent
+        # turns referencing it report previous_response_not_found.
+        if building_on is not None:
+            session_cache.pop(building_on, None)
+    elif final_response is not None and store is False:
+        session_cache[final_response.id] = (
+            sent_input,
+            [item.model_dump() for item in final_response.output],
+        )
+
+
 def create_router(impl: Responses) -> APIRouter:
     """Create a FastAPI router for the Responses API.
 
@@ -230,6 +358,20 @@ def create_router(impl: Responses) -> APIRouter:
         route_class=FormURLEncodedRoute,
     )
 
+    @router.websocket("/responses")
+    async def create_openai_response_ws(websocket: WebSocket) -> None:
+        await websocket.accept()
+        # Connection-local cache of store=false response output, keyed by response
+        # id. Lets a follow-up previous_response_id on the same socket continue a
+        # chain that was never persisted. A fresh connection starts empty.
+        session_cache: dict[str, tuple[list[Any], list[Any]]] = {}
+        try:
+            while True:
+                raw = await websocket.receive_text()
+                await _handle_ws_responses_turn(websocket, impl, raw, session_cache)
+        except WebSocketDisconnect:
+            return
+
     @router.post(
         "/responses/compact",
         response_model=OpenAICompactedResponse,