openai: support websocket deflate

Co-authored-by: Codex <noreply@openai.com>

Cargo.lock

@@ -377,6 +377,7 @@ dependencies = [
  "chrono",
  "config",
  "dirs",
+ "flate2",
  "futures",
  "hex",
  "hmac",

Cargo.toml

@@ -19,6 +19,7 @@ chrono = { version = "0.4.45", features = ["serde", "clock"] }
 config = "0.15.23"
 dirs = "5.0.1"
 futures = "0.3.32"
+flate2 = "1.1.9"
 hex = "0.4.3"
 hmac = "0.13.0"
 indexmap = { version = "2.0", features = ["serde"] }

docs/OPENAI_CHAIN_PARITY_REAUDIT.md

@@ -26,7 +26,7 @@ Rust 仓库：`/home/zyy/Codes/codex-proxy-rs`
 本轮 10 个 OpenAI/Codex 链路点已全部复核并写入账本。结论：当前不能声明与原版 100% 一致；除“代码证据索引与审计边界”外，其余链路均为 `部分对齐`。
 
 最高优先级未对齐项：
-- 安全模拟仍缺 TLS 指纹证明，以及 WS permessage-deflate 被上游接受后的压缩 frame 解码证明。
+- 安全模拟仍缺 TLS 指纹证明；WS permessage-deflate offer、协商响应和服务端压缩 frame 解码已补齐测试与实现。
 - `/v1/responses` 请求构造仍存在非 `/v1` alias route、`use_websocket:false` 本地扩展等差异；`/v1/responses/review` 强制 review subagent 和 `/v1/responses/compact` 已补齐。
 - response 转换和错误恢复仍缺最终错误外壳对齐；tuple schema request conversion/reconvert、streaming premature close 合成、reasoning replay、implicit resume failure restore、`response.failed`/`error` 的 429/402/403 分类、账号 fallback、`previous_response_not_found`/unanswered function call strip-and-retry、5xx same-account retry、CF path-block 404、model unsupported fallback、401 token invalid fallback 已补齐。
 - session affinity 的显式续链、implicit resume 主路径、implicit resume failure restore、reasoning replay cache、function call continuation 预检、variant_hash 基础算法、variant identity 和 OpenAI chat `user` conversation identity 已接入。
@@ -116,10 +116,9 @@ Rust 证据：
 
 未完全对齐：
 - Rust 当前没有测试证明真实传输层 TLS/HTTP 指纹与原版 native transport 行为完全一致。IP 代理/VPN 明确非目标，但 TLS/header 指纹仍属于安全模拟链路，不能在未验证时声明 100%。
-- WebSocket 已按原版 `ws@8.19.0` 发出 `Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits`，但 Rust 后续 frame 解码仍由 `tungstenite` 负责；若上游真的协商并发送压缩 frame，当前还没有真实上游证据证明不会触发 RSV/压缩解码问题。
 
 缺口/后续动作：
-- 若目标是安全模拟 100%，需要继续处理 TLS 指纹和 WS compression negotiation 的真实上游验证；TLS 是否要追到 native transport 级别需要单独决策。
+- 若目标是安全模拟 100%，需要继续处理 TLS 指纹真实上游验证；TLS 是否要追到 native transport 级别需要单独决策。
 
 ## 3. OAuth/device/refresh 链路
 
@@ -298,6 +297,7 @@ Rust 证据：
 - 本轮修正后，WS handshake header profile 已独立于 HTTP SSE profile，不再携带 `Content-Type` 与 `Accept: text/event-stream`；`tests/codex_gateway/websocket.rs` 已覆盖该行为。
 - 本轮修正后，`src/codex/gateway/transport/websocket/opening.rs` 不再使用 `tokio-tungstenite` 默认 `HeaderMap` opening 序列化，而是显式生成原版 `ws@8.19.0` opening request：`Sec-WebSocket-Extensions` 位于 `Sec-WebSocket-Key` 前，业务 headers 按原版对象插入顺序输出，并恢复 `Authorization`、`ChatGPT-Account-Id`、`User-Agent`、`Accept-Encoding`、`Accept-Language`、`OpenAI-Beta`、`Cookie` 等原版大小写。
 - `tests/codex_gateway/websocket.rs::websocket_handshake_should_offer_original_permessage_deflate_extension` 通过 raw TCP 断言 WS opening request 的关键 header 顺序、大小写和 permessage-deflate offer 与原版抓包一致。
+- 本轮修正后，`src/codex/gateway/transport/websocket/deflate.rs` 在服务端接受 `Sec-WebSocket-Extensions: permessage-deflate` 时，会在底层流上解码 server-to-client RSV1 text/binary frame，再交给 `tungstenite` 处理；未协商时保持原始流直通。`tests/codex_gateway/websocket.rs::websocket_should_decode_permessage_deflate_response_frame_when_server_accepts_extension` 先以 `NonZeroReservedBits` 失败，再修复通过，覆盖服务端接受扩展并返回压缩 `response.completed` frame。
 - `src/codex/gateway/transport/websocket/mod.rs:230-328` pooled/reused WS stale failure 会重试一次 one-shot；内部 `codex.rate_limits` 会转为 rate-limit headers；terminal event 转为 SSE chunk。
 - 本轮修正后，`src/codex/gateway/transport/websocket/mod.rs` 的 one-shot WebSocket handshake 使用 20 秒 open timeout；`connect_websocket_with_timeout_should_fail_when_handshake_stalls` 覆盖握手停滞时返回 timeout。
 - `src/codex/gateway/transport/websocket/pool.rs:58-76` 默认 `max_age=55min`、`max_per_account=8`、`maintenance_interval=25s`、`ping_interval=25s`、`ping_timeout=5s`、`liveness_timeout=62.5s`。
@@ -319,6 +319,7 @@ Rust 证据：
 - WS URL、`response.create` frame 主字段、tool 默认值、service tier、prompt cache key、client metadata、internal rate-limit event、terminal event SSE 包装均已基本对齐。
 - WS handshake headers 已去掉原版没有的 `Content-Type` 与 `Accept: text/event-stream`。
 - WS opening request 的关键 wire 顺序、大小写和 `Sec-WebSocket-Extensions` offer 已按原版 `ws@8.19.0` 对齐，并由 raw TCP 测试覆盖。
+- WS permessage-deflate 协商后的 server-to-client 压缩 text/binary frame 已支持解码，避免上游接受扩展后因 RSV1 frame 触发 `tungstenite` `NonZeroReservedBits`。
 - one-shot WebSocket handshake 已增加原版同等 20 秒 open timeout；普通无历史请求超时后仍可按原版策略降级 HTTP SSE，`previous_response_id` required WS 不降级。
 - 连接池核心状态机基本对齐：55 分钟 max age、每账号最多 8 条、严格单 in-flight、busy/cap bypass、stale reuse 一次 fresh retry、idle keepalive、liveness/GC、shutdown、账号状态变更驱逐。
 - WebSocket pool 配置面已对齐原版：配置文件显式提供 `ws_pool.enabled/max_age_ms/max_per_account`，运行时按配置启停复用和调整 max age/account cap。
@@ -328,10 +329,9 @@ Rust 证据：
 未完全对齐：
 - pool key 表达不完全一致。原版 key 是 `${entryId}:${rawConversationId}`；Rust key 是 `(base_url, account_id, accountScopedConversationId)`。行为上通过测试证明同 conversation 会复用，但若后续要与原版日志/配置/诊断完全一致，命名与 key 语义仍不同。
 - Rust pool maintenance 是集中 sweep ping idle connection；原版是每条 `PersistentWs` 自带 interval，busy 时跳过 ping。默认行为接近，但调度模型不是逐字等价。
-- Rust 已发出原版 permessage-deflate offer，但后续 frame 解码仍由 `tungstenite` 负责；需要真实上游或压缩 frame fixture 证明 OpenAI 不会返回需要解压的 RSV1 frame，或补齐压缩解码能力。
 
 缺口/后续动作：
-- 增加 WS compression negotiation 验证：覆盖上游接受 `Sec-WebSocket-Extensions` 后的响应 header 与 frame 行为。
+- 若需要更强证明，可增加真实 OpenAI 上游抓包验证，确认当前 Codex 后端是否实际返回压缩 frame；本地压缩 frame fixture 已覆盖客户端能力。
 
 ## 7. response/SSE/WS frame 转换链路
 

src/codex/gateway/transport/websocket/deflate.rs

@@ -0,0 +1,273 @@
+use std::{
+    collections::VecDeque,
+    io,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use flate2::{Decompress, FlushDecompress, Status};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+
+const SERVER_FRAME_CHUNK_BYTES: usize = 8192;
+const PERMESSAGE_DEFLATE_TAIL: [u8; 4] = [0x00, 0x00, 0xff, 0xff];
+const OPCODE_CONTINUATION: u8 = 0x0;
+const OPCODE_TEXT: u8 = 0x1;
+const OPCODE_BINARY: u8 = 0x2;
+pub(super) struct PerMessageDeflateStream<S> {
+    inner: S,
+    enabled: bool,
+    raw_input: Vec<u8>,
+    decoded_input: VecDeque<u8>,
+    compressed_message: Option<CompressedMessage>,
+}
+
+struct CompressedMessage {
+    opcode: u8,
+    payload: Vec<u8>,
+}
+
+struct FrameParts {
+    fin: bool,
+    rsv1: bool,
+    opcode: u8,
+    masked: bool,
+    payload_offset: usize,
+    payload_len: usize,
+    frame_len: usize,
+}
+
+impl<S> PerMessageDeflateStream<S> {
+    pub(super) fn new(inner: S, enabled: bool, preloaded: Vec<u8>) -> Self {
+        Self {
+            inner,
+            enabled,
+            raw_input: preloaded,
+            decoded_input: VecDeque::new(),
+            compressed_message: None,
+        }
+    }
+}
+
+impl<S> AsyncRead for PerMessageDeflateStream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buffer: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        let this = self.get_mut();
+        if !this.enabled {
+            return Pin::new(&mut this.inner).poll_read(cx, buffer);
+        }
+
+        loop {
+            if copy_decoded_input(&mut this.decoded_input, buffer) {
+                return Poll::Ready(Ok(()));
+            }
+
+            if this.rewrite_next_frame()? {
+                continue;
+            }
+
+            let mut chunk = [0_u8; SERVER_FRAME_CHUNK_BYTES];
+            let mut read_buffer = ReadBuf::new(&mut chunk);
+            match Pin::new(&mut this.inner).poll_read(cx, &mut read_buffer) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
+                Poll::Ready(Ok(())) => {
+                    let filled = read_buffer.filled();
+                    if filled.is_empty() {
+                        return Poll::Ready(Ok(()));
+                    }
+                    this.raw_input.extend_from_slice(filled);
+                }
+            }
+        }
+    }
+}
+
+impl<S> AsyncWrite for PerMessageDeflateStream<S>
+where
+    S: AsyncRead + AsyncWrite + Unpin,
+{
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        data: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.get_mut().inner).poll_write(cx, data)
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.get_mut().inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.get_mut().inner).poll_shutdown(cx)
+    }
+}
+
+impl<S> PerMessageDeflateStream<S> {
+    fn rewrite_next_frame(&mut self) -> io::Result<bool> {
+        let Some(frame) = parse_frame_parts(&self.raw_input)? else {
+            return Ok(false);
+        };
+        let raw_frame = self.raw_input.drain(..frame.frame_len).collect::<Vec<_>>();
+        let payload = &raw_frame[frame.payload_offset..frame.payload_offset + frame.payload_len];
+
+        if frame.masked {
+            self.decoded_input.extend(raw_frame);
+            return Ok(true);
+        }
+
+        if frame.rsv1 && is_data_frame(frame.opcode) {
+            if frame.fin {
+                let inflated = inflate_message(payload)?;
+                self.decoded_input
+                    .extend(encode_server_frame(frame.opcode, false, &inflated));
+            } else {
+                self.compressed_message = Some(CompressedMessage {
+                    opcode: frame.opcode,
+                    payload: payload.to_vec(),
+                });
+            }
+            return Ok(true);
+        }
+
+        if frame.opcode == OPCODE_CONTINUATION {
+            if let Some(message) = self.compressed_message.as_mut() {
+                message.payload.extend_from_slice(payload);
+                if frame.fin {
+                    let message = self.compressed_message.take().ok_or_else(|| {
+                        io::Error::new(io::ErrorKind::InvalidData, "missing compressed message")
+                    })?;
+                    let inflated = inflate_message(&message.payload)?;
+                    self.decoded_input.extend(encode_server_frame(
+                        message.opcode,
+                        false,
+                        &inflated,
+                    ));
+                }
+                return Ok(true);
+            }
+        }
+
+        self.decoded_input.extend(raw_frame);
+        Ok(true)
+    }
+}
+
+fn copy_decoded_input(decoded_input: &mut VecDeque<u8>, buffer: &mut ReadBuf<'_>) -> bool {
+    if decoded_input.is_empty() || buffer.remaining() == 0 {
+        return false;
+    }
+
+    let copy_len = decoded_input.len().min(buffer.remaining());
+    let bytes = decoded_input.drain(..copy_len).collect::<Vec<_>>();
+    buffer.put_slice(&bytes);
+    true
+}
+
+fn parse_frame_parts(bytes: &[u8]) -> io::Result<Option<FrameParts>> {
+    if bytes.len() < 2 {
+        return Ok(None);
+    }
+
+    let first = bytes[0];
+    let second = bytes[1];
+    let mut offset = 2;
+    let payload_len = match second & 0x7f {
+        len @ 0..=125 => usize::from(len),
+        126 => {
+            if bytes.len() < offset + 2 {
+                return Ok(None);
+            }
+            let len = u16::from_be_bytes([bytes[offset], bytes[offset + 1]]);
+            offset += 2;
+            usize::from(len)
+        }
+        127 => {
+            if bytes.len() < offset + 8 {
+                return Ok(None);
+            }
+            let len = u64::from_be_bytes([
+                bytes[offset],
+                bytes[offset + 1],
+                bytes[offset + 2],
+                bytes[offset + 3],
+                bytes[offset + 4],
+                bytes[offset + 5],
+                bytes[offset + 6],
+                bytes[offset + 7],
+            ]);
+            offset += 8;
+            usize::try_from(len).map_err(|_| {
+                io::Error::new(io::ErrorKind::InvalidData, "websocket frame is too large")
+            })?
+        }
+        _ => unreachable!("websocket length marker is masked to 7 bits"),
+    };
+    let masked = second & 0x80 != 0;
+    if masked {
+        offset += 4;
+    }
+    let frame_len = offset
+        .checked_add(payload_len)
+        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "websocket frame overflow"))?;
+    if bytes.len() < frame_len {
+        return Ok(None);
+    }
+
+    Ok(Some(FrameParts {
+        fin: first & 0x80 != 0,
+        rsv1: first & 0x40 != 0,
+        opcode: first & 0x0f,
+        masked,
+        payload_offset: offset,
+        payload_len,
+        frame_len,
+    }))
+}
+
+fn inflate_message(payload: &[u8]) -> io::Result<Vec<u8>> {
+    let mut compressed = Vec::with_capacity(payload.len() + PERMESSAGE_DEFLATE_TAIL.len());
+    compressed.extend_from_slice(payload);
+    compressed.extend_from_slice(&PERMESSAGE_DEFLATE_TAIL);
+
+    let mut decompressor = Decompress::new(false);
+    let mut inflated = Vec::with_capacity(payload.len().saturating_mul(4).max(1024));
+    match decompressor.decompress_vec(&compressed, &mut inflated, FlushDecompress::Sync) {
+        Ok(Status::Ok | Status::StreamEnd) => Ok(inflated),
+        Ok(Status::BufError) if !inflated.is_empty() => Ok(inflated),
+        Ok(Status::BufError) => Err(io::Error::new(
+            io::ErrorKind::InvalidData,
+            "permessage-deflate frame produced no output",
+        )),
+        Err(error) => Err(io::Error::new(io::ErrorKind::InvalidData, error)),
+    }
+}
+
+fn encode_server_frame(opcode: u8, rsv1: bool, payload: &[u8]) -> Vec<u8> {
+    let mut frame = Vec::new();
+    let rsv1_bit = if rsv1 { 0x40 } else { 0 };
+    frame.push(0x80 | rsv1_bit | opcode);
+    match payload.len() {
+        len @ 0..=125 => frame.push(len as u8),
+        len @ 126..=65_535 => {
+            frame.push(126);
+            frame.extend_from_slice(&(len as u16).to_be_bytes());
+        }
+        len => {
+            frame.push(127);
+            frame.extend_from_slice(&(len as u64).to_be_bytes());
+        }
+    }
+    frame.extend_from_slice(payload);
+    frame
+}
+
+fn is_data_frame(opcode: u8) -> bool {
+    matches!(opcode, OPCODE_TEXT | OPCODE_BINARY)
+}

src/codex/gateway/transport/websocket/mod.rs

@@ -19,6 +19,7 @@ use crate::codex::gateway::transport::{
 };
 
 mod codec;
+mod deflate;
 mod opening;
 mod pool;
 

src/codex/gateway/transport/websocket/opening.rs

@@ -21,6 +21,7 @@ use tokio_tungstenite::{
     MaybeTlsStream, WebSocketStream,
 };
 
+use super::deflate::PerMessageDeflateStream;
 use super::pool::CodexWsStream;
 
 const MAX_OPENING_RESPONSE_BYTES: usize = 64 * 1024;
@@ -74,9 +75,11 @@ pub(super) async fn connect_with_original_opening_handshake(
         return Err(WsError::Http(Box::new(response)));
     }
     verify_accept_key(&request, &response)?;
+    let permessage_deflate = response_accepts_permessage_deflate(&response);
 
+    let stream = PerMessageDeflateStream::new(stream, permessage_deflate, buffered);
     let websocket =
-        WebSocketStream::from_partially_read(stream, buffered, Role::Client, None).await;
+        WebSocketStream::from_partially_read(stream, Vec::new(), Role::Client, None).await;
     Ok((websocket, response))
 }
 
@@ -372,3 +375,16 @@ fn verify_accept_key(request: &WsRequest<()>, response: &WsResponse) -> Result<(
         Err(ProtocolError::SecWebSocketAcceptKeyMismatch.into())
     }
 }
+
+fn response_accepts_permessage_deflate(response: &WsResponse) -> bool {
+    response
+        .headers()
+        .get_all("sec-websocket-extensions")
+        .iter()
+        .filter_map(|value| value.to_str().ok())
+        .any(|value| {
+            value
+                .split(',')
+                .any(|extension| extension.trim_start().starts_with("permessage-deflate"))
+        })
+}

src/codex/gateway/transport/websocket/pool.rs

@@ -12,6 +12,7 @@ use tokio_tungstenite::{
 };
 
 use super::codec::{rate_limit_headers, set_cookie_headers, turn_state};
+use super::deflate::PerMessageDeflateStream;
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub struct CodexWebSocketPoolKey {
@@ -356,7 +357,7 @@ struct WebSocketPoolState {
     shutting_down: bool,
 }
 
-pub(super) type CodexWsStream = WebSocketStream<MaybeTlsStream<TcpStream>>;
+pub(super) type CodexWsStream = WebSocketStream<PerMessageDeflateStream<MaybeTlsStream<TcpStream>>>;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(super) struct CodexWebSocketConnectionMetadata {

tests/codex_gateway/websocket.rs

@@ -1,5 +1,6 @@
 use std::{sync::Arc, time::Duration};
 
+use flate2::{Compress, Compression, FlushCompress};
 use futures::{SinkExt, StreamExt};
 use serde_json::{json, Value};
 use tokio::{
@@ -11,6 +12,7 @@ use tokio::{
 use tokio_tungstenite::{
     accept_async, accept_hdr_async,
     tungstenite::{
+        handshake::derive_accept_key,
         handshake::server::{Request as WsRequest, Response as WsResponse},
         Message,
     },
@@ -492,6 +494,49 @@ async fn websocket_handshake_should_offer_original_permessage_deflate_extension(
     );
 }
 
+#[tokio::test]
+async fn websocket_should_decode_permessage_deflate_response_frame_when_server_accepts_extension() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        let request = read_http_upgrade_request(&mut stream).await;
+        let accept_key = websocket_accept_key(&request);
+        let response = format!(
+            "HTTP/1.1 101 Switching Protocols\r\n\
+             Connection: Upgrade\r\n\
+             Upgrade: websocket\r\n\
+             Sec-WebSocket-Accept: {accept_key}\r\n\
+             Sec-WebSocket-Extensions: permessage-deflate\r\n\
+             \r\n"
+        );
+        stream.write_all(response.as_bytes()).await.unwrap();
+        let _client_request = read_client_websocket_frame(&mut stream).await;
+        let payload = websocket_completed_response("resp_ws_deflate", 5, 2);
+        let compressed_frame = compressed_server_text_frame(&payload);
+        stream.write_all(&compressed_frame).await.unwrap();
+        sleep(Duration::from_millis(50)).await;
+    });
+
+    let client = CodexBackendClient::new(
+        build_reqwest_client(false).unwrap(),
+        format!("http://{addr}"),
+        codex_proxy_rs::codex::gateway::fingerprint::model::Fingerprint::default_for_tests(),
+    );
+    let mut request = base_request();
+    request.previous_response_id = Some("resp_deflate_prev".to_string());
+
+    let response = client
+        .create_response(&request, request_context("req_ws_deflate", None))
+        .await
+        .unwrap();
+
+    server.await.unwrap();
+    assert!(response.body.contains("event: response.completed"));
+    assert!(response.body.contains("\"id\":\"resp_ws_deflate\""));
+    assert_eq!(response.usage.unwrap().input_tokens, 5);
+}
+
 #[tokio::test]
 #[expect(
     clippy::result_large_err,
@@ -1038,6 +1083,72 @@ async fn read_http_upgrade_request(stream: &mut tokio::net::TcpStream) -> String
     String::from_utf8(request).unwrap()
 }
 
+async fn read_client_websocket_frame(stream: &mut tokio::net::TcpStream) -> Vec<u8> {
+    let mut header = [0_u8; 2];
+    stream.read_exact(&mut header).await.unwrap();
+    let mut payload_len = u64::from(header[1] & 0x7f);
+    if payload_len == 126 {
+        let mut extended = [0_u8; 2];
+        stream.read_exact(&mut extended).await.unwrap();
+        payload_len = u64::from(u16::from_be_bytes(extended));
+    } else if payload_len == 127 {
+        let mut extended = [0_u8; 8];
+        stream.read_exact(&mut extended).await.unwrap();
+        payload_len = u64::from_be_bytes(extended);
+    }
+    let mut mask = [0_u8; 4];
+    if header[1] & 0x80 != 0 {
+        stream.read_exact(&mut mask).await.unwrap();
+    }
+    let mut payload = vec![0_u8; payload_len as usize];
+    stream.read_exact(&mut payload).await.unwrap();
+    if header[1] & 0x80 != 0 {
+        for (index, byte) in payload.iter_mut().enumerate() {
+            *byte ^= mask[index % mask.len()];
+        }
+    }
+    payload
+}
+
+fn websocket_accept_key(raw_request: &str) -> String {
+    let key = raw_request
+        .lines()
+        .find_map(|line| line.strip_prefix("Sec-WebSocket-Key: "))
+        .expect("raw websocket request should contain Sec-WebSocket-Key");
+    derive_accept_key(key.as_bytes())
+}
+
+fn compressed_server_text_frame(text: &str) -> Vec<u8> {
+    let mut compressor = Compress::new(Compression::fast(), false);
+    let mut payload = Vec::with_capacity(text.len() + 32);
+    compressor
+        .compress_vec(text.as_bytes(), &mut payload, FlushCompress::Sync)
+        .unwrap();
+    if payload.ends_with(&[0x00, 0x00, 0xff, 0xff]) {
+        payload.truncate(payload.len() - 4);
+    }
+    encode_server_frame(0x1, true, &payload)
+}
+
+fn encode_server_frame(opcode: u8, rsv1: bool, payload: &[u8]) -> Vec<u8> {
+    let mut frame = Vec::new();
+    let first = 0x80 | if rsv1 { 0x40 } else { 0 } | opcode;
+    frame.push(first);
+    match payload.len() {
+        len @ 0..=125 => frame.push(len as u8),
+        len @ 126..=65_535 => {
+            frame.push(126);
+            frame.extend_from_slice(&(len as u16).to_be_bytes());
+        }
+        len => {
+            frame.push(127);
+            frame.extend_from_slice(&(len as u64).to_be_bytes());
+        }
+    }
+    frame.extend_from_slice(payload);
+    frame
+}
+
 fn assert_headers_appear_in_order(raw_request: &str, expected_headers: &[&str]) {
     let mut offset = 0;
     for expected in expected_headers {