diff --git a/README.md b/README.md index 9e558e4..52f38fe 100644 --- a/README.md +++ b/README.md @@ -299,6 +299,8 @@ model_accounts = { "gpt-5.5" = "personal", "gpt-5.6-sol" = "work" } `model_accounts` matches the exact requested model ID. `models_account` selects the account for `GET /models` and `HEAD /models`, including query parameters, independently of model-specific pins. Both settings are optional; pinned accounts must exist and belong to the pool. These are strict assignments: an unavailable pinned account returns an error instead of falling back to another account. A conflicting account-owned continuation or file also returns an error. Unpinned requests keep the normal routing behavior. Removing an account clears its pins. +For Daybreak, newer Codex clients send a base model with `access_programs.cyber` set to `"daybreak_blue"` or `"daybreak_red"`. These requests prefer the `gpt-daybreak-blue-latest` or `gpt-daybreak-red-latest` pin respectively, falling back to the base model's pin when the Daybreak pin is unset. Existing Daybreak pins therefore cover both request formats; standard requests keep their model-specific or normal pool routing. The request body is forwarded unchanged. + Model-specific pins require the default `responses_websocket_mode = "http_bridge"` or `"direct"`. Configuration with model-specific pins and `"raw"` mode is rejected because raw WebSocket traffic cannot be inspected for model IDs. A model listing pin alone is compatible with every WebSocket mode. The equivalent manual configuration is: diff --git a/src/proxy/account_pin_tests.rs b/src/proxy/account_pin_tests.rs index a367b69..fac4560 100644 --- a/src/proxy/account_pin_tests.rs +++ b/src/proxy/account_pin_tests.rs @@ -23,6 +23,23 @@ impl Drop for PinFixture { } async fn fixture(dir: &Path, failure: Option) -> PinFixture { + fixture_with_pins( + dir, + failure, + BTreeMap::from([ + ("pinned-model".into(), "b".into()), + ("gpt-daybreak-blue-latest".into(), "b".into()), + ("gpt-daybreak-red-latest".into(), "b".into()), + ]), + ) + .await +} + +async fn fixture_with_pins( + dir: &Path, + failure: Option, + model_accounts: BTreeMap, +) -> PinFixture { let upstream = TcpListener::bind("127.0.0.1:0").await.unwrap(); let upstream_address = upstream.local_addr().unwrap(); let seen = Arc::new(StdMutex::new(Vec::new())); @@ -117,7 +134,7 @@ async fn fixture(dir: &Path, failure: Option) -> PinFixture { PoolConfig { members: vec!["a".into(), "b".into()], preferred: Some("a".into()), - model_accounts: BTreeMap::from([("pinned-model".into(), "b".into())]), + model_accounts, models_account: Some("b".into()), ..Default::default() }, @@ -224,6 +241,72 @@ async fn http_model_pin_overrides_preferred_and_soft_affinity_without_changing_o ); } +#[tokio::test] +async fn http_access_program_pin_overrides_soft_affinity_and_keeps_standard_routing() { + let dir = tempfile::tempdir().unwrap(); + let fixture = fixture(dir.path(), None).await; + let key = fixture.app.router.affinity.key("thread:soft-thread"); + assert!(fixture.app.router.bind(key, "a").await); + for program in ["standard", "daybreak_blue", "daybreak_red"] { + assert_eq!( + request( + &fixture, + Method::POST, + "/v1/responses", + &format!( + r#"{{"model":"gpt-6-sol","access_programs":{{"cyber":"{program}"}},"input":[]}}"# + ), + None, + ) + .await, + StatusCode::OK + ); + } + let seen = fixture.seen.lock().unwrap(); + assert_eq!( + seen.iter() + .map(|(auth, _)| auth.as_str()) + .collect::>(), + ["Bearer token-a", "Bearer token-b", "Bearer token-b"] + ); +} + +#[tokio::test] +async fn access_program_pin_takes_precedence_but_an_unpinned_program_keeps_the_base_model_pin() { + let dir = tempfile::tempdir().unwrap(); + let fixture = fixture_with_pins( + dir.path(), + None, + BTreeMap::from([ + ("gpt-6-sol".into(), "b".into()), + ("gpt-daybreak-blue-latest".into(), "a".into()), + ]), + ) + .await; + for (program, account) in [("daybreak_blue", "a"), ("daybreak_red", "b")] { + let body = format!( + r#"{{"model":"gpt-6-sol","access_programs":{{"cyber":"{program}"}},"input":[]}}"# + ); + assert_eq!( + request(&fixture, Method::POST, "/v1/responses", &body, None).await, + StatusCode::OK + ); + assert_eq!( + fixture.seen.lock().unwrap().last().unwrap().0, + format!("Bearer token-{account}") + ); + let replay = + ReplayBody::from_bytes(Bytes::from(body), 4096, 4096, fixture.app.stats.clone()) + .unwrap(); + let route = fixture + .app + .route_websocket_frame(&fixture.listener, &hyper::HeaderMap::new(), &replay, None) + .await + .unwrap(); + assert_eq!(route.account_id, account); + } +} + #[tokio::test] async fn listing_pin_applies_to_get_and_head_with_query_parameters() { let dir = tempfile::tempdir().unwrap(); @@ -255,23 +338,30 @@ async fn http_pin_rejects_conflicting_owner_and_unavailable_account_before_dispa let fixture = fixture(dir.path(), None).await; let key = fixture.app.router.affinity.key("turn-state:owned-by-a"); assert!(fixture.app.router.bind(key, "a").await); - let body = r#"{"model":"pinned-model","input":[]}"#; - assert_eq!( - request( - &fixture, - Method::POST, - "/v1/responses", - body, - Some("owned-by-a") - ) - .await, - StatusCode::CONFLICT - ); + let bodies = [ + r#"{"model":"pinned-model","input":[]}"#, + r#"{"model":"gpt-6-sol","access_programs":{"cyber":"daybreak_blue"},"input":[]}"#, + ]; + for body in bodies { + assert_eq!( + request( + &fixture, + Method::POST, + "/v1/responses", + body, + Some("owned-by-a") + ) + .await, + StatusCode::CONFLICT + ); + } fixture.app.router.auth_failure("b").await; - assert_eq!( - request(&fixture, Method::POST, "/v1/responses", body, None).await, - StatusCode::SERVICE_UNAVAILABLE - ); + for body in bodies { + assert_eq!( + request(&fixture, Method::POST, "/v1/responses", body, None).await, + StatusCode::SERVICE_UNAVAILABLE + ); + } assert_eq!( request(&fixture, Method::GET, "/v1/models", "", None).await, StatusCode::SERVICE_UNAVAILABLE @@ -285,27 +375,29 @@ async fn http_pin_does_not_fall_back_after_quota_or_capacity_failure() { StatusCode::TOO_MANY_REQUESTS, StatusCode::SERVICE_UNAVAILABLE, ] { - let dir = tempfile::tempdir().unwrap(); - let fixture = fixture(dir.path(), Some(status)).await; - assert_eq!( - request( - &fixture, - Method::POST, - "/v1/responses", - r#"{"model":"pinned-model","input":[]}"#, - None - ) - .await, - status - ); - let seen = fixture.seen.lock().unwrap(); - assert_eq!(seen.len(), 1, "pin must suppress alternate-account retry"); - assert_eq!(seen[0].0, "Bearer token-b"); + for body in [ + r#"{"model":"pinned-model","input":[]}"#, + r#"{"model":"gpt-6-sol","access_programs":{"cyber":"daybreak_blue"},"input":[]}"#, + ] { + let dir = tempfile::tempdir().unwrap(); + let fixture = fixture(dir.path(), Some(status)).await; + assert_eq!( + request(&fixture, Method::POST, "/v1/responses", body, None).await, + status + ); + let seen = fixture.seen.lock().unwrap(); + assert_eq!(seen.len(), 1, "pin must suppress alternate-account retry"); + assert_eq!(seen[0].0, "Bearer token-b"); + } } } #[tokio::test] async fn direct_frame_pin_overrides_socket_and_soft_preference_and_is_a_hard_owner() { + for body in [ + br#"{"type":"response.create","model":"pinned-model","input":[]}"#.as_slice(), + br#"{"type":"response.create","model":"gpt-6-sol","access_programs":{"cyber":"daybreak_blue"},"input":[]}"#.as_slice(), + ] { let dir = tempfile::tempdir().unwrap(); let fixture = fixture(dir.path(), None).await; let key = fixture.app.router.affinity.key("thread:soft-thread"); @@ -313,7 +405,7 @@ async fn direct_frame_pin_overrides_socket_and_soft_preference_and_is_a_hard_own let mut headers = hyper::HeaderMap::new(); headers.insert("thread-id", "soft-thread".parse().unwrap()); let replay = ReplayBody::from_bytes( - Bytes::from_static(br#"{"type":"response.create","model":"pinned-model","input":[]}"#), + Bytes::copy_from_slice(body), fixture.app.config.proxy.max_request_bytes, fixture.app.config.proxy.max_spool_bytes, fixture.app.stats.clone(), @@ -348,51 +440,66 @@ async fn direct_frame_pin_overrides_socket_and_soft_preference_and_is_a_hard_own .err() .expect("unavailable pinned account must fail"); assert!(error.to_string().contains("unavailable"), "{error:#}"); + } } #[tokio::test] async fn http_bridge_websocket_honors_model_pin() { use tokio_tungstenite::tungstenite::client::IntoClientRequest; - let dir = tempfile::tempdir().unwrap(); - let fixture = fixture(dir.path(), None).await; - let stream = tokio::net::TcpStream::connect(fixture.address) - .await - .unwrap(); - let request = format!( - "ws://{}/{SECRET}/backend-api/codex/responses", - fixture.address - ) - .into_client_request() - .unwrap(); - let (mut websocket, _) = tokio_tungstenite::client_async(request, stream) - .await - .unwrap(); - websocket - .send(Message::Text( - r#"{"type":"response.create","model":"pinned-model","input":[]}"#.into(), - )) - .await + for (body, account) in [ + ( + r#"{"type":"response.create","model":"pinned-model","input":[]}"#, + "b", + ), + ( + r#"{"type":"response.create","model":"gpt-6-sol","access_programs":{"cyber":"daybreak_blue"},"input":[]}"#, + "b", + ), + ( + r#"{"type":"response.create","model":"gpt-6-sol","access_programs":{"cyber":"daybreak_red"},"input":[]}"#, + "b", + ), + ( + r#"{"type":"response.create","model":"gpt-6-sol","access_programs":{"cyber":"standard"},"input":[]}"#, + "a", + ), + ] { + let dir = tempfile::tempdir().unwrap(); + let fixture = fixture(dir.path(), None).await; + let stream = tokio::net::TcpStream::connect(fixture.address) + .await + .unwrap(); + let request = format!( + "ws://{}/{SECRET}/backend-api/codex/responses", + fixture.address + ) + .into_client_request() .unwrap(); - tokio::time::timeout(Duration::from_secs(5), async { - loop { - let message = websocket.next().await.unwrap().unwrap(); - let Message::Text(text) = message else { - continue; - }; - let event: serde_json::Value = serde_json::from_str(text.as_str()).unwrap(); - assert_ne!(event["type"], "error", "{event}"); - assert_ne!(event["type"], "response.failed", "{event}"); - if event["type"] == "response.completed" { - break; + let (mut websocket, _) = tokio_tungstenite::client_async(request, stream) + .await + .unwrap(); + websocket.send(Message::Text(body.into())).await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let message = websocket.next().await.unwrap().unwrap(); + let Message::Text(text) = message else { + continue; + }; + let event: serde_json::Value = serde_json::from_str(text.as_str()).unwrap(); + assert_ne!(event["type"], "error", "{event}"); + assert_ne!(event["type"], "response.failed", "{event}"); + if event["type"] == "response.completed" { + break; + } } - } - }) - .await - .expect("bridge did not complete pinned response"); - let seen = fixture.seen.lock().unwrap(); - assert_eq!(seen.len(), 1); - assert_eq!(seen[0].0, "Bearer token-b"); + }) + .await + .expect("bridge did not complete pinned response"); + let seen = fixture.seen.lock().unwrap(); + assert_eq!(seen.len(), 1); + assert_eq!(seen[0].0, format!("Bearer token-{account}")); + } } #[tokio::test] diff --git a/src/proxy/mod.rs b/src/proxy/mod.rs index 7e293c8..f1311ee 100644 --- a/src/proxy/mod.rs +++ b/src/proxy/mod.rs @@ -1755,12 +1755,8 @@ impl App { .excluded_account .clone() }); - let pinned_account = request_account_pin( - pool, - &method, - &path, - if inspect_model { replay.model() } else { None }, - ); + let pinned_account = + request_account_pin(pool, &method, &path, inspect_model.then_some(&replay)); if let Some(account) = pinned_account { if bound_account .as_deref() @@ -2998,17 +2994,14 @@ impl App { if missing_hard_owner { anyhow::bail!("request carries hard continuity state with no known account owner") } - if let Some(account) = replay - .model() - .and_then(|model| pool.model_accounts.get(model)) - { + if let Some(account) = model_account_pin(pool, replay) { if hard_bound_account .as_ref() .is_some_and(|owner| owner != account) { anyhow::bail!("configured account pin conflicts with request continuity owner") } - hard_bound_account = Some(account.clone()); + hard_bound_account = Some(account.to_owned()); hard_owner = true; non_previous_hard_owner = true; } @@ -4903,15 +4896,27 @@ fn request_account_pin<'a>( pool: &'a PoolConfig, method: &Method, path: &str, - model: Option<&str>, + replay: Option<&ReplayBody>, ) -> Option<&'a str> { if matches!(*method, Method::GET | Method::HEAD) && path.split('?').next() == Some("/models") { pool.models_account.as_deref() } else { - model.and_then(|model| pool.model_accounts.get(model).map(String::as_str)) + replay.and_then(|replay| model_account_pin(pool, replay)) } } +fn model_account_pin<'a>(pool: &'a PoolConfig, replay: &ReplayBody) -> Option<&'a str> { + replay + .routing_model() + .and_then(|model| pool.model_accounts.get(model)) + .or_else(|| { + replay + .model() + .and_then(|model| pool.model_accounts.get(model)) + }) + .map(String::as_str) +} + fn is_legacy_compact_path(path: &str) -> bool { path.split('?').next() == Some("/responses/compact") } diff --git a/src/proxy/replay_body.rs b/src/proxy/replay_body.rs index 9afd69d..f666f71 100644 --- a/src/proxy/replay_body.rs +++ b/src/proxy/replay_body.rs @@ -60,6 +60,7 @@ pub struct ReplayBody { len: usize, stats: Arc, model: Option, + cyber_access_program: Option, model_scan_complete: bool, thread_id: Option, context_session_id: Option, @@ -74,6 +75,8 @@ pub struct ReplayBody { #[derive(Clone, Copy, PartialEq, Eq)] enum KeyKind { Model, + AccessPrograms, + CyberAccessProgram, ClientMetadata, ThreadId, SessionId, @@ -98,17 +101,19 @@ struct Container { struct MetadataScanner { containers: Vec, client_depth: Option, + access_programs_depth: Option, in_string: bool, escape: bool, string_is_key: bool, capture_value: Option, token: Vec, token_overflow: bool, - model_token: Vec, - model_token_overflow: bool, + routing_token: Vec, + routing_token_overflow: bool, key: Option, pending_value: Option, model: Option, + cyber_access_program: Option, model_scan_incomplete: bool, thread_id: Option, context_session_id: Option, @@ -134,24 +139,29 @@ impl MetadataScanner { return; } if self.in_string { - // Keep a bounded JSON spelling for model names and property names, - // so escaped model names route exactly like their decoded spelling. - if ((self.string_is_key && self.containers.len() == 1) - || self.capture_value == Some(KeyKind::Model)) - && !self.model_token_overflow + // Decode both model selectors and root property names so escaped + // spellings route exactly like the upstream JSON parser. + if ((self.string_is_key + && (self.containers.len() == 1 + || self.access_programs_depth == Some(self.containers.len()))) + || matches!( + self.capture_value, + Some(KeyKind::Model | KeyKind::CyberAccessProgram) + )) + && !self.routing_token_overflow && (byte != b'"' || self.escape) { // Each ASCII character can occupy six bytes as a JSON escape. - // Only the five-character root key `model` needs decoding. + // The longest routing key is `access_programs`. let limit = if self.string_is_key { - 5 * 6 + b"access_programs".len() * 6 } else { MAX_MODEL_BYTES * 6 }; - if self.model_token.len() < limit { - self.model_token.push(byte); + if self.routing_token.len() < limit { + self.routing_token.push(byte); } else { - self.model_token_overflow = true; + self.routing_token_overflow = true; } } if self.escape { @@ -188,12 +198,13 @@ impl MetadataScanner { self.in_string = true; self.token.clear(); self.token_overflow = false; - self.model_token.clear(); - self.model_token_overflow = false; + self.routing_token.clear(); + self.routing_token_overflow = false; self.capture_value = self.pending_value.take().filter(|kind| { matches!( kind, KeyKind::Model + | KeyKind::CyberAccessProgram | KeyKind::ThreadId | KeyKind::SessionId | KeyKind::PreviousResponseId @@ -218,6 +229,7 @@ impl MetadataScanner { continue; } let is_client = self.pending_value == Some(KeyKind::ClientMetadata); + let is_access_programs = self.pending_value == Some(KeyKind::AccessPrograms); let native_input_item = self.containers.last().is_some_and(|v| v.input_array); self.pending_value = None; self.containers.push(Container { @@ -231,6 +243,9 @@ impl MetadataScanner { if is_client { self.client_depth = Some(self.containers.len()); } + if is_access_programs { + self.access_programs_depth = Some(self.containers.len()); + } } b'[' => { if self.containers.len() >= 128 { @@ -254,6 +269,9 @@ impl MetadataScanner { if self.client_depth == Some(self.containers.len()) { self.client_depth = None; } + if self.access_programs_depth == Some(self.containers.len()) { + self.access_programs_depth = None; + } if let Some(container) = self.containers.pop() && container.account_scoped_file && let Some(file_id) = container.file_id @@ -294,11 +312,16 @@ impl MetadataScanner { // Also inspect an all-turns request without session metadata so it fails closed. self.context_envelope |= !self.string_is_key && self.token == b"all_turns"; if let Some(kind) = self.capture_value { - if kind == KeyKind::Model { - self.model = self - .decoded_model_token() + if matches!(kind, KeyKind::Model | KeyKind::CyberAccessProgram) { + let value = self + .decoded_routing_token() .filter(|value| !value.is_empty() && value.len() <= MAX_MODEL_BYTES); - self.model_scan_incomplete |= self.model_token_overflow; + match kind { + KeyKind::Model => self.model = value, + KeyKind::CyberAccessProgram => self.cyber_access_program = value, + _ => unreachable!(), + } + self.model_scan_incomplete |= self.routing_token_overflow; } if !self.token_overflow && !self.token.is_empty() { let value = String::from_utf8(self.token.clone()).ok(); @@ -333,7 +356,12 @@ impl MetadataScanner { ); } } - KeyKind::Model | KeyKind::ClientMetadata | KeyKind::Other | KeyKind::Input => {} + KeyKind::Model + | KeyKind::AccessPrograms + | KeyKind::CyberAccessProgram + | KeyKind::ClientMetadata + | KeyKind::Other + | KeyKind::Input => {} } } } else if self.string_is_key { @@ -356,11 +384,20 @@ impl MetadataScanner { { self.nonportable_state = true; } - self.key = if at_root && self.decoded_model_token().as_deref() == Some("model") { + self.key = if at_root && self.decoded_routing_token().as_deref() == Some("model") { // A later non-string or oversized model must not leave an earlier // model active when the upstream parser uses the final property. self.model = None; Some(KeyKind::Model) + } else if at_root && self.decoded_routing_token().as_deref() == Some("access_programs") + { + self.cyber_access_program = None; + Some(KeyKind::AccessPrograms) + } else if self.access_programs_depth == Some(self.containers.len()) + && self.decoded_routing_token().as_deref() == Some("cyber") + { + self.cyber_access_program = None; + Some(KeyKind::CyberAccessProgram) } else if at_root && !self.token_overflow && self.token == b"client_metadata" { Some(KeyKind::ClientMetadata) } else if in_client && !self.token_overflow && self.token == b"thread_id" { @@ -388,16 +425,16 @@ impl MetadataScanner { self.string_is_key = false; } - fn decoded_model_token(&self) -> Option { - if self.model_token_overflow { + fn decoded_routing_token(&self) -> Option { + if self.routing_token_overflow { return None; } - if !self.model_token.contains(&b'\\') { - return String::from_utf8(self.model_token.clone()).ok(); + if !self.routing_token.contains(&b'\\') { + return String::from_utf8(self.routing_token.clone()).ok(); } - let mut quoted = Vec::with_capacity(self.model_token.len() + 2); + let mut quoted = Vec::with_capacity(self.routing_token.len() + 2); quoted.push(b'"'); - quoted.extend_from_slice(&self.model_token); + quoted.extend_from_slice(&self.routing_token); quoted.push(b'"'); serde_json::from_slice("ed).ok() } @@ -465,6 +502,7 @@ impl ReplayBody { storage: Storage::Memory(bytes), stats, model: metadata.model, + cyber_access_program: metadata.cyber_access_program, model_scan_complete: !metadata.disabled && !metadata.model_scan_incomplete, thread_id: metadata.thread_id, context_session_id: metadata.context_session_id, @@ -599,6 +637,7 @@ impl ReplayBody { len, stats, model: metadata.model, + cyber_access_program: metadata.cyber_access_program, model_scan_complete: !metadata.disabled && !metadata.model_scan_incomplete, thread_id: metadata.thread_id, context_session_id: metadata.context_session_id, @@ -615,6 +654,16 @@ impl ReplayBody { self.model.as_deref() } + /// Newer Codex clients request Daybreak as a treatment on a base model. + /// Keep the existing Daybreak account pins effective for both wire formats. + pub fn routing_model(&self) -> Option<&str> { + match self.cyber_access_program.as_deref() { + Some("daybreak_blue") => Some("gpt-daybreak-blue-latest"), + Some("daybreak_red") => Some("gpt-daybreak-red-latest"), + _ => self.model(), + } + } + pub fn model_scan_complete(&self) -> bool { self.model_scan_complete } @@ -748,7 +797,7 @@ mod tests { #[tokio::test] async fn zstd_metadata_and_replay_survive_chunking_and_disk_spooling() { - let payload = Bytes::from_static(br#"{"model":"gpt-5","client_metadata":{"thread_id":"thread-1","session_id":"session-1"},"previous_response_id":"resp-1","prompt_cache_key":"cache-1","input":[{"type":"input_file","file_id":"file-1"},{"type":"compaction","encrypted_content":"owned"}]}"#); + let payload = Bytes::from_static(br#"{"model":"gpt-5","access_programs":{"cyber":"daybreak_blue"},"client_metadata":{"thread_id":"thread-1","session_id":"session-1"},"previous_response_id":"resp-1","prompt_cache_key":"cache-1","input":[{"type":"input_file","file_id":"file-1"},{"type":"compaction","encrypted_content":"owned"}]}"#); let wire = zstd_bytes(&payload).await; for memory_limit in [1, 4096] { let stats = Arc::new(Stats::default()); @@ -762,6 +811,7 @@ mod tests { .await .unwrap(); assert_eq!(replay.model(), Some("gpt-5")); + assert_eq!(replay.routing_model(), Some("gpt-daybreak-blue-latest")); assert!(replay.model_scan_complete()); assert_eq!(replay.thread_id(), Some("thread-1")); assert_eq!(replay.context_session_id(), Some("session-1")); @@ -891,6 +941,73 @@ mod tests { assert!(scanner.model_scan_incomplete); } + #[test] + fn access_program_routing_uses_only_the_final_root_property_across_chunks() { + for (properties, expected) in [ + ( + r#""access_programs":{"cyber":"daybreak_blue"}"#, + "gpt-daybreak-blue-latest", + ), + ( + r#""access_programs":{"cyber":"daybreak_red"}"#, + "gpt-daybreak-red-latest", + ), + (r#""access_programs":{"cyber":"standard"}"#, "gpt-6-sol"), + ( + r#""input":[{"access_programs":{"cyber":"daybreak_blue"}}]"#, + "gpt-6-sol", + ), + ( + r#""access_programs":{"cyber":"daybreak_blue"},"access_programs":null"#, + "gpt-6-sol", + ), + ( + r#""access_programs":{"cyber":"daybreak_blue","cyber":"standard"}"#, + "gpt-6-sol", + ), + ( + r#""access_programs":{"cyber":"daybreak_blue","cyber":null}"#, + "gpt-6-sol", + ), + ( + r#""access_progr\u0061ms":{"cy\u0062er":"daybreak_\u0062lue"}"#, + "gpt-daybreak-blue-latest", + ), + ( + r#""access_programs":{"nested":{"cyber":"daybreak_blue"}}"#, + "gpt-6-sol", + ), + ( + r#""access_programs":{"cyber":"daybreak_blue"},"access_programs":{}"#, + "gpt-6-sol", + ), + ( + r#""access_programs":{"cyber":"daybreak_blue"},"access_programs":{"cyber":"daybreak_red"}"#, + "gpt-daybreak-red-latest", + ), + ] { + let payload = format!(r#"{{"model":"gpt-6-sol",{properties}}}"#); + for size in [1, 3, payload.len()] { + let mut scanner = MetadataScanner::default(); + for chunk in payload.as_bytes().chunks(size) { + scanner.feed(chunk); + } + scanner.finish(); + let replay = ReplayBody::from_bytes( + Bytes::from(payload.clone()), + 4096, + 4096, + Arc::new(Stats::default()), + ) + .unwrap(); + assert_eq!(scanner.cyber_access_program, replay.cyber_access_program); + assert_eq!(replay.model(), Some("gpt-6-sol")); + assert_eq!(replay.routing_model(), Some(expected), "{payload}"); + assert!(replay.model_scan_complete()); + } + } + } + #[test] fn model_scan_reports_limits_and_incomplete_requests() { let nested = format!("{}null{}", "[".repeat(129), "]".repeat(129)); @@ -900,6 +1017,11 @@ mod tests { format!(r#"{{"model":"{}"}}"#, "x".repeat(MAX_MODEL_BYTES * 6 + 1)), r#"{"model":"unfinished"#.to_owned(), r#"{"model":"gpt-5""#.to_owned(), + format!( + r#"{{"access_programs":{{"cyber":"{}"}}}}"#, + "x".repeat(MAX_MODEL_BYTES * 6 + 1) + ), + r#"{"model":"gpt-6-sol","access_programs":{"cyber":"daybreak_blue"#.to_owned(), ] { let replay = ReplayBody::from_bytes( Bytes::from(payload),