From 621a528083060a6a098f70ee879495e43042c997 Mon Sep 17 00:00:00 2001 From: AAEE86 Date: Fri, 31 Jul 2026 10:59:24 +0800 Subject: [PATCH] =?UTF-8?q?test(ws):=20Responses=20WebSocket=20=E7=AB=AF?= =?UTF-8?q?=E5=88=B0=E7=AB=AF=E5=A5=97=E4=BB=B6=E6=8E=A5=E5=85=A5=20CI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 补齐 aether-integration-tests 的 responses_websocket_e2e 集成测试， 并把 CI 的 scenario 任务从 --bins 改为 --bins --tests，否则该套件 不会被执行。 --- .github/workflows/rust-ci.yml | 4 +- Cargo.lock | 2 + crates/aether-testing/integration/Cargo.toml | 2 + .../tests/responses_websocket_e2e.rs | 897 ++++++++++++++++++ 4 files changed, 903 insertions(+), 2 deletions(-) create mode 100644 crates/aether-testing/integration/tests/responses_websocket_e2e.rs diff --git a/.github/workflows/rust-ci.yml b/.github/workflows/rust-ci.yml index 76d14603f..af8295df2 100644 --- a/.github/workflows/rust-ci.yml +++ b/.github/workflows/rust-ci.yml @@ -387,11 +387,11 @@ jobs: - name: Setup sccache uses: mozilla-actions/sccache-action@v0.0.9 - - name: Test scenario binaries + - name: Test scenario binaries and end-to-end suites env: RUSTC_WRAPPER: sccache SCCACHE_GHA_ENABLED: "true" - run: cargo test -p aether-integration-tests --bins + run: cargo test -p aether-integration-tests --bins --tests - name: Show sccache stats if: always() diff --git a/Cargo.lock b/Cargo.lock index 7067f3cd3..32ff749e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -443,6 +443,7 @@ name = "aether-integration-tests" version = "0.1.0" dependencies = [ "aether-contracts", + "aether-crypto", "aether-data", "aether-data-contracts", "aether-gateway", @@ -459,6 +460,7 @@ dependencies = [ "sqlx", "tokio", "tokio-tungstenite 0.28.0", + "uuid", ] [[package]] diff --git a/crates/aether-testing/integration/Cargo.toml b/crates/aether-testing/integration/Cargo.toml index e93236fc3..eafd3d999 100644 --- a/crates/aether-testing/integration/Cargo.toml +++ b/crates/aether-testing/integration/Cargo.toml @@ -9,6 +9,7 @@ description = "Gateway-backed integration scenarios and benchmark binaries" [dependencies] async-stream.workspace = true aether-contracts.workspace = true +aether-crypto.workspace = true aether-data.workspace = true aether-data-contracts.workspace = true aether-gateway = { workspace = true, features = ["testkit"] } @@ -24,3 +25,4 @@ sha2.workspace = true sqlx = { workspace = true, features = ["postgres"] } tokio.workspace = true tokio-tungstenite = { version = "0.28", features = ["rustls-tls-webpki-roots"] } +uuid.workspace = true diff --git a/crates/aether-testing/integration/tests/responses_websocket_e2e.rs b/crates/aether-testing/integration/tests/responses_websocket_e2e.rs new file mode 100644 index 000000000..a89bae268 --- /dev/null +++ b/crates/aether-testing/integration/tests/responses_websocket_e2e.rs @@ -0,0 +1,897 @@ +//! Responses WebSocket end-to-end coverage. +//! +//! Every test starts a protocol-aware mock upstream, seeds a throwaway SQLite +//! store, mounts the real gateway router, and drives the public +//! `/v1/responses` WebSocket the way a client would. +//! +//! The assertions deliberately reach back into the database. A turn settles its +//! billing row from a task that outlives the relay loop, so a client that saw +//! `response.completed` is not evidence that the turn was ever accounted for — +//! only the row is. + +use std::path::PathBuf; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use aether_crypto::{encrypt_python_fernet_plaintext, DEVELOPMENT_ENCRYPTION_KEY}; +use aether_data::repository::auth::CreateStandaloneApiKeyRecord; +use aether_data::repository::wallet::WalletLookupKey; +use aether_data::{ + DataBackends, DataLayerConfig, DatabaseDriver, SqlDatabaseConfig, SqlPoolConfig, +}; +use aether_data_contracts::repository::global_models::{ + CreateAdminGlobalModelRecord, UpsertAdminProviderModelRecord, +}; +use aether_data_contracts::repository::provider_catalog::{ + StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogProvider, +}; +use aether_data_contracts::repository::usage::{StoredRequestUsageAudit, UsageAuditListQuery}; +use aether_gateway::{build_router_with_state, AppState, GatewayDataConfig, UsageRuntimeConfig}; +use aether_testkit::SpawnedServer; +use axum::extract::ws::{Message as AxumWsMessage, WebSocket, WebSocketUpgrade}; +use axum::extract::State; +use axum::http::HeaderMap; +use axum::response::Response; +use axum::routing::get; +use axum::Router; +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use sha2::Digest; +use tokio::net::TcpStream; +use tokio::sync::Mutex; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{MaybeTlsStream, WebSocketStream}; + +type BoxError = Box; +type ClientSocket = WebSocketStream>; + +const CLIENT_API_KEY: &str = "sk-aether-responses-ws-e2e"; +const PROVIDER_API_KEY: &str = "sk-upstream-responses-ws-e2e"; +const PROVIDER_ID: &str = "provider-responses-ws-e2e"; +const ENDPOINT_ID: &str = "endpoint-responses-ws-e2e"; +const PROVIDER_KEY_ID: &str = "provider-key-responses-ws-e2e"; +const GLOBAL_MODEL_ID: &str = "global-model-responses-ws-e2e"; +const PROVIDER_MODEL_ID: &str = "provider-model-responses-ws-e2e"; +const API_KEY_ID: &str = "api-key-responses-ws-e2e"; +const PUBLIC_MODEL: &str = "gpt-responses-ws-e2e"; +const UPSTREAM_MODEL: &str = "gpt-responses-ws-upstream"; + +const INPUT_TOKENS: u64 = 4; +const OUTPUT_TOKENS: u64 = 2; + +/// Generous enough to absorb a loaded CI runner, short enough that a genuinely +/// lost row fails the test instead of hanging the job. +const SETTLE_TIMEOUT: Duration = Duration::from_secs(30); +const RECEIVE_TIMEOUT: Duration = Duration::from_secs(15); + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +/// The headline guarantee: a continuation stays on one physical upstream socket, +/// and both turns are billed independently. +#[tokio::test] +async fn continuation_reuses_one_upstream_connection_and_bills_both_turns() -> Result<(), BoxError> +{ + let harness = Harness::start(UpstreamBehavior::CompleteEveryTurn).await?; + let mut client = harness.connect().await?; + + client + .send(response_create(json!({"input": "first turn"}))) + .await?; + let first = receive_event(&mut client, "response.completed").await?; + assert_eq!( + first.pointer("/response/id").and_then(Value::as_str), + Some("resp-e2e-1") + ); + + client + .send(response_create(json!({ + "previous_response_id": "resp-e2e-1", + "input": "second turn" + }))) + .await?; + let second = receive_event(&mut client, "response.completed").await?; + assert_eq!( + second.pointer("/response/id").and_then(Value::as_str), + Some("resp-e2e-2") + ); + + // The mock records each `response.create` before answering it, so both + // turns completing means both are already on record. + let upstream_events = harness.upstream.observed_events().await; + assert_eq!( + upstream_events.len(), + 2, + "one upstream turn per client turn" + ); + assert_eq!( + harness.upstream.connections(), + 1, + "the continuation must reuse the bound upstream socket" + ); + for event in &upstream_events { + assert_eq!( + event.get("model").and_then(Value::as_str), + Some(UPSTREAM_MODEL), + "every turn is rewritten to the mapped provider model" + ); + } + assert_eq!( + upstream_events[1] + .get("previous_response_id") + .and_then(Value::as_str), + Some("resp-e2e-1"), + "the continuation id survives provider body normalization" + ); + assert_eq!( + harness.upstream.authorization_headers().await, + vec![Some(format!("Bearer {PROVIDER_API_KEY}"))], + "the upstream is opened with the configured provider key" + ); + + let audits = harness + .usage_audits_where(2, "billed turns", is_billed) + .await?; + assert_eq!(audits.len(), 2, "each response.create bills separately"); + for audit in &audits { + assert!( + audit.is_websocket(), + "turns are recorded as WebSocket usage, metadata: {:?}", + audit.request_metadata + ); + assert_eq!(audit.model, PUBLIC_MODEL); + assert_eq!(audit.input_tokens, INPUT_TOKENS); + assert_eq!(audit.output_tokens, OUTPUT_TOKENS); + assert_eq!(audit.total_tokens, INPUT_TOKENS + OUTPUT_TOKENS); + assert_eq!(audit.status_code, Some(200)); + } + assert_ne!( + audits[0].request_id, audits[1].request_id, + "each turn gets its own logical request identity" + ); + + Ok(()) +} + +/// A client that walks away mid-turn must still be billed for what it started. +/// +/// This is the path with no protocol event to announce it: the relay loop owns +/// the turn, and losing the client is an exit the upstream never reports. +#[tokio::test] +async fn client_disconnect_mid_turn_still_settles_the_usage_row() -> Result<(), BoxError> { + let harness = Harness::start(UpstreamBehavior::StallAfterCreated).await?; + let mut client = harness.connect().await?; + + client + .send(response_create(json!({"input": "abandoned turn"}))) + .await?; + // Leave only once the turn is genuinely in flight upstream, so this covers + // an interrupted turn rather than racing turn start. + receive_event(&mut client, "response.created").await?; + drop(client); + + let audits = harness + .usage_audits_where(1, "settled turns", |audit| !is_pending(audit)) + .await?; + assert_eq!(audits.len(), 1, "the abandoned turn is still accounted for"); + let audit = &audits[0]; + assert_eq!(audit.model, PUBLIC_MODEL); + assert!( + !is_pending(audit), + "an abandoned turn must not be left pending: {audit:?}" + ); + + Ok(()) +} + +/// An upstream that dies mid-turn must surface an error and still settle. +#[tokio::test] +async fn upstream_drop_mid_turn_reports_an_error_and_settles_the_usage_row() -> Result<(), BoxError> +{ + let harness = Harness::start(UpstreamBehavior::CloseAfterCreated).await?; + let mut client = harness.connect().await?; + + client + .send(response_create(json!({"input": "doomed turn"}))) + .await?; + let error = receive_error_or_close(&mut client) + .await? + .ok_or("gateway closed without telling the client why")?; + assert_eq!(error.get("type").and_then(Value::as_str), Some("error")); + + let audits = harness + .usage_audits_where(1, "settled turns", |audit| !is_pending(audit)) + .await?; + assert_eq!(audits.len(), 1, "the failed turn is still accounted for"); + let audit = &audits[0]; + assert!( + !is_pending(audit), + "a failed turn must not be left pending: {audit:?}" + ); + + Ok(()) +} + +fn is_pending(audit: &StoredRequestUsageAudit) -> bool { + audit.status.eq_ignore_ascii_case("pending") +} + +/// A turn that finished accounting: settled, and carrying what it consumed. +fn is_billed(audit: &StoredRequestUsageAudit) -> bool { + !is_pending(audit) && audit.total_tokens > 0 +} + +// --------------------------------------------------------------------------- +// Harness +// --------------------------------------------------------------------------- + +/// A live gateway wired to a mock Responses WebSocket upstream over a throwaway +/// SQLite store. +struct Harness { + database: TemporarySqlite, + upstream: Arc, + websocket_url: String, + _upstream_server: SpawnedServer, + _gateway_server: SpawnedServer, +} + +impl Harness { + async fn start(behavior: UpstreamBehavior) -> Result { + let upstream = Arc::new(MockUpstreamState::new(behavior)); + let upstream_server = + SpawnedServer::start(mock_upstream_router(Arc::clone(&upstream))).await?; + + let database = TemporarySqlite::new(); + prepare_and_seed_database(&database.config, upstream_server.base_url()).await?; + + let data_config = GatewayDataConfig::from_database_config(database.config.clone()) + .with_encryption_key(DEVELOPMENT_ENCRYPTION_KEY); + let state = AppState::new()? + .with_data_config_and_background_isolation(data_config, false)? + // The usage runtime defaults to disabled, which silently turns every + // terminal usage write into a no-op. Without this the suite could + // not observe billing at all. Queueing stays off so the terminal + // write lands through the in-process path instead of Redis. + .with_usage_runtime_config(UsageRuntimeConfig { + enabled: true, + ..UsageRuntimeConfig::default() + })?; + let gateway_server = SpawnedServer::start(build_router_with_state(state)).await?; + let websocket_url = format!( + "{}/v1/responses", + gateway_server.base_url().replacen("http://", "ws://", 1) + ); + + Ok(Self { + database, + upstream, + websocket_url, + _upstream_server: upstream_server, + _gateway_server: gateway_server, + }) + } + + async fn connect(&self) -> Result { + let mut request = self.websocket_url.clone().into_client_request()?; + request.headers_mut().insert( + "authorization", + http::HeaderValue::from_str(&format!("Bearer {CLIENT_API_KEY}"))?, + ); + let (socket, response) = + tokio::time::timeout(RECEIVE_TIMEOUT, tokio_tungstenite::connect_async(request)) + .await + .map_err(|_| "timed out connecting to the gateway WebSocket")??; + if response.status() != http::StatusCode::SWITCHING_PROTOCOLS { + return Err( + format!("unexpected gateway handshake status: {}", response.status()).into(), + ); + } + Ok(socket) + } + + /// Waits until `expected` usage rows satisfy `settled`. + /// + /// A row is created `Pending` at turn start and reaches its final shape + /// through several independent writes, so "no longer pending" does not imply + /// "finished": a row can briefly read as completed with zero tokens and no + /// WebSocket metadata before the terminal write lands. Each caller waits for + /// the specific end state it is about to assert. + async fn usage_audits_where( + &self, + expected: usize, + what: &str, + settled: impl Fn(&StoredRequestUsageAudit) -> bool, + ) -> Result, BoxError> { + let deadline = tokio::time::Instant::now() + SETTLE_TIMEOUT; + loop { + let audits = self.usage_audits().await?; + if audits.iter().filter(|audit| settled(audit)).count() >= expected { + return Ok(audits); + } + if tokio::time::Instant::now() >= deadline { + let observed = audits + .iter() + .map(|audit| { + format!( + "{} status={} code={:?} tokens={} websocket={}", + audit.request_id, + audit.status, + audit.status_code, + audit.total_tokens, + audit.is_websocket() + ) + }) + .collect::>(); + return Err(format!( + "timed out waiting for {expected} {what}; observed {}: {observed:?}", + audits.len() + ) + .into()); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + } + + /// Reads the persisted audit rows, oldest first. + /// + /// Opens its own handle per call rather than holding one for the lifetime of + /// the harness: the gateway keeps its own pool on the same SQLite file for + /// the whole test, and an idle second pool only adds contention. + async fn usage_audits(&self) -> Result, BoxError> { + let backends = DataBackends::from_config(DataLayerConfig::from_database( + self.database.config.clone(), + ))?; + let audits = backends + .read() + .usage() + .ok_or("usage reader unavailable")? + .list_usage_audits(&UsageAuditListQuery { + limit: Some(50), + newest_first: false, + ..UsageAuditListQuery::default() + }) + .await?; + drop(backends); + Ok(audits) + } +} + +// --------------------------------------------------------------------------- +// Client protocol helpers +// --------------------------------------------------------------------------- + +/// Builds a `response.create` frame for the seeded public model. +fn response_create(fields: Value) -> Message { + let mut event = json!({"type": "response.create", "model": PUBLIC_MODEL}); + let object = event + .as_object_mut() + .expect("the literal above is an object"); + for (key, value) in fields + .as_object() + .expect("response.create fields must be an object") + { + object.insert(key.clone(), value.clone()); + } + Message::Text(event.to_string().into()) +} + +/// Reads frames until `expected_type` arrives, failing fast on a gateway error. +async fn receive_event( + socket: &mut WebSocketStream, + expected_type: &str, +) -> Result +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + tokio::time::timeout(RECEIVE_TIMEOUT, async { + loop { + let message = socket + .next() + .await + .ok_or("gateway WebSocket closed before the expected event")??; + match message { + Message::Text(text) => { + let event: Value = serde_json::from_str(text.as_ref())?; + match event.get("type").and_then(Value::as_str) { + Some("error") => { + return Err(format!("gateway returned an error event: {event}").into()) + } + Some(event_type) if event_type == expected_type => return Ok(event), + _ => {} + } + } + Message::Ping(payload) => socket.send(Message::Pong(payload)).await?, + Message::Close(frame) => { + return Err(format!("gateway closed before {expected_type}: {frame:?}").into()) + } + _ => {} + } + } + }) + .await + .map_err(|_| format!("timed out waiting for {expected_type}"))? +} + +/// Drains the socket until the gateway reports an error or hangs up. +/// +/// Returns the error event when one arrives, `None` when the gateway closed +/// without explaining itself. +async fn receive_error_or_close( + socket: &mut WebSocketStream, +) -> Result, BoxError> +where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, +{ + tokio::time::timeout(RECEIVE_TIMEOUT, async { + loop { + let Some(message) = socket.next().await else { + return Ok(None); + }; + match message? { + Message::Text(text) => { + let event: Value = serde_json::from_str(text.as_ref())?; + if event.get("type").and_then(Value::as_str) == Some("error") { + return Ok(Some(event)); + } + } + Message::Ping(payload) => socket.send(Message::Pong(payload)).await?, + Message::Close(_) => return Ok(None), + _ => {} + } + } + }) + .await + .map_err(|_| "timed out waiting for a gateway error or close")? +} + +// --------------------------------------------------------------------------- +// Mock upstream +// --------------------------------------------------------------------------- + +/// How the mock upstream answers a `response.create`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum UpstreamBehavior { + /// Announce, stream one delta, and complete — the ordinary turn. + CompleteEveryTurn, + /// Announce the response and then go quiet, leaving the turn in flight. + StallAfterCreated, + /// Announce the response and then hang up mid-turn. + CloseAfterCreated, +} + +#[derive(Debug)] +struct MockUpstreamState { + behavior: UpstreamBehavior, + connections: AtomicUsize, + events: Mutex>, + authorization_headers: Mutex>>, +} + +impl MockUpstreamState { + fn new(behavior: UpstreamBehavior) -> Self { + Self { + behavior, + connections: AtomicUsize::new(0), + events: Mutex::new(Vec::new()), + authorization_headers: Mutex::new(Vec::new()), + } + } + + fn connections(&self) -> usize { + self.connections.load(Ordering::Acquire) + } + + async fn observed_events(&self) -> Vec { + self.events.lock().await.clone() + } + + async fn authorization_headers(&self) -> Vec> { + self.authorization_headers.lock().await.clone() + } +} + +fn mock_upstream_router(state: Arc) -> Router { + Router::new() + .route("/v1/responses", get(mock_responses_websocket)) + .with_state(state) +} + +async fn mock_responses_websocket( + State(state): State>, + headers: HeaderMap, + ws: WebSocketUpgrade, +) -> Response { + let authorization = headers + .get("authorization") + .and_then(|value| value.to_str().ok()) + .map(str::to_string); + ws.on_upgrade(move |socket| run_mock_upstream(socket, state, authorization)) +} + +async fn run_mock_upstream( + mut socket: WebSocket, + state: Arc, + authorization: Option, +) { + state.connections.fetch_add(1, Ordering::AcqRel); + state.authorization_headers.lock().await.push(authorization); + while let Some(message) = socket.recv().await { + let Ok(message) = message else { + break; + }; + match message { + AxumWsMessage::Text(text) => { + let Ok(event) = serde_json::from_str::(text.as_str()) else { + break; + }; + if event.get("type").and_then(Value::as_str) != Some("response.create") { + continue; + } + let turn = { + let mut events = state.events.lock().await; + events.push(event); + events.len() + }; + let response_id = format!("resp-e2e-{turn}"); + match state.behavior { + UpstreamBehavior::CompleteEveryTurn => { + if send_mock_turn(&mut socket, &response_id).await.is_err() { + break; + } + } + UpstreamBehavior::StallAfterCreated => { + if send_mock_created(&mut socket, &response_id).await.is_err() { + break; + } + } + UpstreamBehavior::CloseAfterCreated => { + let _ = send_mock_created(&mut socket, &response_id).await; + break; + } + } + } + AxumWsMessage::Ping(payload) => { + if socket.send(AxumWsMessage::Pong(payload)).await.is_err() { + break; + } + } + AxumWsMessage::Close(_) => break, + _ => {} + } + } +} + +async fn send_mock_created(socket: &mut WebSocket, response_id: &str) -> Result<(), axum::Error> { + send_mock_event( + socket, + json!({ + "type": "response.created", + "response": { + "id": response_id, + "status": "in_progress", + "model": UPSTREAM_MODEL + } + }), + ) + .await +} + +async fn send_mock_turn(socket: &mut WebSocket, response_id: &str) -> Result<(), axum::Error> { + send_mock_created(socket, response_id).await?; + send_mock_event( + socket, + json!({ + "type": "response.output_text.delta", + "response_id": response_id, + "delta": "hello" + }), + ) + .await?; + send_mock_event( + socket, + json!({ + "type": "response.completed", + "response": { + "id": response_id, + "status": "completed", + "model": UPSTREAM_MODEL, + "output": [], + "usage": { + "input_tokens": INPUT_TOKENS, + "output_tokens": OUTPUT_TOKENS, + "total_tokens": INPUT_TOKENS + OUTPUT_TOKENS + } + } + }), + ) + .await +} + +async fn send_mock_event(socket: &mut WebSocket, event: Value) -> Result<(), axum::Error> { + socket + .send(AxumWsMessage::Text(event.to_string().into())) + .await +} + +// --------------------------------------------------------------------------- +// Seeded data store +// --------------------------------------------------------------------------- + +struct TemporarySqlite { + directory: PathBuf, + config: SqlDatabaseConfig, +} + +impl TemporarySqlite { + fn new() -> Self { + let directory = std::env::temp_dir().join(format!( + "aether-responses-ws-e2e-{}-{}", + std::process::id(), + uuid::Uuid::new_v4() + )); + let database_path = directory.join("aether.db"); + Self { + directory, + config: SqlDatabaseConfig { + driver: DatabaseDriver::Sqlite, + url: format!("sqlite://{}", database_path.display()), + pool: SqlPoolConfig { + min_connections: 1, + max_connections: 4, + acquire_timeout_ms: 5_000, + idle_timeout_ms: 30_000, + max_lifetime_ms: 300_000, + statement_cache_capacity: 64, + require_ssl: false, + }, + }, + } + } +} + +impl Drop for TemporarySqlite { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.directory); + } +} + +async fn prepare_and_seed_database( + database: &SqlDatabaseConfig, + upstream_base_url: &str, +) -> Result<(), BoxError> { + let backends = DataBackends::from_config(DataLayerConfig::from_database(database.clone()))?; + let pending = backends + .prepare_database_for_startup() + .await? + .unwrap_or_default(); + if !pending.is_empty() { + backends.run_database_migrations().await?; + } + + seed_provider_catalog(&backends, upstream_base_url).await?; + seed_models(&backends).await?; + let user_id = seed_user(&backends).await?; + seed_client_api_key(&backends, &user_id).await?; + + let candidates = backends + .read() + .minimal_candidate_selection() + .ok_or("candidate selection reader unavailable")? + .list_for_exact_api_format_and_requested_model("openai:responses", PUBLIC_MODEL) + .await?; + if !candidates.iter().any(|candidate| { + candidate.provider_id == PROVIDER_ID + && candidate.endpoint_id == ENDPOINT_ID + && candidate.key_id == PROVIDER_KEY_ID + }) { + return Err("seeded Responses WebSocket candidate is not visible".into()); + } + drop(backends); + Ok(()) +} + +async fn seed_provider_catalog( + backends: &DataBackends, + upstream_base_url: &str, +) -> Result<(), BoxError> { + let writer = backends + .write() + .provider_catalog() + .ok_or("provider catalog writer unavailable")?; + writer + .create_provider( + &StoredProviderCatalogProvider::new( + PROVIDER_ID.to_string(), + "Responses WebSocket E2E".to_string(), + None, + "openai".to_string(), + )? + .with_transport_fields( + true, + false, + false, + None, + Some(0), + None, + Some(30.0), + Some(10.0), + Some(json!({"responses_websocket": {"enabled": true}})), + ), + None, + ) + .await?; + writer + .create_endpoint( + &StoredProviderCatalogEndpoint::new( + ENDPOINT_ID.to_string(), + PROVIDER_ID.to_string(), + "openai:responses".to_string(), + Some("openai".to_string()), + Some("responses".to_string()), + true, + )? + .with_transport_fields( + upstream_base_url.trim_end_matches('/').to_string(), + None, + None, + Some(0), + Some("/v1/responses".to_string()), + None, + None, + None, + )?, + ) + .await?; + writer + .create_key( + &StoredProviderCatalogKey::new( + PROVIDER_KEY_ID.to_string(), + PROVIDER_ID.to_string(), + "Responses WebSocket E2E".to_string(), + "api_key".to_string(), + Some(json!({"streaming": true})), + true, + )? + .with_transport_fields( + Some(json!(["openai:responses"])), + encrypt_python_fernet_plaintext(DEVELOPMENT_ENCRYPTION_KEY, PROVIDER_API_KEY)?, + None, + None, + Some(json!({"openai:responses": 1})), + Some(json!([PUBLIC_MODEL, UPSTREAM_MODEL])), + None, + None, + None, + )? + .with_health_fields( + Some(json!({"openai:responses": {"status": "healthy"}})), + Some(json!({"openai:responses": {"state": "closed"}})), + ), + ) + .await?; + Ok(()) +} + +async fn seed_models(backends: &DataBackends) -> Result<(), BoxError> { + let writer = backends + .write() + .global_models() + .ok_or("global model writer unavailable")?; + writer + .create_admin_global_model(&CreateAdminGlobalModelRecord::new( + GLOBAL_MODEL_ID.to_string(), + PUBLIC_MODEL.to_string(), + PUBLIC_MODEL.to_string(), + true, + Some(0.0), + None, + Some(json!({"streaming": true, "chat": true})), + Some(json!({"model_mappings": [UPSTREAM_MODEL]})), + )?) + .await?; + writer + .create_admin_provider_model(&UpsertAdminProviderModelRecord::new( + PROVIDER_MODEL_ID.to_string(), + PROVIDER_ID.to_string(), + GLOBAL_MODEL_ID.to_string(), + UPSTREAM_MODEL.to_string(), + Some(json!([{ + "name": UPSTREAM_MODEL, + "priority": 0, + "api_formats": ["openai:responses"], + "endpoint_ids": [ENDPOINT_ID] + }])), + Some(0.0), + None, + Some(false), + Some(false), + Some(true), + Some(false), + Some(false), + true, + true, + Some(json!({"responses_websocket_e2e": true})), + )?) + .await?; + Ok(()) +} + +async fn seed_user(backends: &DataBackends) -> Result { + let users = backends.read().users().ok_or("user reader unavailable")?; + let user = users + .create_local_auth_user_with_settings( + Some("responses-ws-e2e@example.test".to_string()), + true, + "responses-ws-e2e".to_string(), + "disabled-password".to_string(), + "user".to_string(), + Some(vec![PROVIDER_ID.to_string()]), + Some(vec!["openai:responses".to_string()]), + Some(vec![PUBLIC_MODEL.to_string()]), + None, + ) + .await? + .ok_or("failed to create E2E user")?; + let wallets = backends + .read() + .wallets() + .ok_or("wallet reader unavailable")?; + wallets + .initialize_auth_user_wallet(&user.id, 0.0, true) + .await?; + Ok(user.id) +} + +async fn seed_client_api_key(backends: &DataBackends, user_id: &str) -> Result<(), BoxError> { + backends + .write() + .auth_api_keys() + .ok_or("auth API key writer unavailable")? + .create_standalone_api_key(CreateStandaloneApiKeyRecord { + user_id: user_id.to_string(), + api_key_id: API_KEY_ID.to_string(), + key_hash: sha256_hex(CLIENT_API_KEY), + key_encrypted: Some(CLIENT_API_KEY.to_string()), + name: Some("Responses WebSocket E2E".to_string()), + allowed_providers: Some(vec![PROVIDER_ID.to_string()]), + allowed_api_formats: Some(vec!["openai:responses".to_string()]), + allowed_models: Some(vec![PUBLIC_MODEL.to_string()]), + ip_rules: None, + rate_limit: Some(0), + concurrent_limit: None, + force_capabilities: None, + is_active: true, + expires_at_unix_secs: None, + auto_delete_on_expiry: false, + total_requests: 0, + total_tokens: 0, + total_cost_usd: 0.0, + }) + .await?; + backends + .read() + .wallets() + .ok_or("wallet reader unavailable")? + .initialize_auth_api_key_wallet(API_KEY_ID, 0.0, true) + .await?; + if backends + .read() + .wallets() + .ok_or("wallet reader unavailable")? + .find(WalletLookupKey::ApiKeyId(API_KEY_ID)) + .await? + .is_none() + { + return Err("failed to initialize E2E API key wallet".into()); + } + Ok(()) +} + +fn sha256_hex(value: &str) -> String { + let mut hasher = sha2::Sha256::new(); + hasher.update(value.as_bytes()); + format!("{:x}", hasher.finalize()) +}