From a498875591ed27a9755035bba56e43783d1b651f Mon Sep 17 00:00:00 2001 From: AAEE86 Date: Fri, 31 Jul 2026 10:59:24 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20Responses=20WebSocket=20?= =?UTF-8?q?=E8=BF=9E=E9=80=9A=E6=80=A7=E6=8E=A2=E9=92=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 aether-codex-ws-probe 与 aether-openai-responses-ws-probe 两个 二进制，用于在不暴露凭据的前提下验证上游 WebSocket 端点可用性：凭据 只从环境变量读取，不写入日志。公共流程放在 bin/support/responses_ws_probe.rs，各 profile 只负责自己的鉴权与 请求头要求。 --- .../src/bin/aether-codex-ws-probe.rs | 124 ++++ .../bin/aether-openai-responses-ws-probe.rs | 104 ++++ .../src/bin/support/responses_ws_probe.rs | 567 ++++++++++++++++++ .../codex-responses-websocket-probe.md | 158 +++++ .../openai-responses-websocket-probe.md | 61 ++ 5 files changed, 1014 insertions(+) create mode 100644 apps/aether-gateway/src/bin/aether-codex-ws-probe.rs create mode 100644 apps/aether-gateway/src/bin/aether-openai-responses-ws-probe.rs create mode 100644 apps/aether-gateway/src/bin/support/responses_ws_probe.rs create mode 100644 docs/operations/codex-responses-websocket-probe.md create mode 100644 docs/operations/openai-responses-websocket-probe.md diff --git a/apps/aether-gateway/src/bin/aether-codex-ws-probe.rs b/apps/aether-gateway/src/bin/aether-codex-ws-probe.rs new file mode 100644 index 000000000..b9def32c9 --- /dev/null +++ b/apps/aether-gateway/src/bin/aether-codex-ws-probe.rs @@ -0,0 +1,124 @@ +//! Credential-safe compatibility probe for the Codex Responses WebSocket path. +//! +//! This binary preserves the established Codex CLI and environment contract. +//! The common Responses WebSocket flow lives in `support/responses_ws_probe`; +//! this profile owns only Codex authentication and header requirements. + +#[path = "support/responses_ws_probe.rs"] +mod responses_ws_probe; + +use aether_gateway::{CODEX_CLIENT_ORIGINATOR, CODEX_CLIENT_USER_AGENT}; +use clap::Parser; +use http::header::{AUTHORIZATION, USER_AGENT}; +use http::{HeaderMap, HeaderName, HeaderValue}; +use responses_ws_probe::{ + bearer_authorization_value, required_env, resolve_probe_url, run_profile_probe, turn_timeout, + ProbeArgs, ProbeConfig, ProbeFailure, ResponsesWebSocketProbeProfile, +}; + +const ACCESS_TOKEN_ENV: &str = "AETHER_CODEX_WS_PROBE_ACCESS_TOKEN"; +const ACCOUNT_ID_ENV: &str = "AETHER_CODEX_WS_PROBE_ACCOUNT_ID"; +const MODEL_ENV: &str = "AETHER_CODEX_WS_PROBE_MODEL"; +const URL_ENV: &str = "AETHER_CODEX_WS_PROBE_URL"; + +#[derive(Parser)] +#[command( + name = "aether-codex-ws-probe", + about = "Verify a Codex Responses WebSocket endpoint without exposing credentials" +)] +struct Args { + /// WebSocket endpoint. If omitted, AETHER_CODEX_WS_PROBE_URL is used. + #[arg(long)] + url: Option, + + /// Per-turn receive timeout in seconds. + #[arg(long, default_value_t = 20, value_parser = clap::value_parser!(u64).range(1..=120))] + timeout_secs: u64, +} + +impl From for ProbeArgs { + fn from(args: Args) -> Self { + Self { + url: args.url, + timeout_secs: args.timeout_secs, + } + } +} + +struct CodexResponsesProbeProfile; + +impl ResponsesWebSocketProbeProfile for CodexResponsesProbeProfile { + fn build_config(args: &ProbeArgs) -> Result { + let url = resolve_probe_url(args, URL_ENV, None)?; + let access_token = required_env(ACCESS_TOKEN_ENV)?; + let account_id = required_env(ACCOUNT_ID_ENV)?; + let model = required_env(MODEL_ENV)?; + Ok(ProbeConfig::new( + url, + model, + turn_timeout(args), + handshake_headers(&access_token, &account_id)?, + Self::sent_header_names(), + )) + } + + fn sent_header_names() -> Vec<&'static str> { + vec![ + "authorization", + "chatgpt-account-id", + "user-agent", + "originator", + ] + } +} + +fn handshake_headers(access_token: &str, account_id: &str) -> Result { + let account_id = + HeaderValue::from_str(account_id).map_err(|_| ProbeFailure::MissingConfiguration)?; + let mut headers = HeaderMap::new(); + headers.insert(AUTHORIZATION, bearer_authorization_value(access_token)?); + headers.insert(HeaderName::from_static("chatgpt-account-id"), account_id); + headers.insert( + USER_AGENT, + HeaderValue::from_static(CODEX_CLIENT_USER_AGENT), + ); + headers.insert( + HeaderName::from_static("originator"), + HeaderValue::from_static(CODEX_CLIENT_ORIGINATOR), + ); + Ok(headers) +} + +#[tokio::main] +async fn main() { + let exit_code = run_profile_probe::(Args::parse().into()).await; + if exit_code != 0 { + std::process::exit(exit_code); + } +} + +#[cfg(test)] +mod tests { + use http::header::{AUTHORIZATION, USER_AGENT}; + + use super::{handshake_headers, CodexResponsesProbeProfile, ResponsesWebSocketProbeProfile}; + + #[test] + fn codex_profile_keeps_its_required_handshake_headers() { + let headers = + handshake_headers("test-token", "test-account").expect("headers should build"); + assert!(headers.contains_key(AUTHORIZATION)); + assert!(headers.contains_key("chatgpt-account-id")); + assert!(headers.contains_key(USER_AGENT)); + assert!(headers.contains_key("originator")); + assert_eq!( + CodexResponsesProbeProfile::sent_header_names(), + vec![ + "authorization", + "chatgpt-account-id", + "user-agent", + "originator", + ] + ); + } +} diff --git a/apps/aether-gateway/src/bin/aether-openai-responses-ws-probe.rs b/apps/aether-gateway/src/bin/aether-openai-responses-ws-probe.rs new file mode 100644 index 000000000..08838fc26 --- /dev/null +++ b/apps/aether-gateway/src/bin/aether-openai-responses-ws-probe.rs @@ -0,0 +1,104 @@ +//! Credential-safe compatibility probe for the official OpenAI Responses +//! WebSocket endpoint. +//! +//! This profile uses standard API-key Bearer authentication and shares the +//! protocol flow with the Codex probe without inheriting Codex-specific +//! account headers or quota assumptions. + +#[path = "support/responses_ws_probe.rs"] +mod responses_ws_probe; + +use clap::Parser; +use http::header::AUTHORIZATION; +use http::HeaderMap; +use responses_ws_probe::{ + bearer_authorization_value, required_env, resolve_probe_url, run_profile_probe, turn_timeout, + ProbeArgs, ProbeConfig, ProbeFailure, ResponsesWebSocketProbeProfile, +}; + +const API_KEY_ENV: &str = "AETHER_OPENAI_WS_PROBE_API_KEY"; +const MODEL_ENV: &str = "AETHER_OPENAI_WS_PROBE_MODEL"; +const URL_ENV: &str = "AETHER_OPENAI_WS_PROBE_URL"; +const DEFAULT_URL: &str = "wss://api.openai.com/v1/responses"; + +#[derive(Parser)] +#[command( + name = "aether-openai-responses-ws-probe", + about = "Verify an OpenAI Responses WebSocket endpoint without exposing credentials" +)] +struct Args { + /// WebSocket endpoint. If omitted, AETHER_OPENAI_WS_PROBE_URL or the + /// official OpenAI endpoint is used. + #[arg(long)] + url: Option, + + /// Per-turn receive timeout in seconds. + #[arg(long, default_value_t = 20, value_parser = clap::value_parser!(u64).range(1..=120))] + timeout_secs: u64, +} + +impl From for ProbeArgs { + fn from(args: Args) -> Self { + Self { + url: args.url, + timeout_secs: args.timeout_secs, + } + } +} + +struct OpenAiResponsesProbeProfile; + +impl ResponsesWebSocketProbeProfile for OpenAiResponsesProbeProfile { + fn build_config(args: &ProbeArgs) -> Result { + let url = resolve_probe_url(args, URL_ENV, Some(DEFAULT_URL))?; + let api_key = required_env(API_KEY_ENV)?; + let model = required_env(MODEL_ENV)?; + let mut headers = HeaderMap::new(); + headers.insert(AUTHORIZATION, bearer_authorization_value(&api_key)?); + Ok(ProbeConfig::new( + url, + model, + turn_timeout(args), + headers, + Self::sent_header_names(), + )) + } + + fn sent_header_names() -> Vec<&'static str> { + vec!["authorization"] + } +} + +#[tokio::main] +async fn main() { + let exit_code = run_profile_probe::(Args::parse().into()).await; + if exit_code != 0 { + std::process::exit(exit_code); + } +} + +#[cfg(test)] +mod tests { + use http::header::AUTHORIZATION; + + use super::{ + bearer_authorization_value, responses_ws_probe::parse_probe_url, + OpenAiResponsesProbeProfile, ResponsesWebSocketProbeProfile, DEFAULT_URL, + }; + + #[test] + fn openai_profile_exposes_only_standard_bearer_authentication() { + let authorization = bearer_authorization_value("test-key").expect("header should build"); + assert_eq!(authorization.to_str().ok(), Some("Bearer test-key")); + assert_eq!( + OpenAiResponsesProbeProfile::sent_header_names(), + vec![AUTHORIZATION.as_str()] + ); + } + + #[test] + fn openai_profile_uses_the_official_responses_websocket_endpoint_by_default() { + let url = parse_probe_url(DEFAULT_URL).expect("default OpenAI endpoint should be valid"); + assert_eq!(url.as_str(), DEFAULT_URL); + } +} diff --git a/apps/aether-gateway/src/bin/support/responses_ws_probe.rs b/apps/aether-gateway/src/bin/support/responses_ws_probe.rs new file mode 100644 index 000000000..061a97cf6 --- /dev/null +++ b/apps/aether-gateway/src/bin/support/responses_ws_probe.rs @@ -0,0 +1,567 @@ +//! Shared, credential-safe engine for Responses WebSocket compatibility probes. +//! +//! Provider profiles own their environment variables and handshake headers. +//! This module owns the common Responses WebSocket contract: two sequential +//! `response.create` warmups, continuation with `previous_response_id`, safe +//! event observation, and a redacted JSON report. + +use std::env; +use std::time::{Duration, Instant}; + +use http::{HeaderMap, HeaderValue}; +use serde::Serialize; +use serde_json::{json, Value}; +use url::Url; +use wreq::ws::message::Message as WreqWsMessage; + +const MAX_FRAME_SIZE: usize = 1 << 20; +const MAX_EVENTS_PER_TURN: usize = 16; + +pub(crate) struct ProbeArgs { + pub(crate) url: Option, + pub(crate) timeout_secs: u64, +} + +pub(crate) struct ProbeConfig { + url: Url, + model: String, + turn_timeout: Duration, + handshake_headers: HeaderMap, + sent_header_names: Vec<&'static str>, +} + +impl ProbeConfig { + pub(crate) fn new( + url: Url, + model: String, + turn_timeout: Duration, + handshake_headers: HeaderMap, + sent_header_names: Vec<&'static str>, + ) -> Self { + Self { + url, + model, + turn_timeout, + handshake_headers, + sent_header_names, + } + } +} + +/// A profile retains provider-specific authentication and configuration while +/// reusing one Responses protocol probe engine. +pub(crate) trait ResponsesWebSocketProbeProfile { + fn build_config(args: &ProbeArgs) -> Result; + fn sent_header_names() -> Vec<&'static str>; +} + +#[derive(Debug, Clone, Copy)] +pub(crate) enum ProbeFailure { + MissingConfiguration, + InvalidEndpoint, + ClientBuild, + Handshake, + Upgrade, + Send, + ReceiveTimeout, + Receive, + RemoteError, + MissingResponseId, + UnexpectedFrame, +} + +impl ProbeFailure { + const fn code(self) -> &'static str { + match self { + Self::MissingConfiguration => "missing_configuration", + Self::InvalidEndpoint => "invalid_endpoint", + Self::ClientBuild => "client_build_failed", + Self::Handshake => "handshake_failed", + Self::Upgrade => "upgrade_failed", + Self::Send => "send_failed", + Self::ReceiveTimeout => "receive_timeout", + Self::Receive => "receive_failed", + Self::RemoteError => "upstream_error_event", + Self::MissingResponseId => "response_id_not_observed", + Self::UnexpectedFrame => "unexpected_frame", + } + } +} + +#[derive(Serialize)] +struct ProbeReport { + status: &'static str, + target_host: Option, + handshake_status: Option, + sent_header_names: Vec<&'static str>, + received_header_names: Vec, + observed_event_types: Vec, + continuation_confirmed: bool, + elapsed_ms: u64, + error: Option<&'static str>, +} + +impl ProbeReport { + fn failed( + config: Option<&ProbeConfig>, + sent_header_names: Vec<&'static str>, + started_at: Instant, + error: ProbeFailure, + ) -> Self { + Self { + status: "failed", + target_host: config.and_then(target_host), + handshake_status: None, + sent_header_names, + received_header_names: Vec::new(), + observed_event_types: Vec::new(), + continuation_confirmed: false, + elapsed_ms: started_at.elapsed().as_millis() as u64, + error: Some(error.code()), + } + } +} + +/// Runs a profile and returns the process exit code after emitting exactly one +/// credential-safe JSON report. +pub(crate) async fn run_profile_probe(args: ProbeArgs) -> i32 { + let started_at = Instant::now(); + let config = match P::build_config(&args) { + Ok(config) => config, + Err(error) => { + print_report(&ProbeReport::failed( + None, + P::sent_header_names(), + started_at, + error, + )); + return 2; + } + }; + + match run_probe(&config, started_at).await { + Ok(report) => { + print_report(&report); + 0 + } + Err(error) => { + print_report(&ProbeReport::failed( + Some(&config), + config.sent_header_names.clone(), + started_at, + error, + )); + 1 + } + } +} + +pub(crate) fn required_env(name: &str) -> Result { + env::var(name) + .ok() + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) + .ok_or(ProbeFailure::MissingConfiguration) +} + +pub(crate) fn resolve_probe_url( + args: &ProbeArgs, + url_env: &str, + default_url: Option<&str>, +) -> Result { + let raw_url = args + .url + .as_deref() + .map(str::to_owned) + .or_else(|| env::var(url_env).ok()) + .or_else(|| default_url.map(str::to_owned)); + let Some(raw_url) = raw_url + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()) + else { + return Err(ProbeFailure::MissingConfiguration); + }; + parse_probe_url(raw_url) +} + +pub(crate) fn parse_probe_url(raw: &str) -> Result { + let url = Url::parse(raw).map_err(|_| ProbeFailure::InvalidEndpoint)?; + if !matches!(url.scheme(), "ws" | "wss") + || url.host_str().is_none() + || !url.username().is_empty() + || url.password().is_some() + || url.query().is_some() + || url.fragment().is_some() + { + return Err(ProbeFailure::InvalidEndpoint); + } + Ok(url) +} + +pub(crate) fn bearer_authorization_value(token: &str) -> Result { + HeaderValue::from_str(format!("Bearer {token}").as_str()) + .map_err(|_| ProbeFailure::MissingConfiguration) +} + +pub(crate) const fn turn_timeout(args: &ProbeArgs) -> Duration { + Duration::from_secs(args.timeout_secs) +} + +async fn run_probe(config: &ProbeConfig, started_at: Instant) -> Result { + let client = wreq::Client::builder() + .connect_timeout(config.turn_timeout) + .timeout(config.turn_timeout) + .build() + .map_err(|_| ProbeFailure::ClientBuild)?; + let response = client + .websocket(config.url.as_str()) + .headers(config.handshake_headers.clone()) + .max_frame_size(MAX_FRAME_SIZE) + .max_message_size(MAX_FRAME_SIZE) + .send() + .await + .map_err(|_| ProbeFailure::Handshake)?; + let handshake_status = response.status().as_u16(); + let received_header_names = response + .headers() + .keys() + .map(|name| name.as_str().to_string()) + .collect(); + let mut socket = response + .into_websocket() + .await + .map_err(|_| ProbeFailure::Upgrade)?; + let mut observed_event_types = Vec::new(); + + send_warmup(&mut socket, &config.model, None).await?; + let first_response_id = + receive_completed_response_id(&mut socket, config.turn_timeout, &mut observed_event_types) + .await?; + + send_warmup(&mut socket, &config.model, Some(&first_response_id)).await?; + let _second_response_id = + receive_completed_response_id(&mut socket, config.turn_timeout, &mut observed_event_types) + .await?; + + Ok(ProbeReport { + status: "passed", + target_host: target_host(config), + handshake_status: Some(handshake_status), + sent_header_names: config.sent_header_names.clone(), + received_header_names, + observed_event_types, + continuation_confirmed: true, + elapsed_ms: started_at.elapsed().as_millis() as u64, + error: None, + }) +} + +fn target_host(config: &ProbeConfig) -> Option { + config.url.host_str().map(|host| match config.url.port() { + Some(port) => format!("{host}:{port}"), + None => host.to_string(), + }) +} + +async fn send_warmup( + socket: &mut wreq::ws::WebSocket, + model: &str, + previous_response_id: Option<&str>, +) -> Result<(), ProbeFailure> { + let mut event = json!({ + "type": "response.create", + "model": model, + "store": false, + "generate": false, + "input": [], + "tools": [], + }); + if let Some(previous_response_id) = previous_response_id { + event["previous_response_id"] = Value::String(previous_response_id.to_string()); + } + socket + .send(WreqWsMessage::text(event.to_string())) + .await + .map_err(|_| ProbeFailure::Send) +} + +async fn receive_completed_response_id( + socket: &mut wreq::ws::WebSocket, + timeout: Duration, + observed_event_types: &mut Vec, +) -> Result { + let mut response_id = None; + for _ in 0..MAX_EVENTS_PER_TURN { + let message = tokio::time::timeout(timeout, socket.recv()) + .await + .map_err(|_| ProbeFailure::ReceiveTimeout)? + .ok_or(ProbeFailure::MissingResponseId)? + .map_err(|_| ProbeFailure::Receive)?; + match message { + WreqWsMessage::Text(text) => { + let event: Value = serde_json::from_str(text.as_str()) + .map_err(|_| ProbeFailure::UnexpectedFrame)?; + let event_type = event + .get("type") + .and_then(Value::as_str) + .map(safe_event_label) + .unwrap_or_else(|| "unknown".to_string()); + let is_remote_error = event_type == "error"; + let is_completed = event_type == "response.completed"; + observed_event_types.push(event_type); + if is_remote_error { + return Err(ProbeFailure::RemoteError); + } + if let Some(observed_response_id) = event + .pointer("/response/id") + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + { + response_id = Some(observed_response_id.to_string()); + } + if is_completed { + return response_id.ok_or(ProbeFailure::MissingResponseId); + } + } + WreqWsMessage::Ping(_) | WreqWsMessage::Pong(_) => continue, + WreqWsMessage::Close(_) => return Err(ProbeFailure::MissingResponseId), + _ => return Err(ProbeFailure::UnexpectedFrame), + } + } + Err(ProbeFailure::MissingResponseId) +} + +fn safe_event_label(value: &str) -> String { + let trimmed = value.trim(); + if trimmed.is_empty() + || trimmed.len() > 80 + || !trimmed + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')) + { + return "unknown".to_string(); + } + trimmed.to_string() +} + +fn print_report(report: &ProbeReport) { + match serde_json::to_string(report) { + Ok(json) => println!("{json}"), + Err(_) => println!("{{\"status\":\"failed\",\"error\":\"report_serialization_failed\"}}"), + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::{Duration, Instant}; + + use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; + use axum::extract::State; + use axum::http::header::AUTHORIZATION; + use axum::http::{HeaderMap, HeaderValue}; + use axum::response::IntoResponse; + use axum::routing::get; + use axum::Router; + use futures_util::{SinkExt, StreamExt}; + use serde_json::Value; + use tokio::sync::{oneshot, Mutex}; + + use super::{parse_probe_url, run_probe, ProbeConfig}; + + #[derive(Default)] + struct MockState { + observed: Mutex>>, + } + + struct ObservedClientMessages { + authorization_present: bool, + profile_header_present: bool, + second_before_first_completion: bool, + first: Value, + second: Value, + } + + #[tokio::test] + async fn probe_confirms_sequential_response_continuation_without_exposing_values() { + let (url, observed, server) = spawn_mock_server().await; + let mut headers = HeaderMap::new(); + headers.insert( + AUTHORIZATION, + HeaderValue::from_static("Bearer test-token-that-must-not-be-reported"), + ); + headers.insert( + "x-aether-probe-profile", + HeaderValue::from_static("test-profile-id"), + ); + let config = ProbeConfig::new( + parse_probe_url(url.as_str()).expect("mock URL should be valid"), + "gpt-test".to_string(), + Duration::from_secs(2), + headers, + vec!["authorization", "x-aether-probe-profile"], + ); + + let report = run_probe(&config, Instant::now()) + .await + .expect("probe should complete against mock server"); + let client_messages = observed.await.expect("mock should observe client messages"); + server.abort(); + + assert_eq!(report.status, "passed"); + assert!(report.continuation_confirmed); + assert!(report + .observed_event_types + .contains(&"response.created".to_string())); + assert!(report + .observed_event_types + .contains(&"response.completed".to_string())); + assert!(client_messages.authorization_present); + assert!(client_messages.profile_header_present); + assert!(!client_messages.second_before_first_completion); + assert_eq!(client_messages.first["type"], "response.create"); + assert_eq!(client_messages.first["generate"], false); + assert_eq!(client_messages.first["store"], false); + assert_eq!(client_messages.second["previous_response_id"], "resp-first"); + let report_json = serde_json::to_string(&report).expect("report should serialize"); + assert!(!report_json.contains("test-token-that-must-not-be-reported")); + assert!(!report_json.contains("test-profile-id")); + assert!(!report_json.contains("resp-first")); + } + + #[test] + fn probe_url_rejects_credentials_and_query_strings() { + assert!(parse_probe_url("wss://example.test/v1/responses").is_ok()); + assert!(parse_probe_url("https://example.test/v1/responses").is_err()); + assert!(parse_probe_url("wss://token@example.test/v1/responses").is_err()); + assert!(parse_probe_url("wss://example.test/v1/responses?token=secret").is_err()); + } + + async fn spawn_mock_server() -> ( + String, + oneshot::Receiver, + tokio::task::JoinHandle<()>, + ) { + let (observed_tx, observed_rx) = oneshot::channel(); + let state = Arc::new(MockState { + observed: Mutex::new(Some(observed_tx)), + }); + let app = Router::new() + .route("/v1/responses", get(mock_websocket)) + .with_state(state); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("mock listener should bind"); + let address = listener + .local_addr() + .expect("mock listener should expose address"); + let server = tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("mock server should run"); + }); + (format!("ws://{address}/v1/responses"), observed_rx, server) + } + + async fn mock_websocket( + ws: WebSocketUpgrade, + State(state): State>, + headers: HeaderMap, + ) -> impl IntoResponse { + let authorization_present = headers + .get(AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value.starts_with("Bearer ")); + let profile_header_present = headers.contains_key("x-aether-probe-profile"); + ws.on_upgrade(move |socket| async move { + serve_mock_socket(socket, state, authorization_present, profile_header_present).await; + }) + } + + async fn serve_mock_socket( + socket: WebSocket, + state: Arc, + authorization_present: bool, + profile_header_present: bool, + ) { + let (mut sender, mut receiver) = socket.split(); + let first = receive_json(&mut receiver).await; + let _ = sender + .send(Message::Text( + serde_json::json!({ + "type": "response.created", + "response": {"id": "resp-first"} + }) + .to_string() + .into(), + )) + .await; + let early_second = tokio::select! { + message = receiver.next() => Some(message), + _ = tokio::time::sleep(Duration::from_millis(50)) => None, + }; + let second_before_first_completion = early_second.is_some(); + let _ = sender + .send(Message::Text( + serde_json::json!({ + "type": "response.completed", + "response": {"id": "resp-first", "status": "completed"} + }) + .to_string() + .into(), + )) + .await; + let second = match early_second { + Some(Some(Ok(Message::Text(text)))) => { + serde_json::from_str(text.as_str()).expect("early client message should be JSON") + } + Some(Some(Ok(_))) => panic!("expected text continuation message"), + Some(Some(Err(error))) => panic!("client message should be valid: {error}"), + Some(None) => panic!("client closed before continuation"), + None => receive_json(&mut receiver).await, + }; + let _ = sender + .send(Message::Text( + serde_json::json!({ + "type": "response.created", + "response": {"id": "resp-second"} + }) + .to_string() + .into(), + )) + .await; + let _ = sender + .send(Message::Text( + serde_json::json!({ + "type": "response.completed", + "response": {"id": "resp-second", "status": "completed"} + }) + .to_string() + .into(), + )) + .await; + if let Some(observed) = state.observed.lock().await.take() { + let _ = observed.send(ObservedClientMessages { + authorization_present, + profile_header_present, + second_before_first_completion, + first, + second, + }); + } + } + + async fn receive_json(receiver: &mut futures_util::stream::SplitStream) -> Value { + let message = receiver + .next() + .await + .expect("client should send a message") + .expect("client message should be valid"); + let Message::Text(text) = message else { + panic!("expected text message"); + }; + serde_json::from_str(text.as_str()).expect("client message should be JSON") + } +} diff --git a/docs/operations/codex-responses-websocket-probe.md b/docs/operations/codex-responses-websocket-probe.md new file mode 100644 index 000000000..b6987af1e --- /dev/null +++ b/docs/operations/codex-responses-websocket-probe.md @@ -0,0 +1,158 @@ +# Codex Responses WebSocket probe + +`aether-codex-ws-probe` is a P0 compatibility probe for a Codex-compatible +Responses WebSocket upstream. It verifies two sequential `response.create` +warmups on one socket, with the second request continuing from the first +response ID. + +The command, environment variables, JSON report shape, and Codex-specific +handshake headers remain stable. It now shares only the protocol-driving core +with the separate [OpenAI Responses WebSocket probe](openai-responses-websocket-probe.md); +the two probes intentionally retain independent authentication profiles and +provider-specific assertions. + +The probe is intentionally not a production proxy. It does not persist, +refresh, log, or print credentials, account IDs, response IDs, request bodies, +or response bodies. + +## Prerequisites + +Use a dedicated, non-production Codex test account. Rotate any credential that +has been pasted into a chat, terminal history, issue, or source file before +using this probe. + +Set these values only in the process environment or your secret manager: + +```bash +export AETHER_CODEX_WS_PROBE_URL='wss://your-codex-upstream.example/backend-api/codex/responses' +export AETHER_CODEX_WS_PROBE_ACCESS_TOKEN='your-short-lived-access-token' +export AETHER_CODEX_WS_PROBE_ACCOUNT_ID='your-account-id' +export AETHER_CODEX_WS_PROBE_MODEL='your-codex-model' +``` + +The endpoint must use `ws://` or `wss://`, with no credentials, query string, +or fragment. The access token is accepted only through +`AETHER_CODEX_WS_PROBE_ACCESS_TOKEN`; there is deliberately no command-line +flag for it. + +## Run + +```bash +cargo run -p aether-gateway --bin aether-codex-ws-probe +``` + +Use `--url` to override only the endpoint and `--timeout-secs` to set a +per-turn receive timeout (1–120 seconds): + +```bash +cargo run -p aether-gateway --bin aether-codex-ws-probe -- \ + --url 'wss://your-codex-upstream.example/backend-api/codex/responses' \ + --timeout-secs 30 +``` + +The probe emits one JSON line. A successful run has +`"continuation_confirmed":true`; its event and header fields contain names +only, never values. A failure emits a stable error code such as +`"handshake_failed"`, `"upstream_error_event"`, or +`"response_id_not_observed"`. + +## Interpretation + +A successful probe establishes that the selected upstream accepts the +Responses WebSocket handshake and retains continuation state on one socket. +It does not establish that all Codex models, account plans, or tunnel egress +paths are supported. In particular, the current `aether-tunnel` HTTP relay +does not forward WebSocket upgrades, so a successful direct probe is a +prerequisite rather than tunnel support. + +## Gateway bridge + +The gateway exposes WebSocket mode at the same public Responses path: + +```text +wss:///v1/responses +``` + +It is disabled by default per provider. In **添加提供商** or **编辑提供商**, +enable **Responses WebSocket 模式** under **功能开关** only after the selected +upstream has passed a compatible WebSocket probe. The setting takes effect for +new WebSocket connections without a gateway restart. It is available to every +provider type; candidate planning still requires a selected +`openai:responses` endpoint. + +Authenticate the upgrade request with the normal Aether API key. The first +client frame must be a text JSON `response.create` containing a non-empty +`model`. Aether then applies its regular Responses candidate selection, but +accepts only an eligible, WebSocket-enabled endpoint using `openai:responses`. +It opens an upstream WebSocket using the selected provider key. + +The selected provider's model mapping and request headers are applied to every +turn, along with the rest of that candidate's provider-body normalization: +model-directive patches, endpoint body rules, and the Codex body contract +(unsupported-field stripping, `store: false`, `tool_choice` defaulting). A +continuation turn is normalized against the binding it is pinned to rather than +being re-planned, so it can never move to another provider key. +`previous_response_id` and `generate` are re-applied after normalization +because they are WebSocket protocol state that the provider body contract +otherwise strips. `stream` and `background` are removed because they are HTTP +transport fields, not WebSocket-mode fields. If a later `response.create` changes the +public model, Aether runs access checks and candidate planning again. It keeps +the existing upstream when the same target remains eligible, or transparently +replaces the upstream between responses when the selected target changes. +Overlapping responses on one client socket remain rejected. + +Each `response.create` is tracked as an independent Aether logical request: +it receives its own request/candidate identity, usage lifecycle, and terminal +audit record. `response.completed`, `response.failed`, +`response.incomplete`, `response.cancelled`, client disconnects, and upstream +transport failures all settle that turn through the existing stream reporting +path. + +Example client setup: + +```python +from websocket import create_connection +import json +import os + +ws = create_connection( + "wss://gateway.example/v1/responses", + header=[f"Authorization: Bearer {os.environ['AETHER_API_KEY']}"], +) +ws.send(json.dumps({ + "type": "response.create", + "model": "your-public-model", + "store": False, + "input": "Explain this repository.", +})) +``` + +### Operating limits + +- Maximum frame and message size: 16 MiB. +- An idle connection must send its first `response.create` within 60 seconds. +- A connection is closed after 60 minutes; reconnect before then for long runs. +- Each `response.create` must receive its first upstream event within the + selected provider's `stream_first_byte_timeout` (30 seconds by default), + and finish within its `request_timeout` (20 minutes by default). Aether + sends `responses_websocket_first_event_timeout` or + `responses_websocket_turn_timeout` and closes the bound socket when either + deadline expires. +- Responses are sequential; no multiplexing is supported on one socket. +- Each `response.create` consumes the normal Aether user/API-key RPM budget. +- Same-model turns stay on the bound provider key. A model change is planned + again and can rebind the upstream between completed turns when necessary. +- Direct provider proxy settings are honored through the selected transport + profile. Tunnel-mode proxy nodes are not supported for this bridge yet. + +Usage and audit finalization now runs for every accepted `response.create`. +Existing usage body-capture and header-redaction policies apply to the resulting +records. Newly created WebSocket usage records expose `is_websocket=true`, and +the usage-record type column renders them as `WS`. For diagnosis, enable debug +logging for `aether_gateway::handlers::proxy::responses_ws`; event logs contain +only the event type and frame size, never request or response contents. Codex +quota-extension logs remain under `aether_gateway::handlers::proxy::codex_ws`. +Every WebSocket-specific log carries `transport="websocket"` and +`websocket=true`; keep `log_type` for its existing access/event/ops +classification, and render the transport flag as a `WS` label in a log viewer +if desired. diff --git a/docs/operations/openai-responses-websocket-probe.md b/docs/operations/openai-responses-websocket-probe.md new file mode 100644 index 000000000..b7a33c7d8 --- /dev/null +++ b/docs/operations/openai-responses-websocket-probe.md @@ -0,0 +1,61 @@ +# OpenAI Responses WebSocket probe + +`aether-openai-responses-ws-probe` verifies the official OpenAI Responses +WebSocket protocol using standard API-key Bearer authentication. It sends two +sequential `response.create` warmups on one socket, chaining the second from +the first response ID with `previous_response_id`. + +It shares its protocol-driving core with the Codex probe, but it does **not** +send Codex account headers or require Codex quota events. This makes it the +compatibility gate for Aether's standard Responses WebSocket adapter, rather +than a replacement for the Codex probe. + +## Prerequisites + +Use a dedicated API project and a model that your key can access. Keep values +only in your process environment or secret manager: + +```bash +export AETHER_OPENAI_WS_PROBE_API_KEY='your-api-key' +export AETHER_OPENAI_WS_PROBE_MODEL='your-openai-model' +``` + +The default endpoint is the official Responses WebSocket endpoint: + +```text +wss://api.openai.com/v1/responses +``` + +To test a compatible endpoint explicitly, set +`AETHER_OPENAI_WS_PROBE_URL` or pass `--url`. The endpoint must use `ws://` or +`wss://` and may not contain credentials, a query string, or a fragment. The +API key has no command-line flag and is never printed. + +## Run + +```bash +cargo run -p aether-gateway --bin aether-openai-responses-ws-probe +``` + +For an explicit endpoint and timeout: + +```bash +cargo run -p aether-gateway --bin aether-openai-responses-ws-probe -- \ + --url 'wss://api.openai.com/v1/responses' \ + --timeout-secs 30 +``` + +The probe uses `generate:false`, so the warmups prepare continuation state but +do not request model output. A successful JSON report contains +`"continuation_confirmed":true`; header and event arrays contain names only, +never credentials, response IDs, request bodies, or response bodies. + +## Interpretation + +Success establishes that this key, model, and endpoint support the Responses +WebSocket handshake plus an in-socket continuation. It does not establish +support for every model, tool, service tier, proxy path, or Aether provider +configuration. Treat a successful direct probe as a prerequisite before +enabling **Responses WebSocket mode** for the matching Aether provider. + +For protocol details, see the official [WebSocket Mode guide](https://developers.openai.com/api/docs/guides/websocket-mode).