diff --git a/.rust-file-sizes.json b/.rust-file-sizes.json index addd59ac..10c9467d 100644 --- a/.rust-file-sizes.json +++ b/.rust-file-sizes.json @@ -14,7 +14,7 @@ "crates/agentic-server-core/src/types/io/output.rs": 1110, "crates/agentic-server-core/src/types/request_response.rs": 582, "crates/agentic-server-core/src/types/tools/params.rs": 537, - "crates/agentic-server/src/agentic_process.rs": 558, + "crates/agentic-server/src/agentic_process.rs": 557, "crates/agentic-server/src/auth.rs": 799, "crates/agentic-server/src/handler/websocket/responses.rs": 864 }, diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index ea3bb07b..43787759 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -116,12 +116,12 @@ continuous SSE lifecycle. The crate produces a library plus two independent binaries. `src/lib.rs` exports `agentic_cli`, `agentic_harness`, `agentic_output`, `agentic_process`, `app`, `auth`, -and `handler`. On top of that: +`handler`, and `model_capabilities`. On top of that: | Binary | Entry point | Uses | |---|---|---| -| `agentic-server` (the gateway) | `src/main.rs` | `app`, `auth`, `handler`, plus binary-private `server.rs` and `config_file.rs` | -| `agentic` (the CLI launcher) | `src/bin/agentic.rs` | `agentic_cli`, `agentic_harness`, `agentic_output`, `agentic_process` only | +| `agentic-server` (the gateway) | `src/main.rs` | `app`, `auth`, `handler`, `model_capabilities`, plus binary-private `server.rs` and `config_file.rs` | +| `agentic` (the CLI launcher) | `src/bin/agentic.rs` | `agentic_cli`, `agentic_harness`, `agentic_output`, `agentic_process`, `model_capabilities` | These are two unrelated concerns bundled in one crate. If you're working on request handling, ignore `agentic_cli*`/`agentic_harness.rs`/`agentic_output.rs`/ @@ -129,6 +129,12 @@ handling, ignore `agentic_cli*`/`agentic_harness.rs`/`agentic_output.rs`/ a coding harness (Codex or Claude Code) as subprocesses for local, single-command use (`agentic serve `), and never touch the request path. +`model_capabilities.rs` is the one deliberate exception: it is the shared contract both +sides speak. The gateway resolves each model's `InputModalities` there and serves them in +the Codex catalog; the launcher parses that same catalog back through +`CodexCatalogCapabilities` before writing an isolated Codex home. Keeping one definition is +what stops the HTTP catalog and a launcher catalog from disagreeing about image support. + ### `app.rs`, `server.rs`, `main.rs` - **`app.rs`** (library) builds the router: `AppState` (the per-request-shared state: diff --git a/CHANGELOG.md b/CHANGELOG.md index afd92930..53ba7b6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ All notable changes to Agentic API are documented here. ### Added +- Added typed per-model input-modality overrides to `config.toml` + (`[models.""] input_modalities = ["text", "image"]`), validated at startup: + unknown modality names, empty lists, duplicates, and image-only lists are rejected with the + offending file and line (#252). - Added Brave Search as a selectable backend for the gateway-owned `web_search` tool (#294, Phase 2 of #291). Select it with `AGENTIC_WEB_SEARCH_PROVIDER=brave` or `[web_search] provider = "brave"` and supply `BRAVE_API_KEY`; the endpoint defaults to `https://api.search.brave.com` and can be overridden with `AGENTIC_WEB_SEARCH_BASE_URL` @@ -23,6 +27,15 @@ All notable changes to Agentic API are documented here. ### Changed +- Modeled the Codex model catalog and the upstream model listing as typed Rust structs instead of + untyped JSON, and reported an undecodable upstream `/v1/models` payload as `502` rather than + serving it as an empty catalog (#252). +- `agentic run codex` and `agentic harness codex` now resolve the model and its input modalities + from a single gateway catalog snapshot before writing an isolated Codex home, retrying a warming + gateway and failing with an actionable error when the catalog cannot be fetched or does not list + the selected model. A gateway behind OIDC now requires `--api-key` for `agentic harness codex`. + `agentic_harness::prepare_codex_home` requires the resolved modalities and is no longer public + (#252). - `WebSearchProviderConfig` is now `#[non_exhaustive]` and gains `provider` and `max_concurrent_queries` fields; construct it with `WebSearchProviderConfig::new(api_key, base_url)` plus the `with_provider` and `with_max_concurrent_queries` builders. Downstream crates that built it with a struct literal must switch to the @@ -33,6 +46,19 @@ All notable changes to Agentic API are documented here. `config.toml` now records `provider = "you"` and leaves `api_key_env` unset so provider switches select the matching default credential variable. +### Fixed + +- Resolved Codex image capabilities consistently: the HTTP model catalog and both launcher modes + now advertise the same resolved `input_modalities`, so a vision-capable model no longer has image + content stripped client-side because an isolated catalog hardcoded `["text"]`. Existing persistent + Codex session homes must be regenerated to pick this up (#252). + +### Testing + +- Extended the pinned Codex 0.149.1 smoke with actual PNG attachments through both launcher modes, exact upstream + image-byte assertions, and a text-only negative control. The smoke replays the committed vision recording without + live API credentials (#261). + ## [0.7.0] - 2026-09-14 ### Added diff --git a/README.md b/README.md index eaa626d1..5567f928 100644 --- a/README.md +++ b/README.md @@ -248,12 +248,28 @@ max_request_body_size_bytes = 10485760 # Must be greater than zero. max_concurrent_gateway_calls = 5 +[models."Qwen/Qwen3-VL-8B-Instruct"] +# Input modalities to advertise for this served model ID. +# Accepted values are "text" and "text" + "image"; text is always required. +input_modalities = ["text", "image"] + [mcp_servers.counter] url = "https://mcp.example.com/mcp" allowed_tools = ["tool_1_name", "tool_2_name"] require_approval = "never" ``` +`[models.""]` declares capabilities the gateway cannot infer. `input_modalities` resolves in this +order: an explicit override here, then `capabilities: ["image"]` from the upstream `/v1/models` entry, then a +conservative text-only fallback. Capabilities are never guessed from a model name, so a vision model served without +capability metadata needs this override. An explicit `["text"]` wins over upstream image metadata, which pins a model +to text. Unknown modality names, empty lists, duplicates, and image-only lists are rejected at startup with the +offending file and line. + +This matters because Codex reads its local model catalog and strips image content from a request before sending it +when the catalog says the model is text-only, so a missing override blocks image input even against a vision-capable +upstream. + `max_request_body_size_bytes` bounds the serialized request the gateway accepts on `/v1/responses`, `/v1/responses/compact`, `/v1/conversations`, the Anthropic Messages endpoints, and the Responses WebSocket. It counts encoded bytes — JSON overhead, replayed conversation history, and base64 image attachments included — and is unrelated diff --git a/crates/agentic-server-core/tests/cassettes/README.md b/crates/agentic-server-core/tests/cassettes/README.md index 69cdc506..94bb3db4 100644 --- a/crates/agentic-server-core/tests/cassettes/README.md +++ b/crates/agentic-server-core/tests/cassettes/README.md @@ -59,7 +59,10 @@ Each smoke script starts a replay server and Agentic API, then invokes the insta `MESSAGES_GATEWAY_TOOL_ALIASES=WebSearch=web_search`; it asserts the recorded answer, two Messages rounds, one search request, a hidden `tool_result`, cache-bearing system and user blocks, and the exact Qwen model requested by Claude Code 2.1.245. The Codex job asserts the recorded `HELLO` answer, one streaming Responses request, and the exact Qwen -model requested by Codex 0.149.1. +model requested by Codex 0.149.1. It then runs `scripts/codex_image_smoke.py`, which attaches the committed +`images/inputs/red-blue-64.png` through both launcher modes and compares its bytes with the upstream capture. A third +run explicitly advertises text-only and requires the image to be absent. These cases replay the existing Qwen2.5-VL +single-image SSE recording unchanged; they validate client/catalog propagation, not fresh model inference. ## Modes diff --git a/crates/agentic-server/benches/gateway_bench.rs b/crates/agentic-server/benches/gateway_bench.rs index 18a84953..a43ebeb3 100644 --- a/crates/agentic-server/benches/gateway_bench.rs +++ b/crates/agentic-server/benches/gateway_bench.rs @@ -181,6 +181,7 @@ async fn spawn_gateway(llm_url: &str) -> (Arc, String) { llm_api_base: config.llm_api_base, skip_llm_ready_check: config.skip_llm_ready_check, openai_api_key: config.openai_api_key, + model_capabilities: std::sync::Arc::default(), max_request_body_size: DEFAULT_MAX_REQUEST_BODY_SIZE, }; diff --git a/crates/agentic-server/benches/proxy_bench.rs b/crates/agentic-server/benches/proxy_bench.rs index 15e83a69..fbd9923b 100644 --- a/crates/agentic-server/benches/proxy_bench.rs +++ b/crates/agentic-server/benches/proxy_bench.rs @@ -101,6 +101,7 @@ async fn spawn_gateway(config: Config) -> String { llm_api_base: config.llm_api_base, skip_llm_ready_check: config.skip_llm_ready_check, openai_api_key: config.openai_api_key, + model_capabilities: std::sync::Arc::default(), max_request_body_size: DEFAULT_MAX_REQUEST_BODY_SIZE, }; let server_config = ServerConfig::from_env(); diff --git a/crates/agentic-server/src/agentic_harness.rs b/crates/agentic-server/src/agentic_harness.rs index c8c10fa7..335d5f45 100644 --- a/crates/agentic-server/src/agentic_harness.rs +++ b/crates/agentic-server/src/agentic_harness.rs @@ -5,6 +5,7 @@ use std::{ }; use crate::agentic_output::redact_url; +use crate::model_capabilities::InputModalities; #[derive(Debug)] pub struct HarnessEnv { @@ -19,13 +20,18 @@ pub const CLAUDE_CANONICAL_MODEL: &str = "claude-sonnet-4-5-20250929"; /// Write an isolated Codex home for an Agentic API session. /// +/// `input_modalities` must be the modalities the gateway resolved for `model`: Codex strips +/// image content from a request when this catalog says the model accepts text only, so the +/// value is required rather than defaulted. +/// /// # Errors /// /// Returns an I/O error when the temporary home or generated files cannot be written. -pub fn prepare_codex_home( +pub(crate) fn prepare_codex_home( root: &Path, gateway_url: &str, model: &str, + input_modalities: InputModalities, api_key: Option<&str>, ) -> Result { fs::create_dir_all(root)?; @@ -39,7 +45,7 @@ pub fn prepare_codex_home( "supported_in_api": true, "visibility": "list", "priority": 0, - "input_modalities": ["text"], + "input_modalities": input_modalities, "default_reasoning_level": "medium", "supported_reasoning_levels": [ {"effort": "low", "description": "Fast responses"}, @@ -221,11 +227,13 @@ mod tests { use std::fs; use super::{prepare_claude_home, prepare_claude_home_with_state, prepare_codex_home}; + use crate::model_capabilities::InputModalities; #[test] fn codex_config_is_isolated_and_contains_gateway_provider() { let root = unique_temp_dir("codex"); - let env = prepare_codex_home(&root, "http://127.0.0.1:3000", "Qwen/test", None).expect("config"); + let env = prepare_codex_home(&root, "http://127.0.0.1:3000", "Qwen/test", InputModalities::Text, None) + .expect("config"); let config = fs::read_to_string(root.join("config.toml")).expect("config file"); assert!(config.contains("[model_providers.agentic-api]")); @@ -241,14 +249,20 @@ mod tests { #[test] fn codex_home_removes_inherited_openai_key_unless_explicitly_configured() { let root = unique_temp_dir("codex-credential-isolation"); - let without_key = prepare_codex_home(&root, "http://127.0.0.1:3000", "Qwen/test", None) + let without_key = prepare_codex_home(&root, "http://127.0.0.1:3000", "Qwen/test", InputModalities::Text, None) .expect("Codex config without gateway key"); assert!(without_key.environment_remove.contains(&"OPENAI_API_KEY".to_owned())); assert!(!without_key.environment.contains_key("OPENAI_API_KEY")); - let with_key = prepare_codex_home(&root, "http://127.0.0.1:3000", "Qwen/test", Some("gateway-key")) - .expect("Codex config with gateway key"); + let with_key = prepare_codex_home( + &root, + "http://127.0.0.1:3000", + "Qwen/test", + InputModalities::Text, + Some("gateway-key"), + ) + .expect("Codex config with gateway key"); assert_eq!( with_key.environment.get("OPENAI_API_KEY"), Some(&"gateway-key".to_owned()) @@ -444,6 +458,60 @@ mod tests { assert_eq!(mode, 0o700); } + #[test] + fn codex_catalog_advertises_the_resolved_input_modalities() { + for (modalities, expected) in [ + (InputModalities::Text, serde_json::json!(["text"])), + (InputModalities::TextAndImage, serde_json::json!(["text", "image"])), + ] { + let root = unique_temp_dir("codex-modalities"); + prepare_codex_home(&root, "http://127.0.0.1:3000", "Qwen/test", modalities, None).expect("config"); + let catalog: serde_json::Value = + serde_json::from_str(&fs::read_to_string(root.join("model_catalog.json")).expect("catalog file")) + .expect("valid catalog JSON"); + + assert_eq!(catalog["models"][0]["input_modalities"], expected); + assert_eq!(catalog["models"][0]["slug"], "Qwen/test"); + + fs::remove_dir_all(root).expect("cleanup"); + } + } + + #[test] + fn codex_catalog_keeps_its_launcher_specific_tool_settings() { + let root = unique_temp_dir("codex-tool-settings"); + prepare_codex_home( + &root, + "http://127.0.0.1:3000", + "Qwen/test", + InputModalities::TextAndImage, + None, + ) + .expect("config"); + let catalog: serde_json::Value = + serde_json::from_str(&fs::read_to_string(root.join("model_catalog.json")).expect("catalog file")) + .expect("valid catalog JSON"); + let model = &catalog["models"][0]; + + assert_eq!( + model["shell_type"], "local", + "the launcher runs Codex against a local shell" + ); + assert!( + model.get("apply_patch_tool_type").is_none(), + "the launcher omits apply_patch_tool_type so Codex edits through the shell tool" + ); + assert_eq!( + model["truncation_policy"], + serde_json::json!({"limit": 32768, "mode": "tokens"}) + ); + assert_eq!(model["supports_image_detail_original"], false); + assert_eq!(model["web_search_tool_type"], "text"); + assert_eq!(model["include_skills_usage_instructions"], false); + + fs::remove_dir_all(root).expect("cleanup"); + } + fn unique_temp_dir(name: &str) -> std::path::PathBuf { let path = std::env::temp_dir().join(format!("agentic-api-{name}-{}", std::process::id())); fs::create_dir_all(&path).expect("temp dir"); diff --git a/crates/agentic-server/src/agentic_process.rs b/crates/agentic-server/src/agentic_process.rs index 8cfde177..44a8bd0d 100644 --- a/crates/agentic-server/src/agentic_process.rs +++ b/crates/agentic-server/src/agentic_process.rs @@ -1,12 +1,19 @@ +mod catalog; + +pub use catalog::resolve_model; +#[cfg(test)] +use catalog::{CATALOG_MODEL_GRACE, CatalogBudget, MAX_CATALOG_BYTES, catalog_selection}; +use catalog::{CodexModelSelection, resolve_codex_selection}; + use std::{ffi::OsString, path::Path, time::Duration}; use agentic_core::error::Error; use reqwest::Client; -use serde::Deserialize; use tokio::time::{Instant, sleep}; use crate::{ - agentic_cli::{CommonOptions, SourceOptions}, + agentic_cli::{CommonOptions, Harness, HarnessOptions, SourceOptions}, + agentic_harness::HarnessEnv, agentic_output::redact_url, }; @@ -18,6 +25,10 @@ pub const DEFAULT_CLAUDE_EFFORT: &str = "medium"; const CLAUDE_EFFORT_ENV: &str = "AGENTIC_CLAUDE_EFFORT"; const PLACEHOLDER_MODEL: &str = "agentic-api"; const CLAUDE_TOOLS: &str = "Bash,Edit,Read,WebSearch"; +/// Readiness budget for a gateway the launcher did not start. +const ATTACHED_GATEWAY_TIMEOUT: Duration = Duration::from_secs(30); +/// Readiness poll interval for a gateway the launcher did not start. +const ATTACHED_GATEWAY_INTERVAL: Duration = Duration::from_millis(250); #[must_use] pub fn server_args(source: &SourceOptions, common: &CommonOptions) -> Vec { @@ -69,6 +80,18 @@ pub fn claude_effort() -> String { .unwrap_or_else(|| DEFAULT_CLAUDE_EFFORT.to_owned()) } +const fn harness_binary_names(harness: Harness) -> (&'static str, &'static str) { + match harness { + Harness::Codex => ("codex", "AGENTIC_CODEX_BIN"), + Harness::Claude => ("claude", "AGENTIC_CLAUDE_BIN"), + } +} + +fn harness_binary(harness: Harness) -> OsString { + let (binary_name, override_name) = harness_binary_names(harness); + std::env::var_os(override_name).unwrap_or_else(|| binary_name.into()) +} + fn harness_launch_args( harness: crate::agentic_cli::Harness, yolo: bool, @@ -117,76 +140,6 @@ fn validate_claude_passthrough(passthrough: &[String]) -> Result<(), Error> { Ok(()) } -#[derive(Debug, Deserialize)] -struct ModelList { - #[serde(default)] - data: Vec, -} - -#[derive(Debug, Deserialize)] -struct ModelEntry { - id: String, -} - -/// Resolve the harness model: the explicit `--model`, or the first model the upstream serves. -/// -/// # Errors -/// -/// Returns a configuration error when no model is given and the upstream lists none. -pub async fn resolve_model(client: &Client, source: &SourceOptions, api_key: Option<&str>) -> Result { - if let Some(model) = &source.model { - return Ok(model.clone()); - } - let Some(upstream) = &source.upstream else { - return Ok(PLACEHOLDER_MODEL.to_owned()); - }; - let models_url = format!("{}/v1/models", agentic_core::config::normalize_base_url(upstream)); - let display_models_url = redact_url(&models_url); - let display_upstream = redact_url(upstream); - let mut request = client.get(&models_url); - if let Some(api_key) = api_key { - request = request.bearer_auth(api_key); - } - let response = request - .send() - .await - .map_err(|error| { - Error::Config(format!( - "failed to list upstream models at {display_models_url}: {}", - error.without_url() - )) - })? - .error_for_status() - .map_err(|error| { - Error::Config(format!( - "upstream model listing at {display_models_url} failed: {}", - error.without_url() - )) - })?; - let body = response.text().await.map_err(|error| { - Error::Config(format!( - "failed to read model listing from {display_models_url}: {}", - error.without_url() - )) - })?; - let list: ModelList = agentic_core::utils::common::deserialize_from_str(&body) - .map_err(|error| Error::Config(format!("invalid model listing from {display_models_url}: {error}")))?; - let mut ids = list.data.into_iter().map(|entry| entry.id); - let Some(model) = ids.next() else { - return Err(Error::Config(format!( - "upstream {display_upstream} serves no models; pass --model explicitly" - ))); - }; - let remaining = ids.count(); - if remaining > 0 { - eprintln!( - "upstream serves {} models; using {model}. Pass --model to choose another.", - remaining + 1 - ); - } - Ok(model) -} - /// Wait until the gateway is live and, unless skipped, its upstream is ready. /// /// # Errors @@ -236,6 +189,45 @@ pub async fn wait_for_gateway( } } +/// The model a harness will run, with the metadata that harness needs to configure it. +#[derive(Debug)] +enum HarnessModel { + Codex(CodexModelSelection), + Claude(String), +} + +/// Resolve the model each harness will run. +/// +/// Codex reads its model and capabilities from the gateway catalog; Claude Code keeps using the +/// upstream model listing, which needs no capability metadata. +/// +/// # Errors +/// +/// Returns a configuration error when no model can be resolved. +async fn resolve_harness_model( + client: &Client, + harness: Harness, + gateway_url: &str, + options: &HarnessOptions, +) -> Result { + match harness { + Harness::Codex => Ok(HarnessModel::Codex( + resolve_codex_selection( + client, + gateway_url, + options.source.model.as_deref(), + options.common.api_key.as_deref(), + Duration::from_secs_f64(options.common.llm_ready_timeout_s), + Duration::from_secs_f64(options.common.llm_ready_interval_s), + ) + .await?, + )), + Harness::Claude => Ok(HarnessModel::Claude( + resolve_model(client, &options.source, options.common.api_key.as_deref()).await?, + )), + } +} + /// Run one gateway-plus-harness session and return the harness exit status. /// /// # Errors @@ -275,17 +267,16 @@ pub async fn run_session( return Err(error); } - let model = match resolve_model(&client, &options.source, options.common.api_key.as_deref()).await { - Ok(model) => model, + let harness_model = match resolve_harness_model(&client, harness, &gateway_url, &options).await { + Ok(harness_model) => harness_model, Err(error) => { cleanup(&mut server, session_root.path()).await; return Err(error); } }; let harness_env = match harness_environment( - harness, + &harness_model, &gateway_url, - &model, &options, session_root.path(), &claude_state_root, @@ -340,25 +331,25 @@ fn start_server( } fn harness_environment( - harness: crate::agentic_cli::Harness, + harness_model: &HarnessModel, gateway_url: &str, - model: &str, - options: &crate::agentic_cli::HarnessOptions, + options: &HarnessOptions, session_root: &Path, claude_state_root: &Path, -) -> Result { +) -> Result { let inherited_auth_token = std::env::var("ANTHROPIC_AUTH_TOKEN") .ok() .filter(|value| !value.trim().is_empty()); - let mut environment = match harness { - crate::agentic_cli::Harness::Codex => crate::agentic_harness::prepare_codex_home( + let mut environment = match harness_model { + HarnessModel::Codex(selection) => crate::agentic_harness::prepare_codex_home( session_root, gateway_url, - model, + &selection.model, + selection.input_modalities, options.common.api_key.as_deref(), ) .map_err(Error::from), - crate::agentic_cli::Harness::Claude => crate::agentic_harness::prepare_claude_home_with_state( + HarnessModel::Claude(model) => crate::agentic_harness::prepare_claude_home_with_state( session_root, claude_state_root, gateway_url, @@ -368,7 +359,7 @@ fn harness_environment( ) .map_err(Error::from), }?; - if matches!(harness, crate::agentic_cli::Harness::Claude) { + if matches!(harness_model, HarnessModel::Claude(_)) { // Claude Code gives CLAUDE_CODE_EFFORT_LEVEL precedence over --effort, so set both // to keep an inherited `high` from reaching the Qwen chat template. environment @@ -384,15 +375,8 @@ fn spawn_harness( passthrough: &[String], harness_env: &crate::agentic_harness::HarnessEnv, ) -> Result { - let binary_name = match harness { - crate::agentic_cli::Harness::Codex => "codex", - crate::agentic_cli::Harness::Claude => "claude", - }; - let override_name = match harness { - crate::agentic_cli::Harness::Codex => "AGENTIC_CODEX_BIN", - crate::agentic_cli::Harness::Claude => "AGENTIC_CLAUDE_BIN", - }; - let binary = std::env::var_os(override_name).unwrap_or_else(|| binary_name.into()); + let (binary_name, override_name) = harness_binary_names(harness); + let binary = harness_binary(harness); let mut harness_command = build_harness_command(&binary, harness, yolo, passthrough, harness_env); harness_command .spawn() @@ -426,18 +410,22 @@ fn build_harness_command( } fn prepare_attached_harness_environment( - harness: crate::agentic_cli::Harness, + harness_model: &HarnessModel, session_root: &Path, claude_state_root: &Path, gateway_url: &str, - model: &str, api_key: Option<&str>, -) -> Result { - match harness { - crate::agentic_cli::Harness::Codex => { - crate::agentic_harness::prepare_codex_home(session_root, gateway_url, model, api_key).map_err(Error::from) - } - crate::agentic_cli::Harness::Claude => crate::agentic_harness::prepare_claude_home_with_state( +) -> Result { + match harness_model { + HarnessModel::Codex(selection) => crate::agentic_harness::prepare_codex_home( + session_root, + gateway_url, + &selection.model, + selection.input_modalities, + api_key, + ) + .map_err(Error::from), + HarnessModel::Claude(model) => crate::agentic_harness::prepare_claude_home_with_state( session_root, claude_state_root, gateway_url, @@ -469,20 +457,33 @@ pub async fn run_attached_harness( wait_for_gateway( &client, &options.gateway_url, - Duration::from_secs(30), - Duration::from_millis(250), + ATTACHED_GATEWAY_TIMEOUT, + ATTACHED_GATEWAY_INTERVAL, false, ) .await?; + let harness_model = match harness { + Harness::Codex => HarnessModel::Codex( + resolve_codex_selection( + &client, + &options.gateway_url, + Some(&options.model), + options.api_key.as_deref(), + ATTACHED_GATEWAY_TIMEOUT, + ATTACHED_GATEWAY_INTERVAL, + ) + .await?, + ), + Harness::Claude => HarnessModel::Claude(options.model.clone()), + }; let mut harness_env = prepare_attached_harness_environment( - harness, + &harness_model, session_root.path(), &claude_state_root, &options.gateway_url, - &options.model, options.api_key.as_deref(), )?; - if matches!(harness, crate::agentic_cli::Harness::Claude) { + if matches!(harness, Harness::Claude) { harness_env .environment .insert("CLAUDE_CODE_EFFORT_LEVEL".to_owned(), claude_effort()); @@ -560,8 +561,307 @@ fn gateway_client() -> Result { mod tests { use std::ffi::OsString; - use super::{DEFAULT_CLAUDE_EFFORT, harness_launch_args, server_args}; + use super::{CodexModelSelection, DEFAULT_CLAUDE_EFFORT, HarnessModel, harness_launch_args, server_args}; use crate::agentic_cli::{CommonOptions, Harness, SourceOptions}; + use crate::model_capabilities::InputModalities; + + /// A gateway that answers catalog requests from a scripted queue and records what it was asked. + struct MockGateway { + url: String, + requests: std::sync::Arc>>, + } + + impl MockGateway { + fn request_count(&self) -> usize { + self.requests.lock().expect("request log").len() + } + + fn first_request(&self) -> String { + self.requests + .lock() + .expect("request log") + .first() + .cloned() + .unwrap_or_default() + } + } + + /// Serve `responses` in order, repeating the last one once the queue is exhausted. + async fn spawn_mock_gateway(responses: Vec<(&'static str, String)>) -> MockGateway { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("listener"); + let address = listener.local_addr().expect("listener address"); + let requests = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let recorded = std::sync::Arc::clone(&requests); + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let mut served = 0; + loop { + let Ok((mut socket, _)) = listener.accept().await else { + return; + }; + let mut buffer = [0_u8; 4096]; + let read = socket.read(&mut buffer).await.unwrap_or_default(); + recorded + .lock() + .expect("request log") + .push(String::from_utf8_lossy(&buffer[..read]).into_owned()); + let (status, body) = responses + .get(served) + .or_else(|| responses.last()) + .cloned() + .unwrap_or(("200 OK", String::new())); + served += 1; + let response = format!( + "HTTP/1.1 {status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + MockGateway { + url: format!("http://{address}"), + requests, + } + } + + fn catalog_body() -> String { + r#"{"models":[ + {"slug":"first-model","input_modalities":["text"]}, + {"slug":"vision-model","input_modalities":["text","image"]} + ]}"# + .to_owned() + } + + async fn select( + gateway: &MockGateway, + requested_model: Option<&str>, + api_key: Option<&str>, + ) -> Result { + super::catalog_selection( + &reqwest::Client::new(), + &gateway.url, + "9.9.9", + requested_model, + api_key, + super::CatalogBudget { + timeout: std::time::Duration::from_millis(400), + interval: std::time::Duration::from_millis(10), + missing_grace: super::CATALOG_MODEL_GRACE, + }, + ) + .await + } + + #[tokio::test] + async fn catalog_selection_reads_the_resolved_modalities() { + let gateway = spawn_mock_gateway(vec![("200 OK", catalog_body())]).await; + + let selection = select(&gateway, Some("vision-model"), None) + .await + .expect("the catalog lists the requested model"); + + assert_eq!(selection.model, "vision-model"); + assert_eq!(selection.input_modalities, InputModalities::TextAndImage); + let request = gateway.first_request(); + assert!( + request.starts_with("GET /v1/models?client_version=9.9.9 "), + "the gateway only transforms its catalog for a client version: {request}" + ); + assert!( + !request.to_ascii_lowercase().contains("authorization:"), + "no credential must be sent when none is configured" + ); + } + + #[tokio::test] + async fn catalog_selection_defaults_to_the_first_advertised_model() { + let gateway = spawn_mock_gateway(vec![("200 OK", catalog_body())]).await; + + let selection = select(&gateway, None, None).await.expect("a catalog entry is selected"); + + assert_eq!(selection.model, "first-model"); + assert_eq!(selection.input_modalities, InputModalities::Text); + } + + #[tokio::test] + async fn catalog_selection_sends_the_configured_credential() { + let gateway = spawn_mock_gateway(vec![("200 OK", catalog_body())]).await; + + select(&gateway, Some("first-model"), Some("gateway-key")) + .await + .expect("the catalog lists the requested model"); + + assert!( + gateway + .first_request() + .to_ascii_lowercase() + .contains("authorization: bearer gateway-key"), + "the configured API key must reach a protected gateway" + ); + } + + #[tokio::test] + async fn catalog_selection_reports_a_model_the_gateway_does_not_serve() { + let gateway = spawn_mock_gateway(vec![("200 OK", catalog_body())]).await; + + let error = select(&gateway, Some("absent-model"), None) + .await + .expect_err("a model the gateway does not serve must fail"); + let message = error.to_string(); + + assert!(message.contains("absent-model"), "{message}"); + assert!(message.contains("first-model, vision-model"), "{message}"); + } + + #[tokio::test] + async fn catalog_selection_does_not_retry_rejected_credentials() { + let gateway = spawn_mock_gateway(vec![("401 Unauthorized", "{}".to_owned())]).await; + + let error = select(&gateway, Some("first-model"), None) + .await + .expect_err("a rejected credential must fail"); + let message = error.to_string(); + + assert!(message.contains("401"), "{message}"); + assert!(message.contains("--api-key"), "{message}"); + assert_eq!( + gateway.request_count(), + 1, + "authentication failures must not be retried" + ); + } + + #[tokio::test] + async fn catalog_selection_retries_a_warming_gateway() { + let gateway = spawn_mock_gateway(vec![ + ("503 Service Unavailable", "{}".to_owned()), + ("200 OK", catalog_body()), + ]) + .await; + + let selection = select(&gateway, Some("vision-model"), None) + .await + .expect("a warming gateway must be retried"); + + assert_eq!(selection.input_modalities, InputModalities::TextAndImage); + assert_eq!(gateway.request_count(), 2); + } + + #[tokio::test] + async fn catalog_selection_retries_an_empty_catalog() { + let gateway = spawn_mock_gateway(vec![ + ("200 OK", r#"{"models":[]}"#.to_owned()), + ("200 OK", catalog_body()), + ]) + .await; + + let selection = select(&gateway, None, None) + .await + .expect("an upstream that is still loading must be retried"); + + assert_eq!(selection.model, "first-model"); + assert_eq!(gateway.request_count(), 2); + } + + #[tokio::test] + async fn catalog_selection_starts_the_missing_model_grace_at_the_first_miss() { + let warming = ("503 Service Unavailable", "{}".to_owned()); + let without_the_model = ( + "200 OK", + r#"{"models":[{"slug":"other-model","input_modalities":["text"]}]}"#.to_owned(), + ); + let gateway = spawn_mock_gateway(vec![ + warming.clone(), + warming.clone(), + warming.clone(), + warming.clone(), + warming.clone(), + warming, + without_the_model, + ("200 OK", catalog_body()), + ]) + .await; + + // Warm-up alone outlasts the grace: six retries at 30ms exceed the 150ms window, so a + // grace anchored at the first attempt would already have expired by the first miss. + let selection = super::catalog_selection( + &reqwest::Client::new(), + &gateway.url, + "9.9.9", + Some("vision-model"), + None, + super::CatalogBudget { + timeout: std::time::Duration::from_secs(3), + interval: std::time::Duration::from_millis(30), + missing_grace: std::time::Duration::from_millis(150), + }, + ) + .await + .expect("a slow warm-up must not consume the model-missing grace"); + + assert_eq!(selection.model, "vision-model"); + assert_eq!(selection.input_modalities, InputModalities::TextAndImage); + assert_eq!( + gateway.request_count(), + 8, + "every warm-up response, the miss, and the successful catalog must each be requested" + ); + } + + #[tokio::test] + async fn catalog_selection_rejects_an_undecodable_catalog() { + let gateway = spawn_mock_gateway(vec![("200 OK", "not a catalog".to_owned())]).await; + + let error = select(&gateway, Some("first-model"), None) + .await + .expect_err("an undecodable catalog must fail"); + + assert!(error.to_string().contains("not a Codex model catalog"), "{error}"); + assert_eq!(gateway.request_count(), 1, "an undecodable catalog must not be retried"); + } + + #[tokio::test] + async fn catalog_selection_rejects_an_oversized_catalog() { + let oversized = format!( + r#"{{"models":[{{"slug":"{}","input_modalities":["text"]}}]}}"#, + "x".repeat(super::MAX_CATALOG_BYTES + 1) + ); + let gateway = spawn_mock_gateway(vec![("200 OK", oversized)]).await; + + let error = select(&gateway, Some("first-model"), None) + .await + .expect_err("an oversized catalog must fail"); + + assert!(error.to_string().contains("larger than"), "{error}"); + assert_eq!(gateway.request_count(), 1, "an oversized catalog must not be retried"); + } + + #[tokio::test] + async fn catalog_selection_redacts_gateway_credentials_when_unreachable() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("listener"); + let address = listener.local_addr().expect("listener address"); + drop(listener); + + let error = super::catalog_selection( + &reqwest::Client::new(), + &format!("http://agentic:gateway-secret@{address}"), + "9.9.9", + Some("first-model"), + None, + super::CatalogBudget { + timeout: std::time::Duration::from_millis(50), + interval: std::time::Duration::from_millis(10), + missing_grace: super::CATALOG_MODEL_GRACE, + }, + ) + .await + .expect_err("an unreachable gateway must fail"); + let message = error.to_string(); + + assert!(!message.contains("gateway-secret"), "{message}"); + assert!(message.contains("[REDACTED]"), "{message}"); + } #[test] fn integrated_mode_builds_server_arguments() { @@ -708,9 +1008,8 @@ mod tests { let root = std::env::temp_dir().join(format!("agentic-api-effort-test-{}", std::process::id())); let state_root = std::env::temp_dir().join(format!("agentic-api-state-test-{}", std::process::id())); let environment = super::harness_environment( - Harness::Claude, + &HarnessModel::Claude("served-discovered".to_owned()), "http://127.0.0.1:3000", - "served-discovered", &options, &root, &state_root, @@ -746,9 +1045,8 @@ mod tests { harness_args: Vec::new(), }; let environment = super::harness_environment( - Harness::Claude, + &HarnessModel::Claude("served-discovered".to_owned()), "http://127.0.0.1:3000", - "served-discovered", &options, settings_root.path(), state_root.path(), @@ -798,11 +1096,13 @@ mod tests { fn attached_codex_uses_an_isolated_responses_provider() { let root = std::env::temp_dir().join(format!("agentic-api-attached-codex-test-{}", std::process::id())); let environment = super::prepare_attached_harness_environment( - Harness::Codex, + &HarnessModel::Codex(CodexModelSelection { + model: "Qwen/Qwen3-8B".to_owned(), + input_modalities: InputModalities::TextAndImage, + }), &root, &root, "http://127.0.0.1:9000", - "Qwen/Qwen3-8B", None, ) .expect("Codex environment"); @@ -816,6 +1116,14 @@ mod tests { assert!(config.contains("model = \"Qwen/Qwen3-8B\"")); assert!(config.contains("base_url = \"http://127.0.0.1:9000/v1\"")); assert!(config.contains("wire_api = \"responses\"")); + let catalog: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string(root.join("model_catalog.json")).expect("Codex catalog")) + .expect("valid catalog JSON"); + assert_eq!( + catalog["models"][0]["input_modalities"], + serde_json::json!(["text", "image"]), + "the attached launcher must write the modalities the gateway resolved" + ); std::fs::remove_dir_all(root).expect("cleanup"); } @@ -825,11 +1133,10 @@ mod tests { let session_root = tempfile::tempdir().expect("session root"); let state_root = tempfile::tempdir().expect("state root"); let environment = super::prepare_attached_harness_environment( - Harness::Claude, + &HarnessModel::Claude("Qwen/Qwen3-8B".to_owned()), session_root.path(), state_root.path(), "http://127.0.0.1:9000", - "Qwen/Qwen3-8B", None, ) .expect("Claude environment"); @@ -847,11 +1154,10 @@ mod tests { let session_root = tempfile::tempdir().expect("session root"); let state_root = tempfile::tempdir().expect("state root"); let environment = super::prepare_attached_harness_environment( - Harness::Claude, + &HarnessModel::Claude("Qwen/Qwen3-8B".to_owned()), session_root.path(), state_root.path(), "http://127.0.0.1:9000", - "Qwen/Qwen3-8B", None, ) .expect("Claude environment"); @@ -1279,9 +1585,8 @@ mod tests { }; let root = std::env::temp_dir().join(format!("agentic-api-yolo-test-{}", std::process::id())); let environment = super::harness_environment( - Harness::Claude, + &HarnessModel::Claude("served-test".to_owned()), "http://127.0.0.1:3000", - "served-test", &options, &root, &root, diff --git a/crates/agentic-server/src/agentic_process/catalog.rs b/crates/agentic-server/src/agentic_process/catalog.rs new file mode 100644 index 00000000..dd41b033 --- /dev/null +++ b/crates/agentic-server/src/agentic_process/catalog.rs @@ -0,0 +1,443 @@ +//! Resolve harness models and Codex input capabilities from upstream catalogs. + +use std::time::Duration; + +use agentic_core::error::Error; +use reqwest::Client; +use serde::Deserialize; +use tokio::time::{Instant, sleep}; + +use crate::agentic_cli::{Harness, SourceOptions}; +use crate::agentic_output::redact_url; +use crate::model_capabilities::{CodexCatalogCapabilities, InputModalities}; + +use super::{PLACEHOLDER_MODEL, harness_binary}; + +/// Operator-provided Codex version, used instead of probing the Codex binary. +const CODEX_CLIENT_VERSION_ENV: &str = "AGENTIC_CODEX_CLIENT_VERSION"; +/// Bound on the `codex --version` probe. +const CODEX_VERSION_PROBE_TIMEOUT: Duration = Duration::from_secs(5); +/// Upper bound on the catalog payload the launcher reads from a gateway. +pub(super) const MAX_CATALOG_BYTES: usize = 1024 * 1024; +/// How long a served, non-empty catalog may keep omitting the selected model. +/// +/// A catalog that already lists other models proves the upstream is warm, so a missing model +/// is a configuration error rather than a cold start and must not consume the whole budget. +pub(super) const CATALOG_MODEL_GRACE: Duration = Duration::from_secs(10); +#[derive(Debug, Deserialize)] +struct ModelList { + #[serde(default)] + data: Vec, +} + +#[derive(Debug, Deserialize)] +struct ModelEntry { + id: String, +} + +/// Resolve the harness model: the explicit `--model`, or the first model the upstream serves. +/// +/// # Errors +/// +/// Returns a configuration error when no model is given and the upstream lists none. +pub async fn resolve_model(client: &Client, source: &SourceOptions, api_key: Option<&str>) -> Result { + if let Some(model) = &source.model { + return Ok(model.clone()); + } + let Some(upstream) = &source.upstream else { + return Ok(PLACEHOLDER_MODEL.to_owned()); + }; + let models_url = format!("{}/v1/models", agentic_core::config::normalize_base_url(upstream)); + let display_models_url = redact_url(&models_url); + let display_upstream = redact_url(upstream); + let mut request = client.get(&models_url); + if let Some(api_key) = api_key { + request = request.bearer_auth(api_key); + } + let response = request + .send() + .await + .map_err(|error| { + Error::Config(format!( + "failed to list upstream models at {display_models_url}: {}", + error.without_url() + )) + })? + .error_for_status() + .map_err(|error| { + Error::Config(format!( + "upstream model listing at {display_models_url} failed: {}", + error.without_url() + )) + })?; + let body = response.text().await.map_err(|error| { + Error::Config(format!( + "failed to read model listing from {display_models_url}: {}", + error.without_url() + )) + })?; + let list: ModelList = agentic_core::utils::common::deserialize_from_str(&body) + .map_err(|error| Error::Config(format!("invalid model listing from {display_models_url}: {error}")))?; + let mut ids = list.data.into_iter().map(|entry| entry.id); + let Some(model) = ids.next() else { + return Err(Error::Config(format!( + "upstream {display_upstream} serves no models; pass --model explicitly" + ))); + }; + let remaining = ids.count(); + if remaining > 0 { + eprintln!( + "upstream serves {} models; using {model}. Pass --model to choose another.", + remaining + 1 + ); + } + Ok(model) +} + +/// The Codex model the launcher runs and the input modalities the gateway resolved for it. +#[derive(Debug)] +pub(super) struct CodexModelSelection { + pub(super) model: String, + pub(super) input_modalities: InputModalities, +} + +/// The polling budget for one catalog resolution. +#[derive(Clone, Copy, Debug)] +pub(super) struct CatalogBudget { + /// Overall wall-clock budget for resolving the catalog. + pub(super) timeout: Duration, + /// Delay between attempts, unless the gateway asks for a longer one. + pub(super) interval: Duration, + /// How long a served, non-empty catalog may keep omitting the selected model. + pub(super) missing_grace: Duration, +} + +/// Why one catalog attempt failed, and whether another attempt could succeed. +enum CatalogAttempt { + Resolved(CodexModelSelection), + /// The gateway or its upstream may still be warming up. + Transient(Error, Option), + /// Another attempt cannot change the result. + Permanent(Error), + /// The catalog is served and lists models, but not the selected one. + ModelMissing(Error), +} + +enum BodyError { + TooLarge, + Transport(reqwest::Error), +} + +/// Statuses a warming gateway can return before it can serve its catalog. +fn is_transient_status(status: reqwest::StatusCode) -> bool { + status.is_server_error() + || matches!( + status, + reqwest::StatusCode::REQUEST_TIMEOUT + | reqwest::StatusCode::TOO_EARLY + | reqwest::StatusCode::TOO_MANY_REQUESTS + ) +} + +/// `Retry-After` expressed in whole seconds; the HTTP-date form is not honored. +fn retry_after(response: &reqwest::Response) -> Option { + response + .headers() + .get(reqwest::header::RETRY_AFTER)? + .to_str() + .ok()? + .trim() + .parse::() + .ok() + .map(Duration::from_secs) +} + +/// Read a catalog response without trusting the gateway to bound it. +async fn read_bounded_body(mut response: reqwest::Response) -> Result, BodyError> { + if response + .content_length() + .is_some_and(|length| length > MAX_CATALOG_BYTES as u64) + { + return Err(BodyError::TooLarge); + } + let mut body = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(BodyError::Transport)? { + if body.len() + chunk.len() > MAX_CATALOG_BYTES { + return Err(BodyError::TooLarge); + } + body.extend_from_slice(&chunk); + } + Ok(body) +} + +/// The models a catalog advertises, for an error message that names the alternatives. +fn advertised_models(catalog: &CodexCatalogCapabilities) -> String { + const LISTED: usize = 5; + let listed = catalog + .models + .iter() + .take(LISTED) + .map(|entry| entry.slug.as_str()) + .collect::>() + .join(", "); + if catalog.models.len() > LISTED { + format!("{listed}, ...") + } else { + listed + } +} + +/// Resolve the Codex CLI version the gateway catalog is requested for. +/// +/// The gateway only transforms its model list when a client version is present, and Codex +/// reports its own version, so the launcher asks the same binary it is about to run instead of +/// inventing a value. [`CODEX_CLIENT_VERSION_ENV`] skips the probe where it cannot run. +/// +/// # Errors +/// +/// Returns a configuration error when the Codex binary cannot be run or reports no version. +async fn codex_client_version() -> Result { + if let Some(version) = std::env::var(CODEX_CLIENT_VERSION_ENV) + .ok() + .map(|value| value.trim().to_owned()) + .filter(|value| !value.is_empty()) + { + return Ok(version); + } + let binary = harness_binary(Harness::Codex); + let display_binary = binary.to_string_lossy().into_owned(); + let mut command = tokio::process::Command::new(&binary); + command + .arg("--version") + .kill_on_drop(true) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::null()); + let output = tokio::time::timeout(CODEX_VERSION_PROBE_TIMEOUT, command.output()) + .await + .map_err(|_| { + Error::Config(format!( + "{display_binary} --version timed out after {}s; set {CODEX_CLIENT_VERSION_ENV} to skip the probe", + CODEX_VERSION_PROBE_TIMEOUT.as_secs() + )) + })? + .map_err(|error| { + Error::Config(format!( + "failed to run {display_binary} --version: {error}; install Codex or set AGENTIC_CODEX_BIN" + )) + })?; + if !output.status.success() { + return Err(Error::Config(format!( + "{display_binary} --version failed with {}; set {CODEX_CLIENT_VERSION_ENV} to skip the probe", + output.status + ))); + } + String::from_utf8_lossy(&output.stdout) + .lines() + .find(|line| !line.trim().is_empty()) + .and_then(|line| line.split_whitespace().next_back()) + .map(str::to_owned) + .ok_or_else(|| { + Error::Config(format!( + "could not read a version from {display_binary} --version; set {CODEX_CLIENT_VERSION_ENV} to provide it" + )) + }) +} + +/// Ask the gateway once for the model catalog and select the requested model. +async fn catalog_attempt( + client: &Client, + catalog_url: &str, + display_url: &str, + client_version: &str, + requested_model: Option<&str>, + api_key: Option<&str>, +) -> CatalogAttempt { + let mut request = client.get(catalog_url).query(&[("client_version", client_version)]); + if let Some(api_key) = api_key { + request = request.bearer_auth(api_key); + } + let response = match request.send().await { + Ok(response) => response, + Err(error) => { + return CatalogAttempt::Transient( + Error::Config(format!( + "failed to reach the gateway model catalog at {display_url}: {}", + error.without_url() + )), + None, + ); + } + }; + + let status = response.status(); + if !status.is_success() { + let retry_after = retry_after(&response); + let message = format!("the gateway model catalog at {display_url} returned HTTP {status}"); + return if matches!( + status, + reqwest::StatusCode::UNAUTHORIZED | reqwest::StatusCode::FORBIDDEN + ) { + CatalogAttempt::Permanent(Error::Config(format!( + "{message}; pass --api-key if the gateway requires authentication" + ))) + } else if is_transient_status(status) { + CatalogAttempt::Transient(Error::Config(message), retry_after) + } else { + CatalogAttempt::Permanent(Error::Config(message)) + }; + } + + let body = match read_bounded_body(response).await { + Ok(body) => body, + Err(BodyError::TooLarge) => { + return CatalogAttempt::Permanent(Error::Config(format!( + "the gateway model catalog at {display_url} is larger than {MAX_CATALOG_BYTES} bytes" + ))); + } + Err(BodyError::Transport(error)) => { + return CatalogAttempt::Transient( + Error::Config(format!( + "failed to read the gateway model catalog at {display_url}: {}", + error.without_url() + )), + None, + ); + } + }; + + let catalog: CodexCatalogCapabilities = match serde_json::from_slice(&body) { + Ok(catalog) => catalog, + Err(error) => { + return CatalogAttempt::Permanent(Error::Config(format!( + "the gateway model catalog at {display_url} is not a Codex model catalog: {error}" + ))); + } + }; + if catalog.models.is_empty() { + return CatalogAttempt::Transient( + Error::Config(format!("the gateway model catalog at {display_url} lists no models")), + None, + ); + } + let Some(entry) = catalog.select(requested_model) else { + return CatalogAttempt::ModelMissing(Error::Config(format!( + "the gateway model catalog at {display_url} does not list model {:?}; it serves: {}", + requested_model.unwrap_or_default(), + advertised_models(&catalog) + ))); + }; + if requested_model.is_none() && catalog.models.len() > 1 { + eprintln!( + "gateway serves {} models; using {}. Pass --model to choose another.", + catalog.models.len(), + entry.slug + ); + } + CatalogAttempt::Resolved(CodexModelSelection { + model: entry.slug.clone(), + input_modalities: entry.input_modalities, + }) +} + +/// Resolve the Codex model and its input modalities from one gateway catalog snapshot. +/// +/// Selecting the model and reading its capabilities from the same response keeps the isolated +/// Codex catalog consistent with what the gateway serves over HTTP. Transient failures are +/// retried until `timeout` expires, because a gateway can answer `/health` before its upstream +/// can list models; authentication failures and undecodable catalogs are reported immediately. +/// +/// # Errors +/// +/// Returns a configuration error when the catalog cannot be fetched within `timeout`, the +/// gateway rejects the request, or the catalog does not list the selected model. +pub(super) async fn resolve_codex_selection( + client: &Client, + gateway_url: &str, + requested_model: Option<&str>, + api_key: Option<&str>, + timeout: Duration, + interval: Duration, +) -> Result { + let client_version = codex_client_version().await?; + catalog_selection( + client, + gateway_url, + &client_version, + requested_model, + api_key, + CatalogBudget { + timeout, + interval, + missing_grace: CATALOG_MODEL_GRACE, + }, + ) + .await +} + +/// Poll the gateway catalog for `requested_model` until it resolves or the budget expires. +pub(super) async fn catalog_selection( + client: &Client, + gateway_url: &str, + client_version: &str, + requested_model: Option<&str>, + api_key: Option<&str>, + budget: CatalogBudget, +) -> Result { + let base = gateway_url.trim_end_matches('/'); + let catalog_url = format!("{base}/v1/models"); + let display_url = redact_url(base); + let deadline = Instant::now() + budget.timeout; + // Set on the first miss rather than up front: a slow warm-up must not consume the grace a + // served catalog is owed once it starts answering. + let mut missing_deadline = None; + let mut last_error = None; + + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + let Ok(attempt) = tokio::time::timeout( + remaining, + catalog_attempt( + client, + &catalog_url, + &display_url, + client_version, + requested_model, + api_key, + ), + ) + .await + else { + break; + }; + + let delay = match attempt { + CatalogAttempt::Resolved(selection) => return Ok(selection), + CatalogAttempt::Permanent(error) => return Err(error), + CatalogAttempt::ModelMissing(error) => { + let now = Instant::now(); + if now >= *missing_deadline.get_or_insert((now + budget.missing_grace).min(deadline)) { + return Err(error); + } + last_error = Some(error); + budget.interval + } + CatalogAttempt::Transient(error, retry_after) => { + last_error = Some(error); + retry_after.unwrap_or(budget.interval) + } + }; + + let now = Instant::now(); + if now >= deadline { + break; + } + sleep(delay.min(deadline - now)).await; + } + + Err(last_error.unwrap_or_else(|| { + Error::Config(format!( + "the gateway model catalog at {display_url} did not become available" + )) + })) +} diff --git a/crates/agentic-server/src/app.rs b/crates/agentic-server/src/app.rs index ee72794a..8f0e4a32 100644 --- a/crates/agentic-server/src/app.rs +++ b/crates/agentic-server/src/app.rs @@ -19,6 +19,7 @@ use crate::auth::{ANTHROPIC_COUNT_TOKENS_PATH, ANTHROPIC_MESSAGES_PATH, OidcAuth use crate::handler::{ compact_response, conversations, count_tokens, health, messages, models, ready, responses, responses_ws_with_auth, }; +use crate::model_capabilities::ModelCapabilities; /// Default ceiling on serialized inbound request bytes for HTTP bodies and /// WebSocket messages. @@ -250,6 +251,8 @@ pub struct AppState { /// Server-configured API key; used as fallback when the request carries no /// `Authorization` header on the executor path. pub openai_api_key: Option, + /// Configured per-model input-modality overrides applied to the Codex model catalog. + pub model_capabilities: Arc, /// Ceiling on serialized inbound request bytes, applied uniformly to every /// request-bearing endpoint and to WebSocket messages and frames. /// diff --git a/crates/agentic-server/src/config_file.rs b/crates/agentic-server/src/config_file.rs index abcb91b7..743dfc99 100644 --- a/crates/agentic-server/src/config_file.rs +++ b/crates/agentic-server/src/config_file.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::io::Write; use std::num::NonZeroUsize; use std::path::Path; @@ -6,6 +6,7 @@ use std::path::Path; use agentic_core::McpServerEntry; use agentic_core::config::{CONFIG_FILE_NAME, WebSearchProviderKind}; use agentic_core::error::Error; +use agentic_server::model_capabilities::{InputModalities, ModelCapabilities}; use serde::{Deserialize, Serialize}; #[derive(Debug, Default, Deserialize, Serialize)] @@ -86,6 +87,20 @@ impl MessagesGatewayFileConfig { } } +/// Per-model overrides for capabilities the gateway cannot infer from upstream metadata. +#[derive(Debug, Default, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +pub(crate) struct ModelFileConfig { + #[serde(skip_serializing_if = "Option::is_none")] + pub input_modalities: Option, +} + +impl ModelFileConfig { + fn is_empty(&self) -> bool { + self.input_modalities.is_none() + } +} + #[derive(Debug, Default, Deserialize, Serialize)] #[serde(default, deny_unknown_fields)] pub(crate) struct FileConfig { @@ -103,6 +118,8 @@ pub(crate) struct FileConfig { pub tools: ToolsFileConfig, #[serde(skip_serializing_if = "MessagesGatewayFileConfig::is_empty")] pub messages_gateway: MessagesGatewayFileConfig, + #[serde(skip_serializing_if = "BTreeMap::is_empty")] + pub models: BTreeMap, #[serde(skip_serializing_if = "HashMap::is_empty")] pub mcp_servers: HashMap, } @@ -205,6 +222,16 @@ impl FileConfig { Ok(self) } + /// Build the capability resolver from the configured per-model overrides. + pub(crate) fn model_capabilities(&self) -> ModelCapabilities { + ModelCapabilities::new( + self.models + .iter() + .filter_map(|(model_id, model)| Some((model_id.clone(), model.input_modalities?))) + .collect(), + ) + } + fn validate(&self, path: &Path) -> Result<(), Error> { if self .web_search @@ -223,6 +250,20 @@ impl FileConfig { path.display() ))); } + for (model_id, model) in &self.models { + if model_id.trim().is_empty() { + return Err(Error::Config(format!( + "configuration file {} contains an empty model ID", + path.display() + ))); + } + if model.is_empty() { + return Err(Error::Config(format!( + "configuration file {} contains no settings for model {model_id:?}; set input_modalities", + path.display() + ))); + } + } if let Some(label) = self.mcp_servers.keys().find(|label| label.trim().is_empty()) { return Err(Error::Config(format!( "configuration file {} contains an empty MCP server label: {label:?}", @@ -261,6 +302,7 @@ mod tests { use std::num::NonZeroUsize; use agentic_core::McpServerEntry; + use agentic_server::model_capabilities::{InputModalities, UpstreamCapabilities}; use tempfile::tempdir; use super::{FileConfig, McpFileConfig, ServerFileConfig, WebSearchFileConfig, WebSearchProviderKind}; @@ -307,6 +349,7 @@ mod tests { assert!(contents.contains("max_request_body_size_bytes = 20971520")); assert!(!contents.contains("YOU_API_KEY =")); assert!(!contents.contains("[mcp_servers]")); + assert!(!contents.contains("[models")); #[cfg(unix)] { @@ -341,6 +384,123 @@ mod tests { Some(["say_hello".to_owned(), "sum".to_owned()].as_slice()) ); assert_eq!(config.mcp_servers["remote"].require_approval(), Some("never")); + assert_eq!( + config.models["Qwen/Qwen3-VL-8B-Instruct"].input_modalities, + Some(InputModalities::TextAndImage) + ); + } + + #[test] + fn model_overrides_take_precedence_over_upstream_metadata() { + let home = tempdir().expect("temp home"); + fs::write( + home.path().join("config.toml"), + concat!( + "[models.\"vision-model\"]\ninput_modalities = [\"text\", \"image\"]\n\n", + "[models.\"pinned-text-model\"]\ninput_modalities = [\"text\"]\n", + ), + ) + .expect("write config"); + + let capabilities = FileConfig::load(home.path()) + .expect("per-model overrides must parse") + .expect("existing config") + .model_capabilities(); + let advertises_image = UpstreamCapabilities { + image: true, + reasoning: false, + }; + + assert_eq!( + capabilities.resolve("vision-model", UpstreamCapabilities::default()), + InputModalities::TextAndImage + ); + assert_eq!( + capabilities.resolve("pinned-text-model", advertises_image), + InputModalities::Text + ); + assert_eq!( + capabilities.resolve("unconfigured-model", advertises_image), + InputModalities::TextAndImage + ); + assert_eq!( + capabilities.resolve("unconfigured-model", UpstreamCapabilities::default()), + InputModalities::Text + ); + } + + #[test] + fn rejects_unknown_input_modality() { + let home = tempdir().expect("temp home"); + fs::write( + home.path().join("config.toml"), + "[models.\"vision-model\"]\ninput_modalities = [\"text\", \"video\"]\n", + ) + .expect("write config"); + + let error = FileConfig::load(home.path()).expect_err("an unknown modality must fail"); + let message = error.to_string(); + + assert!(message.contains("config.toml"), "{message}"); + assert!(message.contains("video"), "{message}"); + assert!(message.contains("image"), "{message}"); + } + + #[test] + fn rejects_unusable_input_modality_lists() { + for (modalities, expected) in [ + ("[]", "at least one modality"), + ("[\"image\"]", "must include \"text\""), + ("[\"text\", \"text\"]", "more than once"), + ] { + let home = tempdir().expect("temp home"); + fs::write( + home.path().join("config.toml"), + format!("[models.\"a-model\"]\ninput_modalities = {modalities}\n"), + ) + .expect("write config"); + + let error = FileConfig::load(home.path()).expect_err("an unusable modality list must fail"); + assert!(error.to_string().contains(expected), "{modalities}: {error}"); + } + } + + #[test] + fn rejects_empty_model_id() { + let home = tempdir().expect("temp home"); + fs::write( + home.path().join("config.toml"), + "[models.\" \"]\ninput_modalities = [\"text\"]\n", + ) + .expect("write config"); + + let error = FileConfig::load(home.path()).expect_err("an empty model ID must fail"); + assert!(error.to_string().contains("empty model ID"), "{error}"); + } + + #[test] + fn rejects_model_section_without_settings() { + let home = tempdir().expect("temp home"); + fs::write(home.path().join("config.toml"), "[models.\"a-model\"]\n").expect("write config"); + + let error = FileConfig::load(home.path()).expect_err("an empty model section must fail"); + let message = error.to_string(); + + assert!(message.contains("no settings for model \"a-model\""), "{message}"); + assert!(message.contains("input_modalities"), "{message}"); + } + + #[test] + fn rejects_unknown_model_setting() { + let home = tempdir().expect("temp home"); + fs::write( + home.path().join("config.toml"), + "[models.\"a-model\"]\noutput_modalities = [\"text\"]\n", + ) + .expect("write config"); + + let error = FileConfig::load(home.path()).expect_err("an unknown model setting must fail"); + assert!(error.to_string().contains("unknown field"), "{error}"); } #[test] diff --git a/crates/agentic-server/src/handler/http/models.rs b/crates/agentic-server/src/handler/http/models.rs index 199ae410..5dcd301d 100644 --- a/crates/agentic-server/src/handler/http/models.rs +++ b/crates/agentic-server/src/handler/http/models.rs @@ -1,11 +1,10 @@ use std::future::Future; -use std::sync::OnceLock; use axum::extract::{Query, State}; use axum::http::HeaderMap; use axum::response::{IntoResponse, Response}; use http::StatusCode; -use serde_json::{Value, json}; +use serde::{Deserialize, Serialize}; use tracing::{debug, info, warn}; use agentic_core::proxy::{ProxyBody, ProxyResponse, error_response, proxy_get}; @@ -13,94 +12,202 @@ use agentic_core::readiness::{LLM_READINESS_PROBE_TIMEOUT, LlmReadiness, probe_l use super::super::common::convert_response; use crate::app::AppState; +use crate::model_capabilities::{InputModalities, ModelCapabilities, UpstreamCapabilities}; -/// Static fields shared by every Codex `ModelInfo` entry. +/// One model entry of an upstream OpenAI-compatible `/v1/models` payload. /// -/// Built once on first use; cloned per model and patched with the per-model -/// values (`slug`, `display_name`, `auto_review_model_override`, -/// `supports_reasoning_summaries`, `input_modalities`, and optionally -/// `context_window` / `max_context_window`). -fn codex_model_template() -> &'static Value { - static TEMPLATE: OnceLock = OnceLock::new(); - TEMPLATE.get_or_init(|| { - json!({ - "supported_in_api": true, - "priority": 1, - "shell_type": "shell_command", - "visibility": "list", - "base_instructions": "", - "supported_reasoning_levels": [ - {"effort": "low", "description": "Fast responses with lighter reasoning"}, - {"effort": "medium", "description": "Balances speed and reasoning depth"}, - {"effort": "high", "description": "Greater reasoning depth for complex problems"} - ], - "default_reasoning_summary": "auto", - "support_verbosity": false, - "default_verbosity": null, - "apply_patch_tool_type": "freeform", - "web_search_tool_type": "text", - "truncation_policy": {"mode": "bytes", "limit": 100_000}, - "supports_parallel_tool_calls": true, - "supports_image_detail_original": false, - "effective_context_window_percent": 95, - "experimental_supported_tools": [], - "supports_search_tool": false, - "use_responses_lite": false, - "tool_mode": null, - "multi_agent_version": null, - }) - }) +/// Every field is optional: the payload comes from a third-party inference service whose +/// schema the gateway does not control, and an entry that omits metadata must degrade to the +/// conservative defaults rather than drop the model. +#[derive(Debug, Default, Deserialize)] +#[serde(default)] +struct UpstreamModel { + id: Option, + name: Option, + /// vLLM reports the context window here. + max_model_len: Option, + /// Other OpenAI-compatible providers report the context window here. + context_length: Option, + /// Vendor-defined capability strings; only recognized values are honored. + capabilities: Option>, +} + +/// An upstream OpenAI-compatible `/v1/models` payload. +#[derive(Debug, Default, Deserialize)] +struct UpstreamModelList { + #[serde(default)] + data: Vec, +} + +/// Reasoning effort levels Codex offers for a model. +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +enum ReasoningEffort { + Low, + Medium, + High, +} + +/// One selectable reasoning level with its Codex-facing description. +#[derive(Debug, Serialize)] +struct SupportedReasoningLevel { + effort: ReasoningEffort, + description: &'static str, +} + +/// How Codex truncates oversized tool output for a model. +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +enum TruncationMode { + Bytes, +} + +/// The truncation policy advertised to Codex. +#[derive(Debug, Serialize)] +struct TruncationPolicy { + mode: TruncationMode, + limit: u32, } -/// Transform a single upstream model entry into a Codex `ModelInfo` object. +/// A Codex `ModelInfo` entry. /// -/// Returns `None` when the entry has no `id` field (malformed upstream data). -fn upstream_model_to_codex(m: &Value) -> Option { - let id = m["id"].as_str()?.to_owned(); - let display_name = m.get("name").and_then(Value::as_str).unwrap_or(&id).to_owned(); - // vLLM uses max_model_len; other providers may use context_length - let context_length = m["max_model_len"].as_i64().or_else(|| m["context_length"].as_i64()); - // Single pass over capabilities for both flags - let (supports_reasoning, supports_image) = m["capabilities"].as_array().map_or((false, false), |c| { - c.iter().fold((false, false), |(r, i), v| { - let s = v.as_str(); - (r || s == Some("reasoning"), i || s == Some("image")) - }) - }); - let input_modalities = if supports_image { - json!(["text", "image"]) - } else { - json!(["text"]) - }; +/// [`Default`] carries every field that is identical for all served models; the per-model +/// values are set by [`upstream_model_to_codex`]. +#[derive(Debug, Serialize)] +// The boolean flags mirror Codex's `ModelInfo` schema; grouping them would change the wire shape. +#[allow(clippy::struct_excessive_bools)] +struct CodexModelInfo { + slug: String, + display_name: String, + auto_review_model_override: String, + supported_in_api: bool, + priority: u8, + shell_type: &'static str, + visibility: &'static str, + base_instructions: &'static str, + supported_reasoning_levels: Vec, + supports_reasoning_summaries: bool, + default_reasoning_summary: &'static str, + support_verbosity: bool, + /// Always null: the gateway does not advertise a verbosity default. + default_verbosity: Option, + apply_patch_tool_type: &'static str, + web_search_tool_type: &'static str, + truncation_policy: TruncationPolicy, + supports_parallel_tool_calls: bool, + /// Original-detail support is not inferred from image input capability. + supports_image_detail_original: bool, + input_modalities: InputModalities, + effective_context_window_percent: u8, + experimental_supported_tools: Vec, + supports_search_tool: bool, + use_responses_lite: bool, + /// Always null: the gateway does not pin a Codex tool mode. + tool_mode: Option, + /// Always null: the gateway does not advertise a multi-agent version. + multi_agent_version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + context_window: Option, + #[serde(skip_serializing_if = "Option::is_none")] + max_context_window: Option, +} - let mut model = codex_model_template().clone(); - let obj = model.as_object_mut().expect("template is object"); - obj.insert("slug".into(), json!(id)); - obj.insert("display_name".into(), json!(display_name)); - obj.insert("auto_review_model_override".into(), json!(id)); - obj.insert("supports_reasoning_summaries".into(), json!(supports_reasoning)); - obj.insert("input_modalities".into(), input_modalities); - if let Some(ctx) = context_length { - obj.insert("context_window".into(), json!(ctx)); - obj.insert("max_context_window".into(), json!(ctx)); +impl Default for CodexModelInfo { + fn default() -> Self { + Self { + slug: String::new(), + display_name: String::new(), + auto_review_model_override: String::new(), + supported_in_api: true, + priority: 1, + shell_type: "shell_command", + visibility: "list", + base_instructions: "", + supported_reasoning_levels: vec![ + SupportedReasoningLevel { + effort: ReasoningEffort::Low, + description: "Fast responses with lighter reasoning", + }, + SupportedReasoningLevel { + effort: ReasoningEffort::Medium, + description: "Balances speed and reasoning depth", + }, + SupportedReasoningLevel { + effort: ReasoningEffort::High, + description: "Greater reasoning depth for complex problems", + }, + ], + supports_reasoning_summaries: false, + default_reasoning_summary: "auto", + support_verbosity: false, + default_verbosity: None, + apply_patch_tool_type: "freeform", + web_search_tool_type: "text", + truncation_policy: TruncationPolicy { + mode: TruncationMode::Bytes, + limit: 100_000, + }, + supports_parallel_tool_calls: true, + supports_image_detail_original: false, + input_modalities: InputModalities::Text, + effective_context_window_percent: 95, + experimental_supported_tools: Vec::new(), + supports_search_tool: false, + use_responses_lite: false, + tool_mode: None, + multi_agent_version: None, + context_window: None, + max_context_window: None, + } } +} + +/// The Codex `ModelsResponse` returned to a Codex client. +#[derive(Debug, Serialize)] +struct CodexModelsResponse { + models: Vec, +} + +/// Transform a single upstream model entry into a Codex `ModelInfo` entry. +/// +/// Returns `None` when the entry has no `id` field (malformed upstream data). +fn upstream_model_to_codex(model: &UpstreamModel, capabilities: &ModelCapabilities) -> Option { + let id = model.id.as_deref()?; + let context_window = model.max_model_len.or(model.context_length); + let upstream = UpstreamCapabilities::from_advertised(model.capabilities.as_deref().unwrap_or_default()); - Some(model) + Some(CodexModelInfo { + slug: id.to_owned(), + display_name: model.name.as_deref().unwrap_or(id).to_owned(), + auto_review_model_override: id.to_owned(), + supports_reasoning_summaries: upstream.reasoning, + input_modalities: capabilities.resolve(id, upstream), + context_window, + max_context_window: context_window, + ..CodexModelInfo::default() + }) } /// Build the Codex `ModelsResponse` from a raw upstream vLLM models payload. -fn build_codex_models_response(upstream_bytes: &[u8]) -> Value { - let models: Vec = serde_json::from_slice::(upstream_bytes) - .ok() - .and_then(|mut v| match v["data"].take() { - Value::Array(arr) => Some(arr), - _ => None, - }) - .into_iter() - .flatten() - .filter_map(|m| upstream_model_to_codex(&m)) - .collect(); - json!({ "models": models }) +/// +/// # Errors +/// +/// Returns the deserialization error when the upstream payload is not a model list. Reporting +/// it keeps an undecodable upstream response from being served as an empty catalog, which +/// Codex would show as a gateway that serves no models. +fn build_codex_models_response( + upstream_bytes: &[u8], + capabilities: &ModelCapabilities, +) -> Result { + let upstream: UpstreamModelList = serde_json::from_slice(upstream_bytes)?; + + Ok(CodexModelsResponse { + models: upstream + .data + .iter() + .filter_map(|model| upstream_model_to_codex(model, capabilities)) + .collect(), + }) } #[cfg_attr(feature = "openapi", utoipa::path( @@ -264,7 +371,7 @@ pub async fn models(State(state): State, headers: HeaderMap, Query(par let ProxyBody::Full(upstream_bytes) = upstream.body else { return convert_response(error_response( - http::StatusCode::BAD_GATEWAY, + StatusCode::BAD_GATEWAY, "upstream_unavailable", "unexpected streaming response from /v1/models", )); @@ -277,16 +384,255 @@ pub async fn models(State(state): State, headers: HeaderMap, Query(par }); } - axum::Json(build_codex_models_response(&upstream_bytes)).into_response() + match build_codex_models_response(&upstream_bytes, &state.model_capabilities) { + Ok(response) => axum::Json(response).into_response(), + Err(error) => { + warn!(error = %error, "upstream /v1/models payload could not be decoded"); + convert_response(error_response( + StatusCode::BAD_GATEWAY, + "upstream_unavailable", + "invalid model list from /v1/models", + )) + } + } } #[cfg(test)] mod tests { + use std::collections::BTreeMap; use std::future; use std::time::Duration; - use super::dependencies_are_ready; + use serde_json::{Value, json}; + + use super::{build_codex_models_response, dependencies_are_ready}; use crate::app::ReadinessTracker; + use crate::model_capabilities::{InputModalities, ModelCapabilities}; + + const UPSTREAM_MODELS: &str = r#"{ + "object": "list", + "data": [ + {"id": "vision-model", "max_model_len": 32768}, + {"id": "upstream-image-model", "capabilities": ["image", "reasoning"]}, + {"id": "pinned-text-model", "capabilities": ["image"]}, + {"id": "plain-model", "name": "Plain Model", "context_length": 8192} + ] + }"#; + + fn configured_capabilities() -> ModelCapabilities { + ModelCapabilities::new(BTreeMap::from([ + ("vision-model".to_owned(), InputModalities::TextAndImage), + ("pinned-text-model".to_owned(), InputModalities::Text), + ])) + } + + fn catalog(payload: &str, capabilities: &ModelCapabilities) -> Vec { + let response = build_codex_models_response(payload.as_bytes(), capabilities).expect("decodable payload"); + let serialized = serde_json::to_value(&response).expect("serialize catalog"); + serialized["models"].as_array().cloned().expect("models array") + } + + fn entry(models: &[Value], slug: &str) -> Value { + models + .iter() + .find(|model| model["slug"] == slug) + .unwrap_or_else(|| panic!("catalog must contain {slug}")) + .clone() + } + + #[test] + fn catalog_resolves_modalities_by_configured_precedence() { + let models = catalog(UPSTREAM_MODELS, &configured_capabilities()); + + assert_eq!( + entry(&models, "vision-model")["input_modalities"], + json!(["text", "image"]), + "a configured vision model must advertise images without upstream metadata" + ); + assert_eq!( + entry(&models, "upstream-image-model")["input_modalities"], + json!(["text", "image"]), + "recognized upstream metadata must be honored without an override" + ); + assert_eq!( + entry(&models, "pinned-text-model")["input_modalities"], + json!(["text"]), + "an explicit text-only override must win over upstream image metadata" + ); + assert_eq!( + entry(&models, "plain-model")["input_modalities"], + json!(["text"]), + "an unknown model without metadata must stay text-only" + ); + } + + #[test] + fn catalog_stays_text_only_without_configuration() { + let models = catalog(UPSTREAM_MODELS, &ModelCapabilities::default()); + + assert_eq!( + entry(&models, "vision-model")["input_modalities"], + json!(["text"]), + "without an override a model with no upstream metadata stays text-only" + ); + assert_eq!( + entry(&models, "upstream-image-model")["input_modalities"], + json!(["text", "image"]), + "upstream image metadata is honored on its own" + ); + assert_eq!( + entry(&models, "pinned-text-model")["input_modalities"], + json!(["text", "image"]), + "the text-only pin comes from configuration, not from upstream metadata" + ); + assert_eq!(entry(&models, "plain-model")["input_modalities"], json!(["text"])); + } + + #[test] + fn catalog_entries_keep_their_static_codex_settings() { + let models = catalog(UPSTREAM_MODELS, &configured_capabilities()); + let model = entry(&models, "vision-model"); + + assert_eq!(model["supports_image_detail_original"], json!(false)); + assert_eq!(model["shell_type"], json!("shell_command")); + assert_eq!(model["apply_patch_tool_type"], json!("freeform")); + assert_eq!(model["web_search_tool_type"], json!("text")); + assert_eq!(model["truncation_policy"], json!({"mode": "bytes", "limit": 100_000})); + assert_eq!(model["effective_context_window_percent"], json!(95)); + assert_eq!(model["supported_in_api"], json!(true)); + assert_eq!(model["visibility"], json!("list")); + assert_eq!(model["default_verbosity"], Value::Null); + assert_eq!(model["tool_mode"], Value::Null); + assert_eq!(model["multi_agent_version"], Value::Null); + assert_eq!( + model["supported_reasoning_levels"], + json!([ + {"effort": "low", "description": "Fast responses with lighter reasoning"}, + {"effort": "medium", "description": "Balances speed and reasoning depth"}, + {"effort": "high", "description": "Greater reasoning depth for complex problems"} + ]) + ); + } + + #[test] + fn catalog_entries_keep_the_codex_key_set() { + let models = catalog(UPSTREAM_MODELS, &configured_capabilities()); + let model = entry(&models, "vision-model"); + let mut keys: Vec<&str> = model + .as_object() + .expect("catalog entry is an object") + .keys() + .map(String::as_str) + .collect(); + keys.sort_unstable(); + + assert_eq!( + keys, + [ + "apply_patch_tool_type", + "auto_review_model_override", + "base_instructions", + "context_window", + "default_reasoning_summary", + "default_verbosity", + "display_name", + "effective_context_window_percent", + "experimental_supported_tools", + "input_modalities", + "max_context_window", + "multi_agent_version", + "priority", + "shell_type", + "slug", + "support_verbosity", + "supported_in_api", + "supported_reasoning_levels", + "supports_image_detail_original", + "supports_parallel_tool_calls", + "supports_reasoning_summaries", + "supports_search_tool", + "tool_mode", + "truncation_policy", + "use_responses_lite", + "visibility", + "web_search_tool_type", + ], + "the Codex ModelInfo key set is a wire contract" + ); + } + + #[test] + fn catalog_entries_keep_their_upstream_identity_and_context_window() { + let models = catalog(UPSTREAM_MODELS, &configured_capabilities()); + + let vision = entry(&models, "vision-model"); + assert_eq!(vision["display_name"], json!("vision-model")); + assert_eq!(vision["auto_review_model_override"], json!("vision-model")); + assert_eq!(vision["context_window"], json!(32768)); + assert_eq!(vision["max_context_window"], json!(32768)); + + let plain = entry(&models, "plain-model"); + assert_eq!(plain["display_name"], json!("Plain Model")); + assert_eq!(plain["context_window"], json!(8192), "context_length is the fallback"); + + let without_window = entry(&models, "pinned-text-model"); + assert!( + without_window.get("context_window").is_none(), + "an unknown context window must stay absent" + ); + assert!(without_window.get("max_context_window").is_none()); + } + + #[test] + fn reasoning_capability_drives_reasoning_summaries() { + let models = catalog(UPSTREAM_MODELS, &configured_capabilities()); + + assert_eq!( + entry(&models, "upstream-image-model")["supports_reasoning_summaries"], + json!(true) + ); + assert_eq!( + entry(&models, "vision-model")["supports_reasoning_summaries"], + json!(false) + ); + } + + #[test] + fn unrecognized_capability_strings_never_advertise_images() { + let payload = r#"{"data": [{"id": "vl-model", "capabilities": ["vision", "multimodal", "IMAGE"]}]}"#; + let models = catalog(payload, &ModelCapabilities::default()); + + assert_eq!(entry(&models, "vl-model")["input_modalities"], json!(["text"])); + } + + #[test] + fn entries_without_an_identifier_are_skipped() { + let payload = r#"{"data": [{"object": "model"}, {"id": "kept-model"}]}"#; + let models = catalog(payload, &ModelCapabilities::default()); + + assert_eq!(models.len(), 1); + assert_eq!(models[0]["slug"], json!("kept-model")); + } + + #[test] + fn a_payload_without_models_yields_an_empty_catalog() { + assert!(catalog("{}", &ModelCapabilities::default()).is_empty()); + assert!(catalog(r#"{"object": "list", "data": []}"#, &ModelCapabilities::default()).is_empty()); + } + + #[test] + fn undecodable_payloads_are_reported_instead_of_emptied() { + for payload in [ + "not json", + r#"{"data": "not-a-list"}"#, + r#"{"data": [{"id": "a-model", "max_model_len": "32k"}]}"#, + ] { + assert!( + build_codex_models_response(payload.as_bytes(), &ModelCapabilities::default()).is_err(), + "{payload} must not be served as an empty catalog" + ); + } + } #[test] fn readiness_tracker_reports_only_state_transitions() { diff --git a/crates/agentic-server/src/lib.rs b/crates/agentic-server/src/lib.rs index 4819b6ce..08b5a457 100644 --- a/crates/agentic-server/src/lib.rs +++ b/crates/agentic-server/src/lib.rs @@ -5,5 +5,6 @@ pub mod agentic_process; pub mod app; pub mod auth; pub mod handler; +pub mod model_capabilities; #[cfg(feature = "openapi")] pub mod openapi; diff --git a/crates/agentic-server/src/main.rs b/crates/agentic-server/src/main.rs index c2edf022..1f4bbaa6 100644 --- a/crates/agentic-server/src/main.rs +++ b/crates/agentic-server/src/main.rs @@ -339,6 +339,7 @@ fn gateway_options<'a>( oidc: Option, ) -> Result, Error> { Ok(GatewayOptions { + model_capabilities: file.model_capabilities(), host: &common.gateway_host, port: common.gateway_port, max_request_body_size: resolve_max_request_body_size( diff --git a/crates/agentic-server/src/model_capabilities.rs b/crates/agentic-server/src/model_capabilities.rs new file mode 100644 index 00000000..915f523a --- /dev/null +++ b/crates/agentic-server/src/model_capabilities.rs @@ -0,0 +1,434 @@ +//! Model input-modality capabilities shared by the Codex catalog handler, the gateway +//! configuration loader, and the harness launcher. +//! +//! Codex strips image content from a request when its local model catalog says the selected +//! model accepts text only, so the catalog served over HTTP and the isolated catalog written +//! by a launcher must advertise the same resolved modalities. This module owns that contract: +//! the validated modality set, the upstream metadata the gateway recognizes, the local +//! override table, and the resolution order between them. + +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; + +/// Upstream capability string recognized as image input support. +const IMAGE_CAPABILITY: &str = "image"; +/// Upstream capability string recognized as reasoning support. +const REASONING_CAPABILITY: &str = "reasoning"; + +/// One input modality a served model accepts. +#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Modality { + /// Text input, which every model served through the gateway accepts. + Text, + /// Image input. + Image, +} + +impl Modality { + /// The wire representation of this modality. + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Text => "text", + Self::Image => "image", + } + } +} + +/// Rejected input-modality configuration. +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum ModalityError { + /// The list held no modalities. + #[error("input_modalities must list at least one modality; expected [\"text\"] or [\"text\", \"image\"]")] + Empty, + /// The list repeated a modality. + #[error("input_modalities lists \"{}\" more than once", .0.as_str())] + Duplicate(Modality), + /// The list omitted text. + #[error("input_modalities must include \"text\"; a coding harness cannot use an image-only model")] + MissingText, +} + +/// The validated input modalities advertised for one served model. +/// +/// A coding harness always sends instructions, history, and tool output as text, so text is +/// mandatory and the only usable combinations are text and text-with-image. Every other +/// combination is rejected while parsing, which keeps empty, duplicated, image-only, and +/// misordered states unrepresentable. +#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)] +#[serde(try_from = "Vec", into = "Vec")] +pub enum InputModalities { + /// Text input only: the conservative fallback when nothing advertises image support. + #[default] + Text, + /// Text and image input. + TextAndImage, +} + +impl InputModalities { + /// The modalities in the canonical wire order. + #[must_use] + pub const fn as_slice(self) -> &'static [Modality] { + match self { + Self::Text => &[Modality::Text], + Self::TextAndImage => &[Modality::Text, Modality::Image], + } + } + + /// Whether `modality` is accepted. + #[must_use] + pub const fn contains(self, modality: Modality) -> bool { + matches!( + (self, modality), + (Self::Text | Self::TextAndImage, Modality::Text) | (Self::TextAndImage, Modality::Image) + ) + } + + /// Whether image input is accepted. + #[must_use] + pub const fn supports_image(self) -> bool { + matches!(self, Self::TextAndImage) + } + + /// Validate an unordered modality list. + /// + /// # Errors + /// + /// Returns [`ModalityError`] when the list is empty, repeats a modality, or omits text. + pub fn try_new(values: &[Modality]) -> Result { + let mut text = false; + let mut image = false; + for value in values { + let seen = match value { + Modality::Text => &mut text, + Modality::Image => &mut image, + }; + if *seen { + return Err(ModalityError::Duplicate(*value)); + } + *seen = true; + } + match (text, image) { + (true, true) => Ok(Self::TextAndImage), + (true, false) => Ok(Self::Text), + (false, true) => Err(ModalityError::MissingText), + (false, false) => Err(ModalityError::Empty), + } + } +} + +impl TryFrom> for InputModalities { + type Error = ModalityError; + + fn try_from(values: Vec) -> Result { + Self::try_new(&values) + } +} + +impl From for Vec { + fn from(modalities: InputModalities) -> Self { + modalities.as_slice().to_vec() + } +} + +/// Capabilities an upstream `/v1/models` entry advertises. +/// +/// Unrecognized capability strings are ignored: the upstream vocabulary is vendor-defined and +/// may grow, and an unknown string is never evidence of a capability. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub struct UpstreamCapabilities { + /// The entry advertises image input. + pub image: bool, + /// The entry advertises reasoning output. + pub reasoning: bool, +} + +impl UpstreamCapabilities { + /// Recognize the capability strings an upstream model entry advertises. + #[must_use] + pub fn from_advertised>(entries: &[S]) -> Self { + entries + .iter() + .fold(Self::default(), |capabilities, entry| match entry.as_ref() { + IMAGE_CAPABILITY => Self { + image: true, + ..capabilities + }, + REASONING_CAPABILITY => Self { + reasoning: true, + ..capabilities + }, + _ => capabilities, + }) + } +} + +/// Local input-modality overrides, keyed by served model ID. +#[derive(Clone, Debug, Default)] +pub struct ModelCapabilities { + overrides: BTreeMap, +} + +impl ModelCapabilities { + /// Build a resolver from configured per-model overrides. + #[must_use] + pub fn new(overrides: BTreeMap) -> Self { + Self { overrides } + } + + /// Resolve the input modalities advertised for `model_id`. + /// + /// Precedence is an explicit local override, then recognized upstream metadata, then the + /// conservative text-only fallback. An explicit text-only override therefore wins over + /// upstream image metadata, and a model with neither an override nor metadata stays + /// text-only rather than being guessed from its name. + #[must_use] + pub fn resolve(&self, model_id: &str, upstream: UpstreamCapabilities) -> InputModalities { + if let Some(modalities) = self.overrides.get(model_id) { + return *modalities; + } + if upstream.image { + InputModalities::TextAndImage + } else { + InputModalities::Text + } + } +} + +/// The subset of the gateway's Codex model catalog a launcher needs. +/// +/// Every other catalog field is ignored so that catalog growth cannot break a launcher. +#[derive(Debug, Default, Deserialize)] +pub struct CodexCatalogCapabilities { + /// Catalog entries in the order the gateway advertises them. + #[serde(default)] + pub models: Vec, +} + +/// One catalog entry's model identity and resolved input modalities. +#[derive(Debug, Deserialize)] +pub struct CodexModelCapabilities { + /// Served model ID, matched exactly and case-sensitively. + pub slug: String, + /// The modalities the gateway resolved for this model. + pub input_modalities: InputModalities, +} + +impl CodexCatalogCapabilities { + /// Select `model`, or the first advertised entry when no model was requested. + #[must_use] + pub fn select(&self, model: Option<&str>) -> Option<&CodexModelCapabilities> { + match model { + Some(model) => self.models.iter().find(|entry| entry.slug == model), + None => self.models.first(), + } + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use serde::Deserialize; + + use super::{ + CodexCatalogCapabilities, InputModalities, Modality, ModalityError, ModelCapabilities, UpstreamCapabilities, + }; + + #[derive(Debug, Deserialize)] + struct ModelSection { + input_modalities: InputModalities, + } + + #[test] + fn validation_canonicalizes_accepted_modality_lists() { + assert_eq!(InputModalities::try_new(&[Modality::Text]), Ok(InputModalities::Text)); + assert_eq!( + InputModalities::try_new(&[Modality::Text, Modality::Image]), + Ok(InputModalities::TextAndImage) + ); + assert_eq!( + InputModalities::try_new(&[Modality::Image, Modality::Text]), + Ok(InputModalities::TextAndImage), + "declaration order must not change the resolved modalities" + ); + } + + #[test] + fn validation_rejects_unusable_modality_lists() { + assert_eq!(InputModalities::try_new(&[]), Err(ModalityError::Empty)); + assert_eq!( + InputModalities::try_new(&[Modality::Image]), + Err(ModalityError::MissingText) + ); + assert_eq!( + InputModalities::try_new(&[Modality::Text, Modality::Text]), + Err(ModalityError::Duplicate(Modality::Text)) + ); + assert_eq!( + InputModalities::try_new(&[Modality::Text, Modality::Image, Modality::Image]), + Err(ModalityError::Duplicate(Modality::Image)) + ); + } + + #[test] + fn modality_errors_name_the_wire_values() { + assert!(ModalityError::Empty.to_string().contains("[\"text\", \"image\"]")); + assert_eq!( + ModalityError::Duplicate(Modality::Image).to_string(), + "input_modalities lists \"image\" more than once" + ); + assert!(ModalityError::MissingText.to_string().contains("must include \"text\"")); + } + + #[test] + fn modalities_serialize_in_canonical_wire_order() { + assert_eq!( + serde_json::to_string(&InputModalities::Text).expect("serialize text"), + "[\"text\"]" + ); + assert_eq!( + serde_json::to_string(&InputModalities::TextAndImage).expect("serialize text and image"), + "[\"text\",\"image\"]" + ); + } + + #[test] + fn modalities_round_trip_through_json_and_toml() { + let parsed: InputModalities = serde_json::from_str("[\"image\",\"text\"]").expect("parse JSON modalities"); + assert_eq!(parsed, InputModalities::TextAndImage); + + let section: ModelSection = + toml::from_str("input_modalities = [\"text\", \"image\"]").expect("parse TOML modalities"); + assert_eq!(section.input_modalities, InputModalities::TextAndImage); + } + + #[test] + fn unknown_modality_names_the_accepted_values() { + let error = toml::from_str::("input_modalities = [\"video\"]") + .expect_err("an unknown modality must be rejected"); + let message = error.to_string(); + + assert!(message.contains("video"), "{message}"); + assert!(message.contains("text"), "{message}"); + assert!(message.contains("image"), "{message}"); + } + + #[test] + fn invalid_modality_lists_are_rejected_while_parsing() { + let error = + toml::from_str::("input_modalities = []").expect_err("an empty list must be rejected"); + assert!(error.to_string().contains("at least one modality"), "{error}"); + + let error = toml::from_str::("input_modalities = [\"image\"]") + .expect_err("an image-only list must be rejected"); + assert!(error.to_string().contains("must include \"text\""), "{error}"); + + let error = toml::from_str::("input_modalities = [\"text\", \"text\"]") + .expect_err("a duplicated modality must be rejected"); + assert!(error.to_string().contains("more than once"), "{error}"); + } + + #[test] + fn modalities_report_their_members() { + assert_eq!(InputModalities::Text.as_slice(), [Modality::Text]); + assert_eq!( + InputModalities::TextAndImage.as_slice(), + [Modality::Text, Modality::Image] + ); + assert!(InputModalities::Text.contains(Modality::Text)); + assert!(!InputModalities::Text.contains(Modality::Image)); + assert!(InputModalities::TextAndImage.contains(Modality::Image)); + assert!(!InputModalities::Text.supports_image()); + assert!(InputModalities::TextAndImage.supports_image()); + } + + #[test] + fn upstream_capabilities_recognize_only_known_strings() { + assert_eq!( + UpstreamCapabilities::from_advertised(&["image", "reasoning"]), + UpstreamCapabilities { + image: true, + reasoning: true + } + ); + assert_eq!( + UpstreamCapabilities::from_advertised(&["vision", "multimodal", "IMAGE"]), + UpstreamCapabilities::default(), + "capability support must never be guessed from unrecognized strings" + ); + assert_eq!( + UpstreamCapabilities::from_advertised::(&[]), + UpstreamCapabilities::default() + ); + } + + #[test] + fn resolution_prefers_configuration_over_upstream_metadata() { + let capabilities = ModelCapabilities::new(BTreeMap::from([ + ("vision-model".to_owned(), InputModalities::TextAndImage), + ("pinned-text-model".to_owned(), InputModalities::Text), + ])); + let advertises_image = UpstreamCapabilities { + image: true, + reasoning: false, + }; + let advertises_nothing = UpstreamCapabilities::default(); + + assert_eq!( + capabilities.resolve("vision-model", advertises_nothing), + InputModalities::TextAndImage, + "a configured vision model must advertise images without upstream metadata" + ); + assert_eq!( + capabilities.resolve("pinned-text-model", advertises_image), + InputModalities::Text, + "an explicit text-only override must win over upstream image metadata" + ); + assert_eq!( + capabilities.resolve("unconfigured-model", advertises_image), + InputModalities::TextAndImage, + "recognized upstream metadata must be used when no override exists" + ); + assert_eq!( + capabilities.resolve("unconfigured-model", advertises_nothing), + InputModalities::Text, + "an unknown model without metadata must stay text-only" + ); + assert_eq!( + ModelCapabilities::default().resolve("vision-model", advertises_nothing), + InputModalities::Text + ); + } + + #[test] + fn catalog_selection_matches_slugs_exactly() { + let catalog: CodexCatalogCapabilities = serde_json::from_str( + r#"{"models":[ + {"slug":"first-model","input_modalities":["text"],"display_name":"ignored"}, + {"slug":"vision-model","input_modalities":["text","image"]} + ]}"#, + ) + .expect("parse catalog"); + + assert_eq!(catalog.select(None).expect("first entry").slug, "first-model"); + let selected = catalog.select(Some("vision-model")).expect("selected entry"); + assert_eq!(selected.input_modalities, InputModalities::TextAndImage); + assert!( + catalog.select(Some("Vision-Model")).is_none(), + "slugs are case-sensitive" + ); + assert!(catalog.select(Some("missing-model")).is_none()); + } + + #[test] + fn catalog_without_models_selects_nothing() { + let catalog: CodexCatalogCapabilities = serde_json::from_str("{}").expect("parse empty catalog"); + + assert!(catalog.models.is_empty()); + assert!(catalog.select(None).is_none()); + assert!(catalog.select(Some("any-model")).is_none()); + } +} diff --git a/crates/agentic-server/src/server.rs b/crates/agentic-server/src/server.rs index dab108ff..d7cd2af9 100644 --- a/crates/agentic-server/src/server.rs +++ b/crates/agentic-server/src/server.rs @@ -11,6 +11,7 @@ use agentic_core::proxy::ProxyState; use agentic_core::readiness::{llm_readiness_client, wait_llm_ready}; use agentic_server::app::{AppState, ReadinessTracker, ServerConfig, WebSocketTracker, build_router_with_auth}; use agentic_server::auth::{OidcAuthError, OidcAuthenticator, OidcConfig}; +use agentic_server::model_capabilities::ModelCapabilities; use tokio::net::TcpListener; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; @@ -22,6 +23,7 @@ const GATEWAY_DRAIN_TIMEOUT: Duration = Duration::from_secs(8); /// These are deliberately separate from [`Config`], which carries inference, /// storage, and tool concerns that core owns. pub struct GatewayOptions<'a> { + pub model_capabilities: ModelCapabilities, pub host: &'a str, pub port: u16, /// Ceiling on serialized inbound request bytes for HTTP bodies and @@ -50,6 +52,7 @@ async fn build_state( config: &Config, shutdown_token: CancellationToken, max_request_body_size: NonZeroUsize, + model_capabilities: ModelCapabilities, ) -> Result { let proxy_state = ProxyState::new(config.clone())?; let exec_ctx = Arc::new(ExecutionContext::from_config(config).await?); @@ -65,6 +68,7 @@ async fn build_state( skip_llm_ready_check: config.skip_llm_ready_check, openai_api_key: config.openai_api_key.clone(), max_request_body_size, + model_capabilities: Arc::new(model_capabilities), }) } @@ -169,13 +173,20 @@ pub async fn run(config: Config, gateway: GatewayOptions<'_>) -> Result<(), Serv port, max_request_body_size, oidc, + model_capabilities, } = gateway; let authenticator = match oidc { Some(oidc) => Some(OidcAuthenticator::discover(oidc).await?), None => None, }; wait_until_llm_ready(&config).await?; - let state = build_state(&config, CancellationToken::new(), max_request_body_size).await?; + let state = build_state( + &config, + CancellationToken::new(), + max_request_body_size, + model_capabilities, + ) + .await?; serve_gateway_until_signal(state, host, port, authenticator).await } @@ -195,6 +206,7 @@ pub async fn run_with_llm( port, max_request_body_size, oidc, + model_capabilities, } = gateway; let authenticator = match oidc { Some(oidc) => Some(OidcAuthenticator::discover(oidc).await?), @@ -226,7 +238,7 @@ pub async fn run_with_llm( } state = async { wait_until_llm_ready(&config).await?; - build_state(&config, shutdown_token.clone(), max_request_body_size).await + build_state(&config, shutdown_token.clone(), max_request_body_size, model_capabilities).await } => state?, }; diff --git a/crates/agentic-server/tests/common/mod.rs b/crates/agentic-server/tests/common/mod.rs index d7ec890c..26ca8ce6 100644 --- a/crates/agentic-server/tests/common/mod.rs +++ b/crates/agentic-server/tests/common/mod.rs @@ -54,6 +54,7 @@ pub fn test_state_with_max_request_body_size(config: &Config, max_request_body_s llm_api_base: config.llm_api_base.clone(), skip_llm_ready_check: config.skip_llm_ready_check, openai_api_key: config.openai_api_key.clone(), + model_capabilities: std::sync::Arc::default(), max_request_body_size, } } diff --git a/crates/agentic-server/tests/fixtures/typed-config.toml b/crates/agentic-server/tests/fixtures/typed-config.toml index 34b34e7d..2cf98b4f 100644 --- a/crates/agentic-server/tests/fixtures/typed-config.toml +++ b/crates/agentic-server/tests/fixtures/typed-config.toml @@ -12,3 +12,6 @@ allowed_hosts = ["mcp.example.com"] url = "https://mcp.example.com/mcp" allowed_tools = ["say_hello", "sum"] require_approval = "never" + +[models."Qwen/Qwen3-VL-8B-Instruct"] +input_modalities = ["text", "image"] diff --git a/crates/agentic-server/tests/models_test.rs b/crates/agentic-server/tests/models_test.rs new file mode 100644 index 00000000..0656da02 --- /dev/null +++ b/crates/agentic-server/tests/models_test.rs @@ -0,0 +1,240 @@ +// The shared helpers serve every integration test; this one needs a models-specific upstream. +#[allow(dead_code)] +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use agentic_server::app::AppState; +use agentic_server::model_capabilities::{CodexCatalogCapabilities, InputModalities, ModelCapabilities}; +use axum::Router; +use axum::response::IntoResponse; +use axum::routing::get; +use common::{spawn_gateway, test_config, test_state}; +use http::StatusCode; +use serde_json::Value; +use tokio::net::TcpListener; + +/// Deliberately irregular whitespace and an unknown field, so a pass-through response can be +/// compared byte for byte against what the upstream actually sent. +const UPSTREAM_MODELS: &str = r#"{"object":"list", "data":[ + {"id":"vision-model","max_model_len":32768,"owned_by":"mock-vllm"}, + {"id":"upstream-image-model","capabilities":["image"]}, + {"id":"pinned-text-model","capabilities":["image"]}, + {"id":"plain-model"} +]}"#; + +async fn spawn_upstream_models(body: &'static str, status: StatusCode) -> (String, tokio::task::JoinHandle<()>) { + let app = Router::new().route( + "/v1/models", + get(move || async move { (status, [(http::header::CONTENT_TYPE, "application/json")], body).into_response() }), + ); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let upstream = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + (format!("http://{addr}"), upstream) +} + +fn state_with_overrides(llm_url: &str, overrides: &[(&str, InputModalities)]) -> AppState { + let config = test_config(llm_url); + let overrides = overrides + .iter() + .map(|(model_id, modalities)| ((*model_id).to_owned(), *modalities)) + .collect::>(); + AppState { + model_capabilities: Arc::new(ModelCapabilities::new(overrides)), + ..test_state(&config) + } +} + +async fn spawn_configured_gateway( + body: &'static str, + status: StatusCode, + overrides: &[(&str, InputModalities)], +) -> (String, tokio::task::JoinHandle<()>, tokio::task::JoinHandle<()>) { + let (upstream_url, upstream) = spawn_upstream_models(body, status).await; + let (gateway_url, gateway) = spawn_gateway(state_with_overrides(&upstream_url, overrides)).await; + (gateway_url, upstream, gateway) +} + +fn modalities(catalog: &Value, slug: &str) -> Value { + catalog["models"] + .as_array() + .expect("models array") + .iter() + .find(|model| model["slug"] == slug) + .unwrap_or_else(|| panic!("catalog must contain {slug}"))["input_modalities"] + .clone() +} + +#[tokio::test] +async fn ordinary_model_listing_is_passed_through_unchanged() { + let (gateway_url, _upstream, _gateway) = spawn_configured_gateway( + UPSTREAM_MODELS, + StatusCode::OK, + &[("vision-model", InputModalities::TextAndImage)], + ) + .await; + + let response = reqwest::get(format!("{gateway_url}/v1/models")).await.unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()[http::header::CONTENT_TYPE], "application/json"); + assert_eq!( + response.text().await.unwrap(), + UPSTREAM_MODELS, + "a request without client_version must not be transformed" + ); +} + +#[tokio::test] +async fn codex_catalog_resolves_capabilities_by_precedence() { + let (gateway_url, _upstream, _gateway) = spawn_configured_gateway( + UPSTREAM_MODELS, + StatusCode::OK, + &[ + ("vision-model", InputModalities::TextAndImage), + ("pinned-text-model", InputModalities::Text), + ], + ) + .await; + + let catalog: Value = reqwest::get(format!("{gateway_url}/v1/models?client_version=1.2.3")) + .await + .unwrap() + .json() + .await + .unwrap(); + + assert_eq!( + modalities(&catalog, "vision-model"), + serde_json::json!(["text", "image"]), + "a configured vision model advertises images without upstream metadata" + ); + assert_eq!( + modalities(&catalog, "upstream-image-model"), + serde_json::json!(["text", "image"]) + ); + assert_eq!( + modalities(&catalog, "pinned-text-model"), + serde_json::json!(["text"]), + "an explicit text-only override wins over upstream image metadata" + ); + assert_eq!(modalities(&catalog, "plain-model"), serde_json::json!(["text"])); + assert_eq!( + catalog["models"][0]["supports_image_detail_original"], + serde_json::json!(false) + ); +} + +#[tokio::test] +async fn codex_catalog_is_text_only_without_configuration() { + let (gateway_url, _upstream, _gateway) = spawn_configured_gateway(UPSTREAM_MODELS, StatusCode::OK, &[]).await; + + let catalog: Value = reqwest::get(format!("{gateway_url}/v1/models?client_version=1.2.3")) + .await + .unwrap() + .json() + .await + .unwrap(); + + assert_eq!(modalities(&catalog, "vision-model"), serde_json::json!(["text"])); + assert_eq!(modalities(&catalog, "plain-model"), serde_json::json!(["text"])); +} + +#[tokio::test] +async fn upstream_failures_are_forwarded_without_transformation() { + let (gateway_url, _upstream, _gateway) = + spawn_configured_gateway(r#"{"error":"upstream exploded"}"#, StatusCode::SERVICE_UNAVAILABLE, &[]).await; + + let response = reqwest::get(format!("{gateway_url}/v1/models?client_version=1.2.3")) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(response.text().await.unwrap(), r#"{"error":"upstream exploded"}"#); +} + +#[tokio::test] +async fn undecodable_upstream_payload_is_reported_as_a_bad_gateway() { + let (gateway_url, _upstream, _gateway) = + spawn_configured_gateway(r#"{"data":"not-a-list"}"#, StatusCode::OK, &[]).await; + + let response = reqwest::get(format!("{gateway_url}/v1/models?client_version=1.2.3")) + .await + .unwrap(); + + assert_eq!( + response.status(), + StatusCode::BAD_GATEWAY, + "an undecodable payload must not be served as an empty catalog" + ); + let body: Value = response.json().await.unwrap(); + assert_eq!(body["error"]["code"], "upstream_unavailable"); + assert_eq!(body["error"]["message"], "invalid model list from /v1/models"); +} + +#[tokio::test] +async fn upstream_without_models_yields_an_empty_catalog() { + let (gateway_url, _upstream, _gateway) = spawn_configured_gateway("{}", StatusCode::OK, &[]).await; + + let catalog: Value = reqwest::get(format!("{gateway_url}/v1/models?client_version=1.2.3")) + .await + .unwrap() + .json() + .await + .unwrap(); + + assert_eq!(catalog["models"], serde_json::json!([])); +} + +/// The launcher reads the served catalog through [`CodexCatalogCapabilities`], so the catalog +/// written into an isolated Codex home can only match the HTTP catalog while that view keeps +/// parsing what this handler serves. +#[tokio::test] +async fn launcher_capability_view_parses_the_served_catalog() { + let (gateway_url, _upstream, _gateway) = spawn_configured_gateway( + UPSTREAM_MODELS, + StatusCode::OK, + &[ + ("vision-model", InputModalities::TextAndImage), + ("pinned-text-model", InputModalities::Text), + ], + ) + .await; + + let catalog: CodexCatalogCapabilities = reqwest::get(format!("{gateway_url}/v1/models?client_version=1.2.3")) + .await + .unwrap() + .json() + .await + .unwrap(); + + assert_eq!( + catalog.select(None).expect("first entry").slug, + "vision-model", + "the launcher selects the first advertised model when none is requested" + ); + assert_eq!( + catalog + .select(Some("vision-model")) + .expect("configured vision model") + .input_modalities, + InputModalities::TextAndImage + ); + assert_eq!( + catalog + .select(Some("upstream-image-model")) + .expect("upstream image model") + .input_modalities, + InputModalities::TextAndImage + ); + assert_eq!( + catalog + .select(Some("pinned-text-model")) + .expect("pinned text model") + .input_modalities, + InputModalities::Text + ); + assert!(catalog.select(Some("absent-model")).is_none()); +} diff --git a/crates/agentic-server/tests/responses_test.rs b/crates/agentic-server/tests/responses_test.rs index d42f0176..1f1dfdc1 100644 --- a/crates/agentic-server/tests/responses_test.rs +++ b/crates/agentic-server/tests/responses_test.rs @@ -216,6 +216,7 @@ async fn storage_backed_state(llm_url: &str) -> StorageBackedState { llm_api_base: config.llm_api_base, skip_llm_ready_check: config.skip_llm_ready_check, openai_api_key: config.openai_api_key, + model_capabilities: std::sync::Arc::default(), max_request_body_size: DEFAULT_MAX_REQUEST_BODY_SIZE, }; StorageBackedState { state, pool, _db: db } diff --git a/crates/agentic-server/tests/responses_websocket_test.rs b/crates/agentic-server/tests/responses_websocket_test.rs index 81a72c2c..a7215555 100644 --- a/crates/agentic-server/tests/responses_websocket_test.rs +++ b/crates/agentic-server/tests/responses_websocket_test.rs @@ -323,6 +323,7 @@ fn persistence_disabled_state(llm_url: &str) -> AppState { llm_api_base: config.llm_api_base, skip_llm_ready_check: config.skip_llm_ready_check, openai_api_key: config.openai_api_key, + model_capabilities: std::sync::Arc::default(), max_request_body_size: DEFAULT_MAX_REQUEST_BODY_SIZE, } } @@ -359,6 +360,7 @@ async fn storage_backed_state_with_web_search(llm_url: &str, web_search_base_url llm_api_base: config.llm_api_base, skip_llm_ready_check: config.skip_llm_ready_check, openai_api_key: config.openai_api_key, + model_capabilities: std::sync::Arc::default(), max_request_body_size: DEFAULT_MAX_REQUEST_BODY_SIZE, }; StorageBackedState { state, pool, _db: db } diff --git a/docs/design/codex-integration.md b/docs/design/codex-integration.md index 96ade3bf..e9f09327 100644 --- a/docs/design/codex-integration.md +++ b/docs/design/codex-integration.md @@ -309,6 +309,67 @@ diagnostic, and rotate it if it was reusable. --- +## Image Capability Resolution + +Codex decides whether a request may carry image content by reading its **local** model catalog. When the entry for the +selected model does not list `image` in `input_modalities`, Codex strips image content client-side and sends a +placeholder instead. A vision-capable upstream is therefore not enough on its own: the catalog Codex reads has to say +the model accepts images. + +Two catalogs exist and they must agree: + +| Catalog | Produced by | Consumed by | +|---|---|---| +| `GET /v1/models?client_version=` | `handler/http/models.rs` | Codex refreshing its model list, and both `agentic` launchers | +| `$CODEX_HOME/model_catalog.json` | `agentic_harness::prepare_codex_home` (and `scripts/agentic-codex.sh`) | Codex reading an isolated session home | + +### Resolution order + +The gateway resolves `input_modalities` for every served model in this order: + +1. An explicit `[models.""] input_modalities` override in `~/.agentic-api/config.toml`. +2. Recognized upstream metadata: `capabilities: ["image"]` on the upstream `/v1/models` entry. +3. A conservative text-only fallback. + +An explicit `["text"]` override wins over upstream image metadata, which is how a vision model gets pinned to text. +Capabilities are never inferred from a model name — an unrecognized capability string such as `vision` or `multimodal` +resolves to text-only. `supports_image_detail_original` stays `false`: the gateway does not relay image detail hints. + +### How the launchers stay consistent + +`agentic run codex` and `agentic harness codex` both fetch `GET {gateway}/v1/models?client_version=` before +writing the isolated home, and take the selected model **and** its modalities from that one response. The client +version is read from the Codex binary itself (`codex --version`, honoring `AGENTIC_CODEX_BIN`); set +`AGENTIC_CODEX_CLIENT_VERSION` to skip the probe where it cannot run. Only the modalities are copied — the isolated +catalog keeps its launcher-specific settings (`shell_type: "local"`, an omitted `apply_patch_tool_type`, and a +token-based truncation policy), which intentionally differ from the HTTP catalog. + +`scripts/agentic-codex.sh` copies the gateway catalog verbatim, so it inherits the resolved modalities with no change. + +### Failure behavior + +A launcher never writes a catalog it could not verify. If the gateway cannot be reached, rejects the request, returns +an undecodable catalog, or does not list the selected model, the launch fails with an actionable error instead of +writing text-only metadata: + +- Transport errors, `5xx`, `408`, `425`, `429`, and an empty catalog are retried until the readiness budget expires + (`--llm-ready-timeout-s`/`--llm-ready-interval-s`; 30s/250ms when attaching to a running gateway). +- `401`/`403` fail immediately with a hint to pass `--api-key`. **A gateway behind OIDC now requires a credential for + `agentic harness codex`**, because `/v1/models` is a protected route. +- A served, non-empty catalog that does not list the selected model is retried for 10 seconds and then fails, naming + the models the gateway does serve. A catalog listing other models proves the upstream is warm, so a missing model is + treated as a configuration error rather than a cold start. +- Catalog responses larger than 1 MiB are rejected. + +### Regenerating existing session homes + +`agentic run` and `agentic harness` create a fresh session home per invocation, so they pick up resolved modalities +automatically. A persistent home does not: delete and regenerate any `model_catalog.json` written before this change +(for example a directory pinned with `AGENTIC_CODEX_HOME`, or a hand-written `-c model_catalog_json=...` file), or +Codex will keep reading the stale text-only entry. + +--- + ## Out Of Scope - Raw proxy namespace flatten/restore. diff --git a/docs/guides/harness-cli-testing.md b/docs/guides/harness-cli-testing.md index 79aa1531..7a5700d3 100644 --- a/docs/guides/harness-cli-testing.md +++ b/docs/guides/harness-cli-testing.md @@ -31,7 +31,10 @@ while verifying [issue #190](https://github.com/vllm-project/agentic-api/issues/ CI pins Claude Code 2.1.245 and Codex 0.149.1 and runs both real CLIs through the attach commands against recorded Qwen/vLLM streams. The Claude job verifies a gateway-owned web-search round trip; the Codex job verifies a completed -Responses answer. Run the same checks locally with `bash scripts/claude-code-smoke.sh` and +Responses answer and a real PNG attachment through both `agentic harness codex` and `agentic run codex`. +The image checks use the committed Qwen2.5-VL response recording, compare the upstream image bytes with the attached +PNG, and verify that the same attachment is absent with an explicit text-only catalog. They exercise the actual pinned +CLI and gateway, without contacting a live model or claiming fresh vision inference. Run the same checks locally with `bash scripts/claude-code-smoke.sh` and `bash scripts/codex-smoke.sh` after building both binaries with `cargo build -p agentic-server --bins`. ## CLI behavior worth knowing @@ -43,6 +46,7 @@ Responses answer. Run the same checks locally with `bash scripts/claude-code-smo | Claude uses isolated settings and state | Per-run settings map Claude Code's canonical `claude-sonnet-4-5-20250929` identifier to the exact served model ID, while every default and small/fast model tier is pinned to that served model. Session history is isolated from the user's normal Claude home under `$AGENTIC_API_HOME/harnesses/claude` (default `~/.agentic-api/harnesses/claude`) so `--resume` and `--continue` work across invocations. Inherited Vertex, Bedrock, and Foundry routing switches are removed. | | Claude effort is pinned to `medium` | Claude Code defaults to `high`, which Qwen's vLLM chat template rejects (`ValueError`). The CLI always passes `--effort medium` and sets `CLAUDE_CODE_EFFORT_LEVEL=medium` (the env var wins inside Claude Code). Override both with `AGENTIC_CLAUDE_EFFORT=low|medium|xhigh`. | | Claude resource limits are pinned | The generated environment sets a 32,768-token context, 2,048 output tokens, and disables extended thinking. These conservative defaults fit the tested Qwen deployment. | +| Codex reads its model and image support from the gateway | Before writing an isolated Codex home, both `run codex` and `harness codex` fetch `GET {gateway}/v1/models?client_version=` and take the model and its `input_modalities` from that one response, so the isolated catalog always matches what the gateway serves. The client version comes from `codex --version` (honoring `AGENTIC_CODEX_BIN`); set `AGENTIC_CODEX_CLIENT_VERSION` to skip that probe. A gateway that is unreachable, rejects the request, or does not serve the selected model fails the launch with an actionable error instead of writing text-only metadata. Against an OIDC-protected gateway, pass `--api-key`. Configure image support with `[models.""] input_modalities = ["text", "image"]`; see [Codex integration](../design/codex-integration.md). | | `--yolo` | Adds `--dangerously-skip-permissions` (Claude) or `--dangerously-bypass-approvals-and-sandbox` (Codex). Use only in an externally isolated environment. | | `--skip-llm-ready-check` | Skips the upstream `/health` probe. Avoid it while testing: the probe is what surfaces an unreachable upstream before the harness starts. | | Arguments after `--` | Forwarded to the harness (`-p`, `--resume`, `exec`, ...). Claude's `--model`, `--settings`, `--setting-sources`, and `--bare` are rejected because they would bypass the generated model and provider isolation. Generated settings are temporary, but Claude session history persists in the isolated Agentic API home. | diff --git a/scripts/claude_code_replay_server.py b/scripts/claude_code_replay_server.py index 2510f511..215062f0 100755 --- a/scripts/claude_code_replay_server.py +++ b/scripts/claude_code_replay_server.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse +import base64 import json import threading from dataclasses import dataclass @@ -127,7 +128,9 @@ def validate_capture(records: list[dict[str, Any]], expected_model: str | None = assert searches[0].get("query"), "expected a non-empty search query" -def validate_responses_capture(records: list[dict[str, Any]], expected_model: str) -> None: +def validate_responses_capture( + records: list[dict[str, Any]], expected_model: str, *, expected_images: list[Path] | None = None +) -> None: responses = [record["body"] for record in records if record["kind"] == "responses"] transports = [record["body"] for record in records if record["kind"] == "responses_transport"] assert len(responses) == 1, f"expected one Responses request, got {len(responses)}" @@ -139,12 +142,32 @@ def validate_responses_capture(records: list[dict[str, Any]], expected_model: st assert request.get("stream") is True, "expected Codex to request a streaming response" assert request.get("input"), "expected a non-empty Responses input" + if expected_images is not None: + items = request["input"] + images = [ + part for item in (items if isinstance(items, list) else []) + if item.get("type", "message") == "message" and item.get("role") == "user" + for part in (item["content"] if isinstance(item.get("content"), list) else []) + if part.get("type") == "input_image" + ] + assert len(images) == len(expected_images), ( + f"expected {len(expected_images)} user image attachments, got {len(images)}" + ) + for image, expected in zip(images, expected_images): + url = image.get("image_url", "") + prefix = "data:image/png;base64," + assert url.startswith(prefix), "expected an inline PNG image attachment" + assert base64.b64decode(url[len(prefix):], validate=True) == expected.read_bytes(), ( + f"image attachment bytes differ from {expected}" + ) + @dataclass class ReplayState: turns: list[ReplayTurn] capture_path: Path next_turn: int = 0 + model: str | None = None def __post_init__(self) -> None: self.lock = threading.Lock() @@ -192,6 +215,17 @@ def do_GET(self) -> None: if parsed.path == "/health": self._send_bytes(200, "text/plain", b"") return + if parsed.path == "/v1/models" and state.model is not None: + # Synthetic discovery metadata, not a captured inference turn. + # The text-only replay must not advertise image support. + self._send_json( + 200, + { + "object": "list", + "data": [{"id": state.model, "object": "model", "capabilities": []}], + }, + ) + return if parsed.path == "/v1/search": query = {key: values[-1] for key, values in parse_qs(parsed.query).items()} self._send_search_response(query) @@ -279,12 +313,19 @@ def parse_args() -> argparse.Namespace: serve.add_argument("--cassette", required=True, type=Path) serve.add_argument("--port", required=True, type=int) serve.add_argument("--capture", required=True, type=Path) + serve.add_argument("--model", help="Model ID to advertise in a synthetic text-only catalog") assert_capture = subparsers.add_parser("assert-capture") assert_capture.add_argument("--capture", required=True, type=Path) assert_capture.add_argument("--api", choices=("messages", "responses"), default="messages") assert_capture.add_argument("--model", required=True) - return parser.parse_args() + images = assert_capture.add_mutually_exclusive_group() + images.add_argument("--expect-image", type=Path, action="append", help="Require these exact PNG attachments in order") + images.add_argument("--expect-no-images", action="store_true", help="Require no user image attachments") + args = parser.parse_args() + if args.command == "assert-capture" and args.api != "responses" and (args.expect_image or args.expect_no_images): + parser.error("image assertions require --api responses") + return args def main() -> None: @@ -292,7 +333,8 @@ def main() -> None: if args.command == "assert-capture": records = load_capture(args.capture) if args.api == "responses": - validate_responses_capture(records, args.model) + expected_images = [] if args.expect_no_images else args.expect_image + validate_responses_capture(records, args.model, expected_images=expected_images) responses = sum(record["kind"] == "responses" for record in records) transports = sum(record["kind"] == "responses_transport" for record in records) print(f"capture valid: responses={responses} transports={transports}") @@ -306,7 +348,7 @@ def main() -> None: args.capture.parent.mkdir(parents=True, exist_ok=True) args.capture.write_text("") - state = ReplayState(load_turns(args.cassette), args.capture) + state = ReplayState(load_turns(args.cassette), args.capture, model=args.model) server = ThreadingHTTPServer(("127.0.0.1", args.port), make_handler(state)) server.serve_forever() diff --git a/scripts/codex-smoke.sh b/scripts/codex-smoke.sh index 37ebe083..17860eda 100755 --- a/scripts/codex-smoke.sh +++ b/scripts/codex-smoke.sh @@ -85,6 +85,7 @@ wait_until_ready() { "$PYTHON_BIN" scripts/claude_code_replay_server.py serve \ --cassette "$CASSETTE" \ + --model "$MODEL" \ --capture "$capture_path" \ --port "$REPLAY_PORT" \ >"$replay_log" 2>&1 & @@ -128,3 +129,7 @@ PY --api responses \ --model "$MODEL" \ --capture "$capture_path" + +# Exercise image capability propagation with the same pinned CLI, including a +# text-only negative control. The image response replays the committed vision cassette. +"$PYTHON_BIN" scripts/codex_image_smoke.py diff --git a/scripts/codex_image_smoke.py b/scripts/codex_image_smoke.py new file mode 100755 index 00000000..f3d86d19 --- /dev/null +++ b/scripts/codex_image_smoke.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Exercise actual Codex image attachments through both Agentic API launchers. + +Responses are replayed from the committed gateway/vLLM vision recording. These +checks verify transport and catalog propagation, not fresh model understanding. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import socket +import signal +import subprocess +import tempfile +import threading +import time +import urllib.error +import urllib.request +from http.server import ThreadingHTTPServer + +import claude_code_replay_server as replay + + +ROOT = Path(__file__).resolve().parent.parent +FIXTURES = ROOT / "crates/agentic-server-core/tests/cassettes/images" +IMAGE = FIXTURES / "inputs/red-blue-64.png" +MODEL = "Qwen/Qwen2.5-VL-3B-Instruct" +CASSETTE = FIXTURES / "responses/image-single-image-gateway-Qwen-Qwen2.5-VL-3B-Instruct-streaming.yaml" +PROMPT = "Reply with exactly two words: the color on the left half of this image, then the color on the right half." + + +def choose_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +def wait_ready(url: str, process: subprocess.Popen) -> None: + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + assert process.poll() is None, "gateway exited before becoming ready" + try: + with opener.open(url, timeout=1) as response: + if response.status == 200: + return + except (urllib.error.URLError, TimeoutError): + pass + time.sleep(0.1) + raise AssertionError("gateway did not become ready within 30 seconds") + + +def stop(process: subprocess.Popen | None) -> None: + if process is None or process.poll() is not None: + return + process.terminate() + try: + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + + +def run_launcher(command: list[str], environment: dict[str, str], timeout: float = 60) -> subprocess.CompletedProcess: + # A separate process group lets a timeout stop the integrated gateway and + # Codex as well as their launcher; SIGKILL alone bypasses Rust destructors. + process = subprocess.Popen( + command, env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True, start_new_session=os.name == "posix", + ) + try: + stdout, stderr = process.communicate(timeout=timeout) + return subprocess.CompletedProcess(command, process.returncode, stdout, stderr) + except BaseException: + if os.name == "posix": + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.communicate(timeout=5) + else: + stop(process) + raise + + +def run_case(mode: str, supports_image: bool) -> None: + label = f"{mode}-{'image' if supports_image else 'text-control'}" + with tempfile.TemporaryDirectory(prefix=f"agentic-codex-{label}-") as temporary: + root = Path(temporary) + capture = root / "capture.jsonl" + capture.write_text("") + state = replay.ReplayState(replay.load_turns(CASSETTE), capture, model=MODEL) + server = ThreadingHTTPServer(("127.0.0.1", 0), replay.make_handler(state)) + thread = threading.Thread(target=server.serve_forever) + thread.start() + gateway = None + try: + gateway_port = choose_port() + upstream = f"http://127.0.0.1:{server.server_port}" + gateway_url = f"http://127.0.0.1:{gateway_port}" + modalities = ["text", "image"] if supports_image else ["text"] + (root / "config.toml").write_text( + f"[models.{json.dumps(MODEL)}]\ninput_modalities = {json.dumps(modalities)}\n" + ) + environment = { + **os.environ, + "AGENTIC_API_HOME": str(root), + "AGENTIC_CODEX_BIN": os.environ.get("CODEX_BIN", "codex"), + "DATABASE_URL": f"sqlite://{root / 'agentic.db'}", + "RUST_LOG": "warn", + "OPENAI_API_KEY": "must-not-be-forwarded", + } + # The launcher must probe the actual pinned client used by this test. + environment.pop("AGENTIC_CODEX_CLIENT_VERSION", None) + agentic = str(Path(os.environ.get("AGENTIC_BIN", "target/debug/agentic")).resolve()) + with (root / "gateway.log").open("w") as gateway_log: + if mode == "harness": + gateway_binary = str(Path(os.environ.get("AGENTIC_SERVER_BIN", "target/debug/agentic-server")).resolve()) + gateway = subprocess.Popen( + [gateway_binary, "--llm-api-base", upstream, "--gateway-host", "127.0.0.1", + "--gateway-port", str(gateway_port), "--skip-llm-ready-check"], + env=environment, stdout=gateway_log, stderr=subprocess.STDOUT, + ) + wait_ready(gateway_url + "/ready", gateway) + launch = [agentic, "harness", "codex", "--gateway-url", gateway_url, "--model", MODEL, "--quiet"] + else: + launch = [agentic, "run", "codex", "--upstream", upstream, "--model", MODEL, + "--gateway-host", "127.0.0.1", "--gateway-port", str(gateway_port), + "--skip-llm-ready-check", "--quiet"] + result = run_launcher( + [*launch, "--", "exec", "--skip-git-repo-check", "--image", str(IMAGE), "--", PROMPT], + environment, + ) + assert result.returncode == 0, f"{label}: launcher failed:\n{result.stdout}\n{result.stderr}" + assert "Red, Blue" in result.stdout, f"{label}: missing recorded answer: {result.stdout!r}" + replay.validate_responses_capture( + replay.load_capture(capture), MODEL, expected_images=[IMAGE] if supports_image else [], + ) + print(f"Codex {label}: recorded answer and exact upstream image capture verified") + except Exception: + for name in ("gateway.log", "capture.jsonl"): + path = root / name + if path.exists(): + print(f"{label} {name}:\n{path.read_text()}") + raise + finally: + stop(gateway) + server.shutdown() + server.server_close() + thread.join(timeout=5) + assert not thread.is_alive(), "replay server failed to stop" + + +def main() -> None: + for mode, images in [("harness", True), ("run", True), ("harness", False)]: + run_case(mode, images) + + +if __name__ == "__main__": + main() diff --git a/scripts/test_claude_code_replay_server.py b/scripts/test_claude_code_replay_server.py index 913e7466..90124a0a 100644 --- a/scripts/test_claude_code_replay_server.py +++ b/scripts/test_claude_code_replay_server.py @@ -1,15 +1,24 @@ +import base64 +import copy import json +import os +import subprocess +import time import sys import tempfile import threading import unittest +import urllib.error import urllib.request +from contextlib import contextmanager from http.server import ThreadingHTTPServer from pathlib import Path +from unittest.mock import patch sys.path.insert(0, str(Path(__file__).parent)) import claude_code_replay_server as replay +import codex_image_smoke as image_smoke REPOSITORY_ROOT = Path(__file__).resolve().parent.parent @@ -91,6 +100,54 @@ def capture_records(first_request: dict, second_request: dict) -> list[dict]: class ReplayServerTests(unittest.TestCase): + @contextmanager + def catalog_server(self, model): + with tempfile.TemporaryDirectory() as temp_dir: + capture_path = Path(temp_dir) / "capture.jsonl" + capture_path.write_text("") + state = replay.ReplayState(replay.load_turns(RESPONSES_CASSETTE), capture_path, model=model) + server = ThreadingHTTPServer(("127.0.0.1", 0), replay.make_handler(state)) + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_port}", state + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + self.assertFalse(thread.is_alive()) + + def test_models_route_is_text_only_and_does_not_consume_recorded_turn(self) -> None: + model = "custom/model-with-override" + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with self.catalog_server(model) as (url, state): + for _ in range(2): + with opener.open(url + "/v1/models", timeout=5) as response: + self.assertEqual(response.status, 200) + self.assertEqual(response.headers.get_content_type(), "application/json") + self.assertEqual( + json.load(response), + {"object": "list", "data": [{"id": model, "object": "model", "capabilities": []}]}, + ) + self.assertEqual(state.next_turn, 0) + self.assertEqual(replay.load_capture(state.capture_path), []) + self.assertEqual(state.take_turn(), replay.load_turns(RESPONSES_CASSETTE)[0]) + + def test_unconfigured_model_catalog_keeps_not_found_response(self) -> None: + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with self.catalog_server(None) as (url, state): + with self.assertRaises(urllib.error.HTTPError) as failure: + opener.open(url + "/v1/models", timeout=5) + self.assertEqual(failure.exception.code, 404) + failure.exception.close() + self.assertEqual(state.next_turn, 0) + + def test_serve_accepts_optional_model_catalog(self) -> None: + arguments = ["replay", "serve", "--cassette", "fixture.yaml", "--capture", "capture.jsonl", "--port", "0"] + for extra, expected in (([], None), (["--model", QWEN_MODEL], QWEN_MODEL)): + with self.subTest(model=expected), patch.object(sys, "argv", arguments + extra): + self.assertEqual(replay.parse_args().model, expected) + def test_load_turns_reads_recorded_streams(self) -> None: turns = replay.load_turns(CASSETTE) @@ -148,6 +205,67 @@ def test_validate_responses_capture_rejects_wrong_model(self) -> None: with self.assertRaisesRegex(AssertionError, "requested model"): replay.validate_responses_capture(records, QWEN_MODEL) + def image_records(self, images=None): + if images is None: + images = [REPOSITORY_ROOT / "crates/agentic-server-core/tests/cassettes/images/inputs/red-blue-64.png"] + return [ + {"kind": "responses_transport", "body": {"path": "/v1/responses"}}, + {"kind": "responses", "body": { + "model": QWEN_MODEL, + "stream": True, + "input": [{"role": "user", "content": [ + {"type": "input_text", "text": "Describe the attached image."}, + *[{"type": "input_image", "image_url": "data:image/png;base64," + + base64.b64encode(image.read_bytes()).decode()} for image in images], + ]}], + }}, + ], images + + def test_responses_capture_checks_exact_attachment_bytes(self) -> None: + records, images = self.image_records() + replay.validate_responses_capture(records, QWEN_MODEL, expected_images=images) + + def test_responses_capture_rejects_dropped_changed_or_duplicate_images(self) -> None: + records, images = self.image_records() + for replacement in [[], [{"type": "input_image", "image_url": "data:image/png;base64,YWJj"}], + records[1]["body"]["input"][0]["content"][1:] * 2]: + with self.subTest(replacement=replacement), self.assertRaises(AssertionError): + changed = copy.deepcopy(records) + changed[1]["body"]["input"][0]["content"] = replacement + replay.validate_responses_capture(changed, QWEN_MODEL, expected_images=images) + + def test_tool_image_output_cannot_satisfy_user_attachment_check(self) -> None: + records, images = self.image_records() + parts = records[1]["body"]["input"][0]["content"] + records[1]["body"]["input"] = [{"type": "function_call_output", "call_id": "call_1", "output": parts}] + with self.assertRaisesRegex(AssertionError, "image"): + replay.validate_responses_capture(records, QWEN_MODEL, expected_images=images) + + def test_text_only_control_requires_no_image_parts(self) -> None: + records, _ = self.image_records([]) + replay.validate_responses_capture(records, QWEN_MODEL, expected_images=[]) + records, _ = self.image_records() + with self.assertRaisesRegex(AssertionError, "image"): + replay.validate_responses_capture(records, QWEN_MODEL, expected_images=[]) + + @unittest.skipUnless(os.name == "posix", "process groups require POSIX") + def test_image_smoke_timeout_stops_launcher_descendants(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + marker = Path(temporary) / "orphan-wrote-this" + child = "import time,pathlib; time.sleep(1); pathlib.Path(" + repr(str(marker)) + ").touch()" + parent = "import subprocess,sys,time; subprocess.Popen([sys.executable, '-c', " + repr(child) + "]); time.sleep(30)" + with self.assertRaises(subprocess.TimeoutExpired): + image_smoke.run_launcher([sys.executable, "-c", parent], dict(os.environ), timeout=0.3) + time.sleep(1) + self.assertFalse(marker.exists(), "a launcher descendant survived the timeout") + + def test_capture_cli_accepts_image_and_no_image_assertions(self) -> None: + base = ["replay", "assert-capture", "--api", "responses", "--capture", "capture.jsonl", "--model", QWEN_MODEL] + with patch.object(sys, "argv", base + ["--expect-image", "image.png"]): + self.assertEqual(replay.parse_args().expect_image, [Path("image.png")]) + with patch.object(sys, "argv", base + ["--expect-no-images"]): + self.assertTrue(replay.parse_args().expect_no_images) + def test_responses_route_replays_recorded_stream_and_captures_request(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: capture_path = Path(temp_dir) / "capture.jsonl" diff --git a/scripts/tests/agentic-cli-e2e-test.py b/scripts/tests/agentic-cli-e2e-test.py index 4d86ea30..a47715d4 100755 --- a/scripts/tests/agentic-cli-e2e-test.py +++ b/scripts/tests/agentic-cli-e2e-test.py @@ -109,6 +109,11 @@ def _json(self, value: dict) -> None: import sys import urllib.request +if "--version" in sys.argv: + # The Codex launcher asks the harness binary for the client version it reports to the gateway. + print("codex-cli 0.0.0-e2e") + raise SystemExit(0) + def post(url, body): request = urllib.request.Request(url, json.dumps(body).encode(), {"Content-Type": "application/json"}) with urllib.request.urlopen(request) as response: @@ -122,6 +127,11 @@ def post(url, body): if mode == "codex": config = open(os.path.join(os.environ["CODEX_HOME"], "config.toml")).read() base = re.search(r'base_url = "([^"]+)"', config).group(1) + catalog = json.load(open(os.path.join(os.environ["CODEX_HOME"], "model_catalog.json"))) + entry = catalog["models"][0] + assert entry["slug"] == model, entry + # The mock upstream advertises no capabilities, so the gateway resolves text-only. + assert entry["input_modalities"] == ["text"], entry first = post(base + "/responses", {"model": model, "input": "Remember APPLE", "store": True, "stream": False}) second = post(base + "/responses", {"model": model, "input": "What word?", "previous_response_id": first["id"], "store": True, "stream": False}) assert second["id"], second diff --git a/scripts/tests/agentic-launchers-test.sh b/scripts/tests/agentic-launchers-test.sh index 5460e62f..40d56215 100755 --- a/scripts/tests/agentic-launchers-test.sh +++ b/scripts/tests/agentic-launchers-test.sh @@ -21,6 +21,14 @@ assert_file_contains() { grep -F -- "$expected" "$file" >/dev/null || fail "$file does not contain: $expected" } +assert_catalog_modalities() { + local file="$1" + local expected="$2" + local actual + actual="$(jq -c '.models[0].input_modalities' "$file")" + [[ "$actual" == "$expected" ]] || fail "$file advertises $actual, expected $expected" +} + assert_file_excludes() { local file="$1" local unexpected="$2" @@ -105,7 +113,7 @@ cat <<'JSON' "use_responses_lite": false, "tool_mode": null, "multi_agent_version": null, - "input_modalities": ["text"] + "input_modalities": ["text", "image"] }] } JSON @@ -160,6 +168,7 @@ assert_file_contains "$codex_home/config.toml" 'model = "agentic-api"' assert_file_contains "$codex_home/config.toml" 'supports_websockets = true' assert_file_contains "$codex_home/model_catalog.json" '"web_search_tool_type": "text"' assert_file_contains "$codex_home/model_catalog.json" '"shell_type": "shell_command"' +assert_catalog_modalities "$codex_home/model_catalog.json" '["text","image"]' assert_file_contains "$capture_dir/curl.txt" 'http://127.0.0.1:3020/v1/models?client_version=0.148.0' PATH="$fake_bin:$PATH" CAPTURE_DIR="$capture_dir" AGENTIC_YOLO=1 CLAUDE_BIN="$fake_bin/claude" \