gproxy_protocol/adapt/responses_ws/
mod.rs

1//! Native Responses WebSocket turns over a single host-authenticated connection.
2//!
3//! Only `connect` constructs a session. Its destination/authentication remains
4//! bound to the host connection; this module never reconnects, reroutes, retries
5//! a generation, or substitutes a `previous_response_id`. The host enforces
6//! wall-clock/idle deadlines and WebSocket wire framing. No native clock is read.
7//!
8//! A turn exclusively borrows its session. Dropping it before a validated
9//! terminal closes the owned socket handles and poisons reuse, including when
10//! cancellation happens before polling, during send readiness/flush, or receive.
11//! The host transport owns the concrete cleanup caused by dropping its handles;
12//! an asynchronous close handshake cannot be promised from `Drop`.
13//!
14//! All 53 declared event shapes decode, but validated turns follow the native
15//! collector's policy: four orphaned audio/transcript shapes are Unsupported
16//! because the canonical response DTO has no audio item. There is no implicit
17//! audio consumer, transcript substitution, or fabricated terminal response.
18
19mod connect;
20mod turn;
21use crate::{
22    WebSocket,
23    connection::WsClose,
24    transform::{
25        Report, TransformError, TransformErrorKind,
26        generate::stream::responses::ResponsesStreamLimits,
27    },
28    wire::openai::responses::{
29        generate::GenerateContentRequestBody, response::ResponseStatus, stream::StreamEvent,
30    },
31};
32pub use connect::{ResponsesWsConnect, ResponsesWsConnectError, connect};
33pub use turn::ResponsesWsTurn;
34
35#[derive(Debug, Clone, Copy)]
36pub struct ResponsesWsLimits {
37    /// Payload cap for each complete data message or control frame.
38    pub max_frame_bytes: usize,
39    /// JSON request/response event cap, additionally bounded by each direction.
40    pub max_event_bytes: usize,
41    /// Per-turn totals include all data and control frame payloads.
42    pub max_receive_bytes: usize,
43    pub max_send_bytes: usize,
44
45    /// Native Responses uses text messages. Binary JSON is an explicit opt-in.
46    pub allow_binary: bool,
47    pub collector: ResponsesStreamLimits,
48}
49
50impl Default for ResponsesWsLimits {
51    fn default() -> Self {
52        Self {
53            max_frame_bytes: 1024 * 1024,
54            max_event_bytes: 1024 * 1024,
55            max_receive_bytes: 16 * 1024 * 1024,
56            max_send_bytes: 16 * 1024 * 1024,
57
58            allow_binary: false,
59            collector: ResponsesStreamLimits::default(),
60        }
61    }
62}
63
64#[derive(Debug, Clone, Copy)]
65pub(super) struct Bounds {
66    send_frame: usize,
67    receive_frame: usize,
68    send_event: usize,
69    receive_event: usize,
70    send_bytes: usize,
71    receive_bytes: usize,
72
73    allow_binary: bool,
74    collector: ResponsesStreamLimits,
75}
76
77#[derive(Debug)]
78pub struct ResponsesWsSession {
79    socket: Option<WebSocket>,
80    bounds: Bounds,
81    poisoned: bool,
82    busy: bool,
83    failure: Option<Box<StreamEvent>>,
84    websocket_failure: Option<Box<crate::wire::openai::responses::websocket::ErrorMessage>>,
85    peer_close: Option<WsClose>,
86    last_response_id: Option<String>,
87    last_status: Option<ResponseStatus>,
88    last_report: Report,
89}
90
91impl ResponsesWsSession {
92    pub(super) fn connected(socket: WebSocket, bounds: Bounds) -> Self {
93        Self {
94            socket: Some(socket),
95            bounds,
96            poisoned: false,
97            busy: false,
98            failure: None,
99            websocket_failure: None,
100            peer_close: None,
101            last_response_id: None,
102            last_status: None,
103            last_report: Report::default(),
104        }
105    }
106    pub fn is_poisoned(&self) -> bool {
107        self.poisoned
108    }
109    pub fn is_busy(&self) -> bool {
110        self.busy
111    }
112    /// A received native Error/Failed event remains available after poisoning.
113    pub fn failure_event(&self) -> Option<&StreamEvent> {
114        self.failure.as_deref()
115    }
116    pub fn websocket_failure(
117        &self,
118    ) -> Option<&crate::wire::openai::responses::websocket::ErrorMessage> {
119        self.websocket_failure.as_deref()
120    }
121    pub fn peer_close(&self) -> Option<&WsClose> {
122        self.peer_close.as_ref()
123    }
124    /// This is the real native ID, never an inferred continuation or alias.
125    pub fn last_response_id(&self) -> Option<&str> {
126        self.last_response_id.as_deref()
127    }
128    pub fn last_status(&self) -> Option<ResponseStatus> {
129        self.last_status
130    }
131    pub fn last_report(&self) -> &Report {
132        &self.last_report
133    }
134    pub fn turn(
135        &mut self,
136        request: GenerateContentRequestBody,
137    ) -> Result<ResponsesWsTurn<'_>, TransformError> {
138        if self.poisoned || self.busy || self.socket.is_none() {
139            return Err(TransformError::new(
140                TransformErrorKind::MissingState,
141                "responses.websocket",
142                "connection is closed or has an uncertain turn",
143            ));
144        }
145        self.turn_message(crate::wire::openai::responses::websocket::RequestMessage {
146            event: crate::wire::openai::responses::websocket::ClientEvent::ResponseCreate(request),
147            stream_id: None,
148            generate: None,
149        })
150    }
151    /// One explicit lane at a time. Native Responses supports concurrent lanes;
152    /// this owned-turn helper intentionally serializes them.
153    pub fn turn_message(
154        &mut self,
155        request: crate::wire::openai::responses::websocket::RequestMessage,
156    ) -> Result<ResponsesWsTurn<'_>, TransformError> {
157        if self.poisoned || self.busy || self.socket.is_none() {
158            return Err(TransformError::new(
159                TransformErrorKind::MissingState,
160                "responses.websocket",
161                "connection is closed or has an uncertain turn",
162            ));
163        }
164        ResponsesWsTurn::new(self, request)
165    }
166    fn poison(&mut self) {
167        self.poisoned = true;
168        self.busy = false;
169        self.socket.take();
170    }
171}
172
173fn limit(field: &'static str, detail: &'static str) -> TransformError {
174    TransformError::new(TransformErrorKind::Limit, field, detail)
175}
176
177fn invalid(detail: &'static str) -> TransformError {
178    TransformError::invalid_result("responses.websocket", detail)
179}
180
181fn codec_limits(max: usize) -> crate::codec::CodecLimits {
182    crate::codec::CodecLimits {
183        max_buffer_bytes: max as u64,
184        max_value_bytes: max as u64,
185        max_body_bytes: max as u64,
186        max_line_bytes: max as u64,
187        max_part_bytes: max as u64,
188        max_parts: 1,
189    }
190}
191
192fn codec_error(error: crate::codec::CodecError, request: bool) -> TransformError {
193    let kind = if error.kind() == crate::codec::CodecErrorKind::Limit {
194        TransformErrorKind::Limit
195    } else if request {
196        TransformErrorKind::InvalidInput
197    } else {
198        TransformErrorKind::InvalidResult
199    };
200    TransformError::with_source(
201        kind,
202        if request {
203            "responses.websocket.request"
204        } else {
205            "responses.websocket.event"
206        },
207        error.to_string(),
208        error,
209    )
210}
211
212fn host_error(error: crate::connection::TransportError) -> TransformError {
213    TransformError::with_source(
214        TransformErrorKind::Host,
215        "responses.websocket",
216        error.to_string(),
217        error,
218    )
219}
gproxy_protocol/adapt/responses_ws/mod.rs

gproxy_protocol/adapt/responses_ws/
mod.rs
